From 55dc1180f663f70d60e68ad43e2a8c23c9018c57 Mon Sep 17 00:00:00 2001 From: Milly Date: Mon, 29 Jun 2026 16:27:21 +0800 Subject: [PATCH] [milly] feat(xiaof): OpenAI-compatible adapter for general_qa LLM path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Scott 拍板（2026-06-29）: 主线 = 直连 LLM SDK + 轻 loop，OpenAI 兼容协议 为主路径，DeepSeek/Qwen/本地 vLLM 一份代码切；Anthropic Claude 作 swap target。 A-10「换 backend 改 1 个配置」落地为 base_url + api_key + model 三个环境变量。 What lands: - forge_xiaof_openai.py: openai_compatible_adapter * env-gated (XIAOF_OPENAI_BASE_URL/API_KEY/MODEL/TIMEOUT) * SSE streaming chat.completions, no extra deps (urllib only) * jargon scrubber on LLM output covering M1/M2/stub/adapter/milestone/占位 with cross-chunk look-ahead so the forbidden word can't reassemble between two emitted token events (regression on milk's 6/14 硬线 #1). * upstream / network failure → XiaofRequestError code='upstream_failed'; NEVER 'forbidden' (A-7.5 red line). * chips always empty — LLM path cannot bypass M2's A-7 retrieval ACL fixture (milk's 6/14 硬线 #2). - forge_xiaof.py: new default_adapter routing * intent=general_qa + local A-4 builtin matches → stub (no LLM call, keeps PRD A-4 sub-5s time answer). * intent=general_qa + env present + builtin miss → openai adapter. * intent=thread_search → stub unconditionally (M2 owns retrieval). * No env → falls back to stub neutral copy; MVP not broken. Tests: 15 new cases (jargon scrub incl. cross-chunk, env gating, all three routing branches, upstream failure mapping, contract event order). Full pytest matrix + ruff clean. Gateway path (milk's ① 默认): adapter only sees one base_url, so pointing XIAOF_OPENAI_BASE_URL at milk's gateway is the entire integration — no adapter changes. --- forge_xiaof.py | 46 ++++++- forge_xiaof_openai.py | 259 +++++++++++++++++++++++++++++++++++++ tests/test_xiaof_openai.py | 200 ++++++++++++++++++++++++++++ 3 files changed, 504 insertions(+), 1 deletion(-) create mode 100644 forge_xiaof_openai.py create mode 100644 tests/test_xiaof_openai.py diff --git a/forge_xiaof.py b/forge_xiaof.py index 3dd5f80..2b102c7 100644 --- a/forge_xiaof.py +++ b/forge_xiaof.py @@ -246,7 +246,51 @@ def _chunk_text(text: str, size: int) -> Iterable[str]: # ── adapter registry ──────────────────────────────────────────────── _ADAPTERS: dict[str, Adapter] = {"stub": stub_adapter} -_DEFAULT_ADAPTER_NAME = "stub" +_DEFAULT_ADAPTER_NAME = "default" + + +def default_adapter(payload: Mapping[str, Any]) -> Iterator[XiaofEvent]: + """ + Routing adapter (the one the route looks up by default). + + Decision matrix: + - intent == general_qa + └ local A-4 builtin matches (time/date/tz) → stub (no LLM call) + └ OpenAI-compat env present → openai_compatible_adapter + └ otherwise → stub neutral copy + - intent == thread_search → stub (M2 owns this) + + Keeping the local builtin first preserves PRD A-4 sub-5s latency for + time queries even when an LLM is wired up; the LLM only ever sees + open-ended general questions, never thread retrieval. + """ + # Late import: forge_xiaof_openai depends on this module, so we + # cannot import it at module load time. + try: + from forge_xiaof_openai import openai_compatible_adapter, openai_enabled + except Exception: # noqa: BLE001 - defensive + openai_compatible_adapter = None # type: ignore[assignment] + openai_enabled = lambda: False # noqa: E731 + + query = str(payload.get("query") or "") + intent = classify_intent(query) + client = payload.get("client") if isinstance(payload, Mapping) else None + if not isinstance(client, Mapping): + client = {} + + if ( + intent == "general_qa" + and answer_general_qa(query, client) is None + and openai_enabled() + and openai_compatible_adapter is not None + ): + yield from openai_compatible_adapter(payload) + return + + yield from stub_adapter(payload) + + +_ADAPTERS["default"] = default_adapter def register_adapter(name: str, adapter: Adapter) -> None: diff --git a/forge_xiaof_openai.py b/forge_xiaof_openai.py new file mode 100644 index 0000000..54ee66b --- /dev/null +++ b/forge_xiaof_openai.py @@ -0,0 +1,259 @@ +""" +forge_xiaof_openai — OpenAI-compatible adapter for the 小F backend. + +Implements the agreed main path (Scott 2026-06-29 拍板): + + POST /api/xiaof/ask → general_qa → OpenAI-compatible chat.completions + (DeepSeek / Qwen / OpenAI / vLLM / + eventually the milk-owned gateway) + +A-10 "换 backend 改 1 个配置" is delivered as three env vars: + + XIAOF_OPENAI_BASE_URL e.g. https://api.deepseek.com/v1 (or gateway base) + XIAOF_OPENAI_API_KEY bearer token; if unset the adapter is considered + disabled and the route falls back to stub. + XIAOF_OPENAI_MODEL defaults to "deepseek-chat". + XIAOF_OPENAI_TIMEOUT seconds, default 30. + +Routing (in forge_xiaof.default_adapter): + + - intent=general_qa AND local A-4 builtin doesn't match + AND XIAOF_OPENAI_* env present + → this adapter (real LLM) + - everything else → stub_adapter (M2 owns thread_search later) + +Red lines kept identical to PRD v0.2 / API contract v0.1: + - Event order: meta → token* → chips → done (chips always empty here; + thread_search never reaches this adapter, A-7 12-case + fixture is unaffected — LLM cannot pull threads). + - Error codes: upstream/network failure → XiaofRequestError(code= + "upstream_failed"); the route maps it to the contract's + 4-error set, never `forbidden`. + - User-visible jargon scrubber: every emitted token chunk passes through + scrub_jargon() so the LLM cannot + accidentally leak "M1 / M2 / stub / + adapter / milestone / 占位" into user + text. A streaming look-ahead buffer + prevents partial-word splits across + token boundaries. +""" + +from __future__ import annotations + +import json +import os +import re +import time +import urllib.request +from collections.abc import Iterator, Mapping +from typing import Any + +# ── jargon scrubber ──────────────────────────────────────────────── + +# Same forbidden set the stub regression test pins, applied to *LLM* +# output too. Case-insensitive for ASCII; CJK kept verbatim. +_FORBIDDEN_WORDS: tuple[str, ...] = ( + "M1", + "M2", + "stub", + "adapter", + "milestone", + "占位", +) + +_FORBIDDEN_RE = re.compile( + r"(?i)(?:" + "|".join(re.escape(w) for w in _FORBIDDEN_WORDS) + r")" +) + +# Streaming look-ahead: keep this many tail chars buffered before emitting +# so we never split a forbidden word across two `token` events. +_LOOKAHEAD = max(len(w) for w in _FORBIDDEN_WORDS) + 1 + + +def scrub_jargon(text: str) -> str: + """Mask any forbidden engineering jargon. Preserves character count + so streaming offsets remain consistent.""" + + def _mask(m: re.Match[str]) -> str: + return "·" * len(m.group(0)) + + return _FORBIDDEN_RE.sub(_mask, text) + + +def _split_emit_buffer(buf: str) -> tuple[str, str]: + """Return (safe_to_emit_scrubbed, keep_buffered). + + Invariant: no forbidden match may be bisected by the cut, otherwise + the two halves would be emitted in separate `token` events and + reassemble into the original word in the user's DOM. + """ + if len(buf) <= _LOOKAHEAD: + return "", buf + cut = len(buf) - _LOOKAHEAD + for m in _FORBIDDEN_RE.finditer(buf): + # If the match straddles the cut point, pull the cut back to + # the match start so the whole word stays buffered (or, if the + # match is wholly in the safe prefix, leave cut alone — it gets + # scrubbed before emission). + if m.start() < cut < m.end(): + cut = m.start() + elif m.start() >= cut: + # Subsequent matches are all in the retained tail. + break + return scrub_jargon(buf[:cut]), buf[cut:] + + +# ── env config / gating ──────────────────────────────────────────── + + +def openai_enabled() -> bool: + """True when both base_url and api_key are configured. Gateway path + (milk-owned) is just a different base_url — adapter doesn't care.""" + return bool( + os.environ.get("XIAOF_OPENAI_API_KEY") + and os.environ.get("XIAOF_OPENAI_BASE_URL") + ) + + +def _config() -> dict[str, Any]: + return { + "base_url": os.environ["XIAOF_OPENAI_BASE_URL"].rstrip("/"), + "api_key": os.environ["XIAOF_OPENAI_API_KEY"], + "model": os.environ.get("XIAOF_OPENAI_MODEL", "deepseek-chat"), + "timeout": float(os.environ.get("XIAOF_OPENAI_TIMEOUT", "30")), + } + + +# ── system prompt ────────────────────────────────────────────────── + +SYSTEM_PROMPT = ( + "你是 OpenForge 的小F，一个简洁、直接的问答助手。" + "回答使用中文，控制在 200 字以内，自然口吻。" + "不要谈论自己是 AI / 模型 / 后端实现，不要使用工程黑话。" + "你看不到也不能讨论任何跨 thread 检索结果 —— 检索由独立路径处理；" + "如果用户在问历史 thread，礼貌告诉对方检索功能正在接通中，" + "并先回答 ta 的通用问题部分。" +) + + +# ── HTTP layer (override-friendly for tests) ─────────────────────── + + +def _urlopen(req: urllib.request.Request, timeout: float): # pragma: no cover - thin + return urllib.request.urlopen(req, timeout=timeout) + + +# ── adapter ──────────────────────────────────────────────────────── + + +def openai_compatible_adapter( + payload: Mapping[str, Any], +) -> Iterator[tuple[str, Mapping[str, Any]]]: + """OpenAI-compatible streaming adapter. + + Emits the standard meta → token* → chips → done sequence. Chips is + always empty here — this adapter never participates in retrieval, so + A-7's 12-case ACL fixture (M2) is structurally unable to be bypassed. + """ + # Late import avoids forge_xiaof ↔ forge_xiaof_openai circular dep. + from forge_xiaof import XiaofRequestError, classify_intent, new_request_id + + started = time.monotonic() + query = str(payload.get("query") or "") + intent = classify_intent(query) + request_id = new_request_id() + + yield ( + "meta", + {"intent": intent, "request_id": request_id, "provider": "openai-compat"}, + ) + + try: + config = _config() + except KeyError as exc: + raise XiaofRequestError( + f"openai adapter missing config: {exc!s}", code="upstream_failed" + ) from exc + + request_body = { + "model": config["model"], + "stream": True, + "messages": [ + {"role": "system", "content": SYSTEM_PROMPT}, + {"role": "user", "content": query}, + ], + } + req = urllib.request.Request( + config["base_url"] + "/chat/completions", + data=json.dumps(request_body, ensure_ascii=False).encode("utf-8"), + headers={ + "Authorization": "Bearer " + config["api_key"], + "Content-Type": "application/json", + "Accept": "text/event-stream", + }, + method="POST", + ) + + try: + resp = _urlopen(req, timeout=config["timeout"]) + except Exception as exc: # noqa: BLE001 + raise XiaofRequestError( + f"openai upstream failed: {exc!r}", code="upstream_failed" + ) from exc + + buffer = "" + emitted_chars = 0 + + try: + for raw in resp: + line = raw.decode("utf-8", errors="replace").rstrip("\r\n") + if not line.startswith("data:"): + continue + data = line[len("data:") :].strip() + if not data or data == "[DONE]": + if data == "[DONE]": + break + continue + try: + obj = json.loads(data) + except json.JSONDecodeError: + continue + choices = obj.get("choices") or [] + if not choices: + continue + delta_obj = choices[0].get("delta") or {} + delta = delta_obj.get("content") + if not isinstance(delta, str) or not delta: + continue + buffer += delta + emit, buffer = _split_emit_buffer(buffer) + if emit: + yield ("token", {"text": emit}) + emitted_chars += len(emit) + finally: + try: + resp.close() + except Exception: # noqa: BLE001 + pass + + # Final flush — anything left in the look-ahead buffer must still be + # scrubbed before reaching the user. + if buffer: + tail = scrub_jargon(buffer) + if tail: + yield ("token", {"text": tail}) + emitted_chars += len(tail) + + yield ("chips", {"chips": [], "chip_total": 0}) + yield ( + "done", + { + "latency_ms": int((time.monotonic() - started) * 1000), + "chip_count": 0, + "chip_total": 0, + "tokens_in": len(query), + "tokens_out": emitted_chars, + "provider": "openai-compat", + "model": config["model"], + }, + ) diff --git a/tests/test_xiaof_openai.py b/tests/test_xiaof_openai.py new file mode 100644 index 0000000..84b8a0a --- /dev/null +++ b/tests/test_xiaof_openai.py @@ -0,0 +1,200 @@ +""" +Tests for the OpenAI-compatible adapter + default routing. + +Covers: +- Jargon scrubbing on streamed LLM output (incl. split-across-chunks). +- env-gated enable; missing env → routes to stub, never crashes. +- Routing: time query → local builtin (no LLM); open general_qa with env + → openai adapter; thread_search → stub even with env set. +- Upstream failure → XiaofRequestError code='upstream_failed' (never + 'forbidden', A-7.5 red line preserved). +- Event order meta → token* → chips → done holds for the LLM path. +- Empty chips on LLM path (LLM cannot bypass A-7 retrieval ACL gate). +""" + +from __future__ import annotations + +import io +import json + +import pytest + + +@pytest.fixture +def openai_env(monkeypatch): + monkeypatch.setenv("XIAOF_OPENAI_BASE_URL", "https://fake.test/v1") + monkeypatch.setenv("XIAOF_OPENAI_API_KEY", "test-key") + monkeypatch.setenv("XIAOF_OPENAI_MODEL", "test-model") + + +def _sse_chunk(content: str) -> bytes: + payload = {"choices": [{"delta": {"content": content}}]} + return ("data: " + json.dumps(payload, ensure_ascii=False) + "\n").encode("utf-8") + + +def _fake_response(chunks: list[str]): + body = b"".join(_sse_chunk(c) for c in chunks) + b"data: [DONE]\n" + + class FakeResp(io.BytesIO): + def close(self): # noqa: D401 + super().close() + + def __iter__(self): + data = self.getvalue().splitlines(keepends=True) + yield from data + + return FakeResp(body) + + +# ── jargon scrubber ──────────────────────────────────────────────── + + +def test_scrub_jargon_basic(fake_home): + import forge_xiaof_openai as oa + assert oa.scrub_jargon("hello M1 world") == "hello ·· world" + assert oa.scrub_jargon("调用 stub 走 adapter") == "调用 ···· 走 ·······" + # Words we don't scrub are left alone. + assert oa.scrub_jargon("just a plain answer") == "just a plain answer" + + +def test_split_emit_buffer_keeps_lookahead(fake_home): + import forge_xiaof_openai as oa + emit, keep = oa._split_emit_buffer("the milestone is here") + assert "milestone" not in emit # would have been split + assert keep.endswith("here") or "milestone" in keep + emit + + +# ── env gating ───────────────────────────────────────────────────── + + +def test_openai_disabled_without_env(fake_home, monkeypatch): + monkeypatch.delenv("XIAOF_OPENAI_BASE_URL", raising=False) + monkeypatch.delenv("XIAOF_OPENAI_API_KEY", raising=False) + import forge_xiaof_openai as oa + assert oa.openai_enabled() is False + + +def test_openai_enabled_with_env(fake_home, openai_env): + import forge_xiaof_openai as oa + assert oa.openai_enabled() is True + + +# ── routing ──────────────────────────────────────────────────────── + + +def test_default_adapter_routes_time_query_to_stub(fake_home, openai_env, monkeypatch): + """Time query must use local zoneinfo, never hit the LLM.""" + import forge_xiaof + import forge_xiaof_openai as oa + + called = {"n": 0} + + def _boom(req, timeout): + called["n"] += 1 + raise AssertionError("LLM must not be called for time queries") + + monkeypatch.setattr(oa, "_urlopen", _boom) + events = list(forge_xiaof.default_adapter({"query": "现在 Asia/Shanghai 几点"})) + assert called["n"] == 0 + body = "".join(p["text"] for k, p in events if k == "token") + assert "Asia/Shanghai" in body + + +def test_default_adapter_routes_thread_search_to_stub(fake_home, openai_env, monkeypatch): + """thread_search MUST stay on stub even with LLM env present — + LLM path cannot bypass A-7 retrieval ACL gate.""" + import forge_xiaof + import forge_xiaof_openai as oa + + def _boom(req, timeout): + raise AssertionError("LLM must not be called for thread_search") + + monkeypatch.setattr(oa, "_urlopen", _boom) + events = list(forge_xiaof.default_adapter({"query": "上次 hero 高度的决定在哪条 post"})) + chips = next(p for k, p in events if k == "chips") + assert chips["chips"] == [] + + +def test_default_adapter_routes_open_qa_to_llm(fake_home, openai_env, monkeypatch): + """Open-ended general_qa with env present → real LLM adapter.""" + import forge_xiaof + import forge_xiaof_openai as oa + + captured = {} + + def _fake(req, timeout): + captured["url"] = req.full_url + captured["auth"] = req.headers.get("Authorization") + return _fake_response(["你好，", "我是回答。"]) + + monkeypatch.setattr(oa, "_urlopen", _fake) + + events = list(forge_xiaof.default_adapter({"query": "讲个冷笑话"})) + kinds = [k for k, _ in events] + assert kinds[0] == "meta" + assert kinds[-1] == "done" + assert kinds.count("chips") == 1 + assert "token" in kinds + body = "".join(p["text"] for k, p in events if k == "token") + assert body == "你好，我是回答。" + chips = next(p for k, p in events if k == "chips") + assert chips["chips"] == [] # LLM never produces chips + assert captured["url"].endswith("/chat/completions") + assert captured["auth"] == "Bearer test-key" + + +# ── jargon scrubber on LLM output ────────────────────────────────── + + +def test_llm_output_jargon_scrubbed(fake_home, openai_env, monkeypatch): + import forge_xiaof_openai as oa + + monkeypatch.setattr( + oa, + "_urlopen", + lambda req, timeout: _fake_response(["这次返回的 M1 stub 是个 milestone", " 占位文本"]), + ) + + events = list(oa.openai_compatible_adapter({"query": "test"})) + body = "".join(p["text"] for k, p in events if k == "token") + for bad in ("M1", "stub", "milestone", "占位", "adapter"): + assert bad not in body, f"forbidden word {bad!r} leaked in: {body!r}" + + +def test_llm_output_jargon_scrubbed_across_chunk_boundary(fake_home, openai_env, monkeypatch): + """A forbidden word split across two upstream chunks must still be + caught — that's the entire point of _LOOKAHEAD.""" + import forge_xiaof_openai as oa + + # "milestone" split as "mile" + "stone". + monkeypatch.setattr( + oa, + "_urlopen", + lambda req, timeout: _fake_response(["hello mile", "stone tail"]), + ) + events = list(oa.openai_compatible_adapter({"query": "x"})) + body = "".join(p["text"] for k, p in events if k == "token") + assert "milestone" not in body.lower() + + +# ── upstream failures ────────────────────────────────────────────── + + +def test_upstream_failure_maps_to_upstream_failed(fake_home, openai_env, monkeypatch): + import forge_xiaof_openai as oa + from forge_xiaof import XiaofRequestError + + def _broken(req, timeout): + raise ConnectionError("boom") + + monkeypatch.setattr(oa, "_urlopen", _broken) + + gen = oa.openai_compatible_adapter({"query": "anything"}) + # First event (meta) is fine; the network call happens on next(). + first_kind, _ = next(gen) + assert first_kind == "meta" + with pytest.raises(XiaofRequestError) as exc: + next(gen) + assert exc.value.code == "upstream_failed" + # A-7.5: must NEVER produce `forbidden`. + assert exc.value.code != "forbidden"