refactor: centralize llm provider transport

This commit is contained in:
leefer
2026-08-01 13:37:23 +08:00
parent 3d4fca7f65
commit 07e14c74b5
15 changed files with 462 additions and 260 deletions
+3 -2
View File
@@ -38,8 +38,9 @@ background scheduler
adapters. Root `database.py` remains the legacy schema/composition anchor and combines the
feature repository mixins; do not add feature queries to it.
- `backend/jobs/` owns job definitions, locks, retries, idempotency, and persisted run state.
- `backend/llm/` owns model selection, membership/quota checks, fallback, streaming rules,
and call audit. Feature agents only prepare context and provider payloads.
- `backend/llm/` owns model selection, membership/quota checks, fallback, provider transport,
streaming rules, and call audit. Feature agents only prepare messages and interpret
feature-specific results.
- `frontend/shared/` is the only browser API/state/Shell/component boundary.
- `frontend/pages/` owns page-local behavior. The original runtime was split mechanically;
source markers and preservation tests prove that the pieces reassemble to the audited
+20 -50
View File
@@ -1,11 +1,10 @@
from __future__ import annotations
import json
import time
import urllib.error
import urllib.request
from typing import Any
from backend.llm import transport as llm_transport
class HeavenAgentError(RuntimeError):
pass
@@ -24,45 +23,33 @@ def interpret_heaven(
if not api_key or not model:
raise HeavenAgentError("LLM API Key 或模型尚未配置。")
system_prompt = _system_prompt(mode)
payload = json.dumps(
messages = [
{"role": "system", "content": system_prompt},
{
"model": model,
"messages": [
{"role": "system", "content": system_prompt},
{
"role": "user",
"content": json.dumps(context, ensure_ascii=False, separators=(",", ":")),
},
],
"stream": False,
"role": "user",
"content": json.dumps(context, ensure_ascii=False, separators=(",", ":")),
},
ensure_ascii=False,
).encode("utf-8")
request = urllib.request.Request(
f"{base_url.rstrip('/')}/chat/completions",
data=payload,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}",
"User-Agent": "XiaobaiReviewWeb/0.7",
},
method="POST",
)
started = time.perf_counter()
]
try:
with urllib.request.urlopen(request, timeout=timeout) as response:
result = json.loads(response.read().decode("utf-8"))
answer = str(result["choices"][0]["message"]["content"]).strip()
result = llm_transport.chat_completion(
api_key=api_key,
base_url=base_url,
model=model,
messages=messages,
timeout=timeout,
user_agent="XiaobaiReviewWeb/0.7",
)
answer = str(result.content).strip()
if not answer:
raise KeyError("empty response")
except urllib.error.HTTPError as exc:
raise HeavenAgentError(_http_error_message(exc)) from exc
except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, KeyError, IndexError) as exc:
except llm_transport.OpenAIHTTPError as exc:
raise HeavenAgentError(exc.describe("问天模型调用失败")) from exc
except (llm_transport.OpenAITransportError, KeyError) as exc:
raise HeavenAgentError(f"问天模型调用失败:{exc}") from exc
return {
"answer": answer,
"model": model,
"latency_ms": round((time.perf_counter() - started) * 1000),
"latency_ms": result.latency_ms,
}
@@ -99,20 +86,3 @@ def _system_prompt(mode: str) -> str:
全文控制在180至350个中文字符。只写一句卦意;一小段动爻与之卦;最后三句极短的问心句。
不要重述六条爻辞,不猜用户未说出口的问题,不以吉凶二字替代思考,不给出股票涨跌预测。语气安静、克制,越短越有余味。
""".strip()
def _http_error_message(exc: urllib.error.HTTPError) -> str:
detail = ""
try:
payload = json.loads(exc.read().decode("utf-8", errors="replace"))
error = payload.get("error")
if isinstance(error, dict):
detail = str(error.get("message") or error.get("code") or "")
elif error:
detail = str(error)
elif payload.get("message"):
detail = str(payload["message"])
except (json.JSONDecodeError, OSError):
detail = ""
suffix = f"{detail[:300]}" if detail else ""
return f"问天模型调用失败(HTTP {exc.code}{suffix}"
+14 -63
View File
@@ -3,14 +3,12 @@ from __future__ import annotations
import json
import re
import time
import urllib.error
import urllib.request
from collections.abc import Iterator
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from llm_stream import OpenAIStreamAccumulator
from backend.llm import transport as llm_transport
class MentorAgentError(RuntimeError):
@@ -195,50 +193,20 @@ def stream_with_mentor(
messages = [{"role": "system", "content": system_prompt}]
messages.extend(history[-10:])
messages.append({"role": "user", "content": question})
payload = json.dumps(
{"model": model, "messages": messages, "stream": True},
ensure_ascii=False,
).encode("utf-8")
request = urllib.request.Request(
f"{base_url.rstrip('/')}/chat/completions",
data=payload,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}",
"User-Agent": "XiaobaiReviewWeb/0.6",
"Accept": "text/event-stream",
},
method="POST",
)
try:
with urllib.request.urlopen(request, timeout=timeout) as response:
yielded = False
accumulator = OpenAIStreamAccumulator()
for raw_line in response:
line = raw_line.decode("utf-8", errors="replace").strip()
if not line or line.startswith(":"):
continue
if line.startswith("data:"):
line = line[5:].strip()
if line == "[DONE]":
break
try:
result = json.loads(line)
except json.JSONDecodeError:
continue
choices = result.get("choices") or []
if not choices:
continue
choice = choices[0] or {}
content = accumulator.feed(choice)
if content:
yielded = True
yield str(content)
if not yielded:
raise MentorAgentError("问师模型未返回有效内容。")
except urllib.error.HTTPError as exc:
raise MentorAgentError(_http_error_message(exc)) from exc
except (urllib.error.URLError, TimeoutError, OSError) as exc:
yield from llm_transport.stream_chat_completion(
api_key=api_key,
base_url=base_url,
model=model,
messages=messages,
timeout=timeout,
user_agent="XiaobaiReviewWeb/0.6",
)
except llm_transport.OpenAIEmptyResponseError as exc:
raise MentorAgentError("问师模型未返回有效内容。") from exc
except llm_transport.OpenAIHTTPError as exc:
raise MentorAgentError(exc.describe("问师模型调用失败")) from exc
except llm_transport.OpenAITransportError as exc:
raise MentorAgentError(f"问师模型调用失败:{exc}") from exc
@@ -298,20 +266,3 @@ def _parse_frontmatter(content: str) -> dict[str, str]:
def _first_sentence(text: str) -> str:
compact = " ".join(line.strip() for line in text.splitlines() if line.strip())
return re.split(r"[。;]", compact, maxsplit=1)[0].strip()
def _http_error_message(exc: urllib.error.HTTPError) -> str:
detail = ""
try:
payload = json.loads(exc.read().decode("utf-8", errors="replace"))
error = payload.get("error")
if isinstance(error, dict):
detail = str(error.get("message") or error.get("code") or "")
elif error:
detail = str(error)
elif payload.get("message"):
detail = str(payload["message"])
except (json.JSONDecodeError, OSError):
detail = ""
suffix = f"{detail[:300]}" if detail else ""
return f"问师模型调用失败(HTTP {exc.code}{suffix}"
+13 -43
View File
@@ -1,12 +1,10 @@
from __future__ import annotations
import json
import urllib.error
import urllib.request
from collections.abc import Iterator
from typing import Any
from llm_stream import OpenAIStreamAccumulator
from backend.llm import transport as llm_transport
class ReviewAssistantError(RuntimeError):
@@ -27,48 +25,20 @@ def stream_review_assistant(
messages = [{"role": "system", "content": _system_prompt(context)}]
messages.extend(history[-12:])
messages.append({"role": "user", "content": question})
request = urllib.request.Request(
f"{base_url.rstrip('/')}/chat/completions",
data=json.dumps(
{"model": model, "messages": messages, "stream": True}, ensure_ascii=False
).encode("utf-8"),
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}",
"User-Agent": "XiaobaiReviewWeb/1.0",
"Accept": "text/event-stream",
},
method="POST",
)
try:
with urllib.request.urlopen(request, timeout=timeout) as response:
yielded = False
accumulator = OpenAIStreamAccumulator()
for raw_line in response:
line = raw_line.decode("utf-8", errors="replace").strip()
if not line or line.startswith(":"):
continue
if line.startswith("data:"):
line = line[5:].strip()
if line == "[DONE]":
break
try:
payload = json.loads(line)
except json.JSONDecodeError:
continue
choices = payload.get("choices") or []
if not choices:
continue
choice = choices[0] or {}
content = accumulator.feed(choice)
if content:
yielded = True
yield str(content)
if not yielded:
raise ReviewAssistantError("智能解读未返回有效内容。")
except urllib.error.HTTPError as exc:
yield from llm_transport.stream_chat_completion(
api_key=api_key,
base_url=base_url,
model=model,
messages=messages,
timeout=timeout,
user_agent="XiaobaiReviewWeb/1.0",
)
except llm_transport.OpenAIEmptyResponseError as exc:
raise ReviewAssistantError("智能解读未返回有效内容。") from exc
except llm_transport.OpenAIHTTPError as exc:
raise ReviewAssistantError(f"智能解读服务暂不可用({exc.code})。") from exc
except (urllib.error.URLError, TimeoutError, OSError) as exc:
except llm_transport.OpenAITransportError as exc:
raise ReviewAssistantError("智能解读连接中断,请稍后重试。") from exc
+27 -73
View File
@@ -1,11 +1,9 @@
from __future__ import annotations
import json
import time
import urllib.error
import urllib.request
from typing import Any
from backend.llm import transport as llm_transport
from screener import FACTOR_FIELDS, REGIMES
@@ -21,39 +19,25 @@ def test_llm_connection(
) -> dict[str, Any]:
if not api_key or not model:
raise LLMCompilerError("API Key 或模型未配置。")
endpoint = f"{base_url.rstrip('/')}/chat/completions"
payload = json.dumps(
{
"model": model,
"messages": [{"role": "user", "content": "只回复 OK"}],
"stream": False,
},
ensure_ascii=False,
).encode("utf-8")
request = urllib.request.Request(
endpoint,
data=payload,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}",
"User-Agent": "XiaobaiReviewWeb/0.5",
},
method="POST",
)
started = time.perf_counter()
try:
with urllib.request.urlopen(request, timeout=timeout) as response:
result = json.loads(response.read().decode("utf-8"))
reply = str(result["choices"][0]["message"]["content"]).strip()
except urllib.error.HTTPError as exc:
raise LLMCompilerError(_http_error_message(exc)) from exc
except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, KeyError, IndexError) as exc:
result = llm_transport.chat_completion(
api_key=api_key,
base_url=base_url,
model=model,
messages=[{"role": "user", "content": "只回复 OK"}],
timeout=timeout,
user_agent="XiaobaiReviewWeb/0.5",
)
reply = str(result.content).strip()
except llm_transport.OpenAIHTTPError as exc:
raise LLMCompilerError(exc.describe("模型连接测试失败")) from exc
except llm_transport.OpenAITransportError as exc:
raise LLMCompilerError(f"模型连接测试失败:{exc}") from exc
return {
"ok": True,
"model": model,
"reply": reply[:100],
"latency_ms": round((time.perf_counter() - started) * 1000),
"latency_ms": result.latency_ms,
}
@@ -67,7 +51,6 @@ def compile_strategy_with_llm(
) -> dict[str, Any]:
if not api_key or not model:
raise LLMCompilerError("尚未配置 LLM API Key 或模型。")
endpoint = f"{base_url.rstrip('/')}/chat/completions"
schema = {
"name": "策略名称",
"description": "策略说明",
@@ -90,57 +73,28 @@ def compile_strategy_with_llm(
"退潮和冰点策略必须提高门槛并允许结果为空。"
f"严格遵循以下结构:{json.dumps(schema, ensure_ascii=False)}"
)
payload = json.dumps(
{
"model": model,
"messages": [
try:
result = llm_transport.chat_completion(
api_key=api_key,
base_url=base_url,
model=model,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": prompt[:3000]},
],
"stream": False,
},
ensure_ascii=False,
).encode("utf-8")
request = urllib.request.Request(
endpoint,
data=payload,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}",
"User-Agent": "XiaobaiReviewWeb/0.4",
},
method="POST",
)
try:
with urllib.request.urlopen(request, timeout=timeout) as response:
result = json.loads(response.read().decode("utf-8"))
content = result["choices"][0]["message"]["content"].strip()
timeout=timeout,
user_agent="XiaobaiReviewWeb/0.4",
)
content = result.content.strip()
if content.startswith("```"):
content = content.strip("`")
if content.startswith("json"):
content = content[4:].strip()
compiled = json.loads(content)
except urllib.error.HTTPError as exc:
raise LLMCompilerError(_http_error_message(exc).replace("模型连接测试", "LLM 策略编译")) from exc
except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, KeyError, IndexError) as exc:
except llm_transport.OpenAIHTTPError as exc:
raise LLMCompilerError(exc.describe("LLM 策略编译失败")) from exc
except (llm_transport.OpenAITransportError, json.JSONDecodeError) as exc:
raise LLMCompilerError(f"LLM 策略编译失败:{exc}") from exc
compiled["compiler"] = "llm"
compiled["model"] = model
return compiled
def _http_error_message(exc: urllib.error.HTTPError) -> str:
detail = ""
try:
payload = json.loads(exc.read().decode("utf-8", errors="replace"))
error = payload.get("error")
if isinstance(error, dict):
detail = str(error.get("message") or error.get("code") or "")
elif error:
detail = str(error)
elif payload.get("message"):
detail = str(payload["message"])
except (json.JSONDecodeError, OSError):
detail = ""
suffix = f"{detail[:300]}" if detail else ""
return f"模型连接测试失败(HTTP {exc.code}{suffix}"
+148
View File
@@ -0,0 +1,148 @@
from __future__ import annotations
import json
import time
import urllib.error
import urllib.request
from collections.abc import Iterator
from dataclasses import dataclass
from typing import Any
from .stream import OpenAIStreamAccumulator
class OpenAITransportError(RuntimeError):
pass
class OpenAIHTTPError(OpenAITransportError):
def __init__(self, code: int, detail: str = "") -> None:
super().__init__(f"HTTP {code}")
self.code = code
self.detail = detail
def describe(self, label: str) -> str:
suffix = f"{self.detail[:300]}" if self.detail else ""
return f"{label}HTTP {self.code}{suffix}"
class OpenAIEmptyResponseError(OpenAITransportError):
pass
@dataclass(frozen=True)
class OpenAIChatCompletion:
content: Any
latency_ms: int
def chat_completion(
*,
api_key: str,
base_url: str,
model: str,
messages: list[dict[str, Any]],
timeout: int,
user_agent: str,
) -> OpenAIChatCompletion:
request = _request(api_key, base_url, model, messages, user_agent, stream=False)
started = time.perf_counter()
try:
with urllib.request.urlopen(request, timeout=timeout) as response:
result = json.loads(response.read().decode("utf-8"))
content = result["choices"][0]["message"]["content"]
except urllib.error.HTTPError as exc:
raise OpenAIHTTPError(exc.code, _http_error_detail(exc)) from exc
except (
urllib.error.URLError,
TimeoutError,
json.JSONDecodeError,
KeyError,
IndexError,
) as exc:
raise OpenAITransportError(str(exc)) from exc
return OpenAIChatCompletion(
content=content,
latency_ms=round((time.perf_counter() - started) * 1000),
)
def stream_chat_completion(
*,
api_key: str,
base_url: str,
model: str,
messages: list[dict[str, Any]],
timeout: int,
user_agent: str,
) -> Iterator[str]:
request = _request(api_key, base_url, model, messages, user_agent, stream=True)
yielded = False
try:
with urllib.request.urlopen(request, timeout=timeout) as response:
accumulator = OpenAIStreamAccumulator()
for raw_line in response:
line = raw_line.decode("utf-8", errors="replace").strip()
if not line or line.startswith(":"):
continue
if line.startswith("data:"):
line = line[5:].strip()
if line == "[DONE]":
break
try:
result = json.loads(line)
except json.JSONDecodeError:
continue
choices = result.get("choices") or []
if not choices:
continue
content = accumulator.feed(choices[0] or {})
if content:
yielded = True
yield str(content)
except urllib.error.HTTPError as exc:
raise OpenAIHTTPError(exc.code, _http_error_detail(exc)) from exc
except (urllib.error.URLError, TimeoutError, OSError) as exc:
raise OpenAITransportError(str(exc)) from exc
if not yielded:
raise OpenAIEmptyResponseError("empty response")
def _request(
api_key: str,
base_url: str,
model: str,
messages: list[dict[str, Any]],
user_agent: str,
*,
stream: bool,
) -> urllib.request.Request:
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}",
"User-Agent": user_agent,
}
if stream:
headers["Accept"] = "text/event-stream"
return urllib.request.Request(
f"{base_url.rstrip('/')}/chat/completions",
data=json.dumps(
{"model": model, "messages": messages, "stream": stream},
ensure_ascii=False,
).encode("utf-8"),
headers=headers,
method="POST",
)
def _http_error_detail(exc: urllib.error.HTTPError) -> str:
try:
payload = json.loads(exc.read().decode("utf-8", errors="replace"))
error = payload.get("error")
if isinstance(error, dict):
return str(error.get("message") or error.get("code") or "")
if error:
return str(error)
return str(payload.get("message") or "")
except (json.JSONDecodeError, OSError):
return ""
+11 -1
View File
@@ -1,6 +1,6 @@
{
"schema_version": 1,
"captured_from": "app modular preservation candidate",
"captured_from": "app accepted modular runtime",
"runtime": {
"http_server": "http.server.ThreadingHTTPServer",
"application_processes": 1,
@@ -242,6 +242,16 @@
"path": "backend/features/screener/compiler.py"
}
],
"llm_transport": [
{
"function": "chat_completion",
"path": "backend/llm/transport.py"
},
{
"function": "stream_chat_completion",
"path": "backend/llm/transport.py"
}
],
"css_layers": [
"/shared/tokens.css?v=20260729-1",
"/styles/styles.css",
+187
View File
@@ -0,0 +1,187 @@
from __future__ import annotations
import io
import json
import unittest
import urllib.error
from pathlib import Path
from unittest.mock import patch
from assistant_agent import ReviewAssistantError, stream_review_assistant
from backend.llm import transport
from heaven_agent import HeavenAgentError, interpret_heaven
from llm_strategy import LLMCompilerError, test_llm_connection
from mentor_agent import MentorAgentError, MentorSkill, stream_with_mentor
ROOT = Path(__file__).resolve().parents[1]
class FakeResponse:
def __init__(self, *, payload: bytes = b"", lines: list[bytes] | None = None) -> None:
self.payload = payload
self.lines = lines or []
def __enter__(self):
return self
def __exit__(self, exc_type, exc_value, traceback):
return False
def read(self) -> bytes:
return self.payload
def __iter__(self):
return iter(self.lines)
class OpenAITransportTests(unittest.TestCase):
def test_chat_completion_builds_one_openai_compatible_request(self) -> None:
captured = {}
response = FakeResponse(
payload=json.dumps(
{"choices": [{"message": {"content": "OK"}}]}
).encode("utf-8")
)
def open_request(request, timeout):
captured["url"] = request.full_url
captured["headers"] = request.headers
captured["payload"] = json.loads(request.data.decode("utf-8"))
captured["timeout"] = timeout
return response
with patch("backend.llm.transport.urllib.request.urlopen", side_effect=open_request):
result = transport.chat_completion(
api_key="secret",
base_url="https://example.test/v1/",
model="model",
messages=[{"role": "user", "content": "ping"}],
timeout=17,
user_agent="XiaobaiReviewWeb/test",
)
self.assertEqual(result.content, "OK")
self.assertGreaterEqual(result.latency_ms, 0)
self.assertEqual(captured["url"], "https://example.test/v1/chat/completions")
self.assertEqual(captured["payload"]["stream"], False)
self.assertEqual(captured["headers"]["Authorization"], "Bearer secret")
self.assertEqual(captured["timeout"], 17)
def test_stream_completion_parses_deltas_and_ignores_final_snapshot(self) -> None:
response = FakeResponse(
lines=[
b'data: {"choices":[{"delta":{"content":"first"}}]}\n',
b'data: {"choices":[{"delta":{"content":" second"}}]}\n',
b'data: {"choices":[{"message":{"content":"first second"}}]}\n',
b"data: [DONE]\n",
]
)
with patch("backend.llm.transport.urllib.request.urlopen", return_value=response):
chunks = list(
transport.stream_chat_completion(
api_key="secret",
base_url="https://example.test/v1",
model="model",
messages=[],
timeout=17,
user_agent="XiaobaiReviewWeb/test",
)
)
self.assertEqual(chunks, ["first", " second"])
def test_empty_stream_has_a_stable_transport_error(self) -> None:
response = FakeResponse(lines=[b"data: [DONE]\n"])
with patch("backend.llm.transport.urllib.request.urlopen", return_value=response):
with self.assertRaises(transport.OpenAIEmptyResponseError):
list(
transport.stream_chat_completion(
api_key="secret",
base_url="https://example.test/v1",
model="model",
messages=[],
timeout=17,
user_agent="XiaobaiReviewWeb/test",
)
)
def test_http_error_keeps_code_and_sanitized_provider_detail(self) -> None:
error = urllib.error.HTTPError(
"https://example.test/v1/chat/completions",
429,
"rate limited",
{},
io.BytesIO(b'{"error":{"message":"capacity"}}'),
)
self.addCleanup(error.close)
with patch("backend.llm.transport.urllib.request.urlopen", side_effect=error):
with self.assertRaises(transport.OpenAIHTTPError) as caught:
transport.chat_completion(
api_key="secret",
base_url="https://example.test/v1",
model="model",
messages=[],
timeout=17,
user_agent="XiaobaiReviewWeb/test",
)
self.assertEqual(caught.exception.code, 429)
self.assertEqual(
caught.exception.describe("模型调用失败"),
"模型调用失败(HTTP 429):capacity",
)
def test_feature_agents_have_no_direct_provider_transport(self) -> None:
paths = (
"backend/features/mentor/agent.py",
"backend/features/heaven/agent.py",
"backend/features/review/agent.py",
"backend/features/screener/compiler.py",
)
for relative in paths:
source = (ROOT / relative).read_text(encoding="utf-8")
with self.subTest(path=relative):
self.assertNotIn("urllib.request", source)
self.assertNotIn("/chat/completions", source)
self.assertIn("llm_transport.", source)
class FeatureErrorMappingTests(unittest.TestCase):
def test_feature_specific_http_messages_are_preserved(self) -> None:
error = transport.OpenAIHTTPError(429, "capacity")
skill = MentorSkill(
skill_id="test",
name="测试老师",
description="",
tagline="",
focus=(),
content="",
path=Path("SKILL.md"),
)
with patch(
"mentor_agent.llm_transport.stream_chat_completion", side_effect=error
):
with self.assertRaisesRegex(
MentorAgentError, "问师模型调用失败(HTTP 429):capacity"
):
list(stream_with_mentor(skill, {}, "问题", [], "key", "https://x", "m"))
with patch("heaven_agent.llm_transport.chat_completion", side_effect=error):
with self.assertRaisesRegex(
HeavenAgentError, "问天模型调用失败(HTTP 429):capacity"
):
interpret_heaven("heart", {}, "key", "https://x", "m")
with patch(
"assistant_agent.llm_transport.stream_chat_completion", side_effect=error
):
with self.assertRaisesRegex(
ReviewAssistantError, "智能解读服务暂不可用(429"
):
list(stream_review_assistant({}, "问题", [], "key", "https://x", "m"))
with patch("llm_strategy.llm_transport.chat_completion", side_effect=error):
with self.assertRaisesRegex(
LLMCompilerError, "模型连接测试失败(HTTP 429):capacity"
):
test_llm_connection("key", "https://x", "m")
if __name__ == "__main__":
unittest.main()
+4 -4
View File
@@ -44,7 +44,7 @@ class MentorStreamTests(unittest.TestCase):
captured["accept"] = request.headers.get("Accept")
return FakeStreamResponse(self.lines)
with patch("mentor_agent.urllib.request.urlopen", side_effect=open_request):
with patch("backend.llm.transport.urllib.request.urlopen", side_effect=open_request):
chunks = list(
stream_with_mentor(
self.skill, {"data_trade_date": "20260723"}, "怎么看?", [],
@@ -58,7 +58,7 @@ class MentorStreamTests(unittest.TestCase):
def test_non_streaming_compatibility_wrapper_collects_chunks(self):
with patch(
"mentor_agent.urllib.request.urlopen",
"backend.llm.transport.urllib.request.urlopen",
return_value=FakeStreamResponse(self.lines),
):
result = chat_with_mentor(
@@ -74,7 +74,7 @@ class MentorStreamTests(unittest.TestCase):
b"data: [DONE]\n",
]
with patch(
"mentor_agent.urllib.request.urlopen",
"backend.llm.transport.urllib.request.urlopen",
return_value=FakeStreamResponse(lines),
):
chunks = list(
@@ -92,7 +92,7 @@ class MentorStreamTests(unittest.TestCase):
b"data: [DONE]\n",
]
with patch(
"mentor_agent.urllib.request.urlopen",
"backend.llm.transport.urllib.request.urlopen",
return_value=FakeStreamResponse(lines),
):
chunks = list(
+6 -5
View File
@@ -91,11 +91,12 @@ class HeavenSliceSourceEquivalenceTests(unittest.TestCase):
for name in sorted(expected - (adapted or set())):
self.assertEqual(migrated[name], original[name], name)
def test_heaven_agent_is_an_exact_file(self) -> None:
self.assertEqual(
sha256(ORIGINAL_ROOT / "heaven_agent.py"),
sha256(APP_ROOT / "backend" / "features" / "heaven" / "agent.py"),
)
def test_heaven_agent_uses_shared_transport(self) -> None:
source = (
APP_ROOT / "backend" / "features" / "heaven" / "agent.py"
).read_text(encoding="utf-8")
self.assertIn("llm_transport.chat_completion", source)
self.assertNotIn("urllib.request", source)
def test_heaven_engine_definitions_are_exact_original_ast(self) -> None:
self.assertEqual(
+5 -4
View File
@@ -105,11 +105,12 @@ class MentorLLMSliceSourceEquivalenceTests(unittest.TestCase):
for name in sorted(expected):
self.assertEqual(migrated[name], original[name], name)
def test_mentor_agent_and_stream_accumulator_are_exact_files(self) -> None:
self.assertEqual(
sha256(ORIGINAL_ROOT / "mentor_agent.py"),
sha256(APP_ROOT / "backend" / "features" / "mentor" / "agent.py"),
def test_mentor_agent_uses_shared_transport_and_stream_accumulator_is_exact(self) -> None:
mentor_source = (APP_ROOT / "backend" / "features" / "mentor" / "agent.py").read_text(
encoding="utf-8"
)
self.assertIn("llm_transport.stream_chat_completion", mentor_source)
self.assertNotIn("urllib.request", mentor_source)
self.assertEqual(
sha256(ORIGINAL_ROOT / "llm_stream.py"),
sha256(APP_ROOT / "backend" / "llm" / "stream.py"),
@@ -109,11 +109,12 @@ class ReviewAlertsSliceSourceEquivalenceTests(unittest.TestCase):
for name in sorted(expected):
self.assertEqual(migrated[name], original[name], name)
def test_review_assistant_agent_is_an_exact_file(self) -> None:
self.assertEqual(
sha256(ORIGINAL_ROOT / "assistant_agent.py"),
sha256(APP_ROOT / "backend" / "features" / "review" / "agent.py"),
)
def test_review_assistant_agent_uses_shared_transport(self) -> None:
source = (
APP_ROOT / "backend" / "features" / "review" / "agent.py"
).read_text(encoding="utf-8")
self.assertIn("llm_transport.stream_chat_completion", source)
self.assertNotIn("urllib.request", source)
def test_review_assistant_compatibility_module_is_canonical(self) -> None:
self.assertIs(assistant_agent, canonical_agent)
+10 -6
View File
@@ -170,12 +170,16 @@ class ScreenerSliceSourceEquivalenceTests(unittest.TestCase):
),
)
def test_library_and_compiler_files_are_exact_copies(self) -> None:
for original, migrated in (
("advanced_strategies.py", "backend/features/screener/strategies.py"),
("llm_strategy.py", "backend/features/screener/compiler.py"),
):
self.assertEqual(sha256(ORIGINAL_ROOT / original), sha256(APP_ROOT / migrated))
def test_library_is_exact_and_compiler_uses_shared_transport(self) -> None:
self.assertEqual(
sha256(ORIGINAL_ROOT / "advanced_strategies.py"),
sha256(APP_ROOT / "backend/features/screener/strategies.py"),
)
compiler_source = (
APP_ROOT / "backend/features/screener/compiler.py"
).read_text(encoding="utf-8")
self.assertEqual(compiler_source.count("llm_transport.chat_completion"), 2)
self.assertNotIn("urllib.request", compiler_source)
def test_compatibility_modules_export_the_canonical_objects(self) -> None:
self.assertIs(screener, engine)
+3 -3
View File
@@ -31,7 +31,7 @@ class ReviewAssistantStreamingTests(unittest.TestCase):
b'data: [DONE]\n',
]
)
with patch("assistant_agent.urllib.request.urlopen", return_value=response):
with patch("backend.llm.transport.urllib.request.urlopen", return_value=response):
chunks = list(
stream_review_assistant({}, "question", [], "key", "https://example.test/v1", "model")
)
@@ -39,7 +39,7 @@ class ReviewAssistantStreamingTests(unittest.TestCase):
def test_empty_stream_is_rejected(self):
response = StreamingResponse([b"data: [DONE]\n"])
with patch("assistant_agent.urllib.request.urlopen", return_value=response):
with patch("backend.llm.transport.urllib.request.urlopen", return_value=response):
with self.assertRaises(ReviewAssistantError):
list(
stream_review_assistant({}, "question", [], "key", "https://example.test/v1", "model")
@@ -54,7 +54,7 @@ class ReviewAssistantStreamingTests(unittest.TestCase):
b"data: [DONE]\n",
]
)
with patch("assistant_agent.urllib.request.urlopen", return_value=response):
with patch("backend.llm.transport.urllib.request.urlopen", return_value=response):
chunks = list(
stream_review_assistant(
{}, "question", [], "key", "https://example.test/v1", "model"
+5 -1
View File
@@ -127,7 +127,7 @@ def build() -> dict[str, Any]:
tables = database_inventory(database_sources)
return {
"schema_version": 1,
"captured_from": "app modular preservation candidate",
"captured_from": "app accepted modular runtime",
"runtime": {
"http_server": "http.server.ThreadingHTTPServer",
"application_processes": 1,
@@ -163,6 +163,10 @@ def build() -> dict[str, Any]:
{"function": "compile_strategy_with_llm", "path": "backend/features/screener/compiler.py"},
{"function": "test_llm_connection", "path": "backend/features/screener/compiler.py"},
],
"llm_transport": [
{"function": "chat_completion", "path": "backend/llm/transport.py"},
{"function": "stream_chat_completion", "path": "backend/llm/transport.py"},
],
"css_layers": css_layers(html),
"code_hotspots": code_hotspots(),
}