feat(v2): T3 ArchitectClient（DeepSeek API，JSON 约束输出）
This commit is contained in:
tzt
@@ -0,0 +1,228 @@
|
||||
"""ArchitectClient —— 大模型（API）客户端，端云协同的"决策/终审"角色。
|
||||
|
||||
职责（对齐《实现方案_v2》5.1 / 6.3）：
|
||||
- brief(query)：开局任务分析 -> 生成交流文本的 brief（goal/constraints/acceptance/plan/tags）
|
||||
- decide(ws)：读 issues 等 -> 输出裁决（reply + patch_plan 修订计划）
|
||||
- final_review(ws)：终审 -> {verdict: done|fix, issues: [...]}
|
||||
|
||||
工程约束（D7 / D8 / D9 / D11）：
|
||||
- 走 OpenAI 兼容 /chat/completions；response_format={"type":"json_object"}，prompt 内嵌 schema 描述。
|
||||
- Architect 输入永不包含工件全文：只传 Workspace 渲染出的 meta+issues+decisions+锚点片段（render_for_architect）。
|
||||
- 所有结构化输出解析为 JSON；失败把错误回喂重写一次，仍失败抛 ArchitectError（由编排层降级，禁止带病继续）。
|
||||
- token 计量回写 ws.meta.budget；调用前先查预算，触顶抛 ArchitectCircuitBreaker（D6）。
|
||||
- httpx 惰性导入（零顶层依赖，对齐仓库既有 APIExpert 模式）；client/transport 可注入，测试用 httpx.MockTransport（D11）。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from typing import Any, Callable, Dict, List, Optional
|
||||
|
||||
from .workspace import Workspace
|
||||
|
||||
# brief 的 JSON schema（描述性提示，约束模型输出结构）
|
||||
_BRIEF_SCHEMA_HINT = {
|
||||
"goal": "string(<=500字)",
|
||||
"constraints": "string[](<=8条)",
|
||||
"tags": "string[](code/math/legal/medical/finance/life/education/general/safety 之一)",
|
||||
"acceptance": "list[{id, check(string), machine_checkable(bool)}]",
|
||||
"plan": "list[{id, task(string<=300字), deps(string[]), done_criteria(string)}](<=5步, 有依赖序)",
|
||||
}
|
||||
|
||||
_DECIDE_SCHEMA_HINT = {
|
||||
"reply": "string(<=600字)",
|
||||
"patch_plan": "list[{id, task(string)}]",
|
||||
}
|
||||
|
||||
_REVIEW_SCHEMA_HINT = {
|
||||
"verdict": "enum(done|fix)",
|
||||
"notes": "string(<=300字)",
|
||||
"fix_issues": "list[string]",
|
||||
}
|
||||
|
||||
_SYSTEM_PROMPT = (
|
||||
"你是任务分析架构师。你的输入是不含工件全文的协作摘要（交流文本），"
|
||||
"你的输出必须是合法 JSON 对象（不要用 markdown 代码块包裹）。"
|
||||
)
|
||||
|
||||
|
||||
class ArchitectError(RuntimeError):
|
||||
"""Architect 调用失败（网络/超时/JSON 解析失败/服务错误）。"""
|
||||
|
||||
|
||||
class ArchitectCircuitBreaker(RuntimeError):
|
||||
"""预算熔断（D6）：api_token_cap / rounds_cap 触顶。"""
|
||||
|
||||
|
||||
class ArchitectClient:
|
||||
"""DeepSeek（或任意 OpenAI 兼容）大模型客户端。"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
model: str,
|
||||
base_url: str = "https://api.deepseek.com/v1",
|
||||
api_key: Optional[str] = None,
|
||||
temperature: float = 0.2,
|
||||
timeout_s: float = 60.0,
|
||||
max_tokens: int = 2048,
|
||||
transport: Any = None,
|
||||
_client: Any = None,
|
||||
):
|
||||
self.model = model
|
||||
self.base_url = base_url.rstrip("/")
|
||||
self.api_key = api_key
|
||||
self.temperature = temperature
|
||||
self.timeout_s = timeout_s
|
||||
self.max_tokens = max_tokens
|
||||
self._transport = transport
|
||||
self._client = _client # 注入的 AsyncClient（测试用 MockTransport）
|
||||
self._owns_client = _client is None
|
||||
|
||||
def _get_client(self):
|
||||
if self._client is None:
|
||||
import httpx
|
||||
kwargs: Dict[str, Any] = {"timeout": self.timeout_s}
|
||||
if self._transport is not None:
|
||||
kwargs["transport"] = self._transport
|
||||
self._client = httpx.AsyncClient(**kwargs)
|
||||
return self._client
|
||||
|
||||
async def aclose(self) -> None:
|
||||
if self._owns_client and self._client is not None:
|
||||
await self._client.aclose()
|
||||
self._client = None
|
||||
|
||||
# ---------------------------------------------------------------
|
||||
# 三个对外能力
|
||||
# ---------------------------------------------------------------
|
||||
async def brief(self, query: str, ws: Workspace) -> Dict[str, Any]:
|
||||
"""生成 brief。返回解析后的 brief dict；token 计量写入 ws。"""
|
||||
user = (
|
||||
"用户原始需求：" + "\n" + query + "\n\n"
|
||||
"请生成任务 brief，仅输出符合如下结构的 JSON 对象：" + "\n"
|
||||
+ json.dumps(_BRIEF_SCHEMA_HINT, ensure_ascii=False)
|
||||
+ "\n注意：plan 中的 id 用 s1..sn，deps 引用已完成步骤 id；"
|
||||
"acceptance 尽量 machine_checkable。"
|
||||
)
|
||||
return await self._chat_with_retry(ws, [("system", _SYSTEM_PROMPT), ("user", user)])
|
||||
|
||||
async def decide(self, ws: Workspace) -> Dict[str, Any]:
|
||||
"""根据交流文本当前状态做裁决。返回 {reply, patch_plan}。"""
|
||||
context = ws.render_for_architect()
|
||||
user = (
|
||||
"以下是交流文本摘要（不含工件全文）：" + "\n\n" + context + "\n\n"
|
||||
"请针对未解决 issues 做出裁决，仅输出符合如下结构的 JSON：" + "\n"
|
||||
+ json.dumps(_DECIDE_SCHEMA_HINT, ensure_ascii=False)
|
||||
+ "\nreply 给 Worker 具体可执行指示；patch_plan 列出需要修订的 step 与任务。"
|
||||
)
|
||||
return await self._chat_with_retry(ws, [("system", _SYSTEM_PROMPT), ("user", user)])
|
||||
|
||||
async def final_review(self, ws: Workspace) -> Dict[str, Any]:
|
||||
"""终审。返回 {verdict: done|fix, notes, fix_issues}。"""
|
||||
context = ws.render_for_architect()
|
||||
user = (
|
||||
"以下是待终审的交流文本摘要：" + "\n\n" + context + "\n\n"
|
||||
"对照 brief 的 acceptance 做终审，仅输出符合如下结构的 JSON：" + "\n"
|
||||
+ json.dumps(_REVIEW_SCHEMA_HINT, ensure_ascii=False)
|
||||
+ "\nverdict=done 表示验收通过；fix 表示打回，fix_issues 列出需修正项。"
|
||||
)
|
||||
return await self._chat_with_retry(ws, [("system", _SYSTEM_PROMPT), ("user", user)])
|
||||
|
||||
# ---------------------------------------------------------------
|
||||
# 底层
|
||||
# ---------------------------------------------------------------
|
||||
async def _chat_with_retry(self, ws: Workspace,
|
||||
msgs: List[tuple]) -> Dict[str, Any]:
|
||||
"""调用 + JSON 解析；解析失败回喂一次重写，再失败抛 ArchitectError。"""
|
||||
if not self.api_key:
|
||||
raise ArchitectError(
|
||||
"Architect 未配置 API Key（env: 见 config.architect.api_key_env）。"
|
||||
"请设置密钥，或使用本地降级模式（pipeline.breach_policy: local_only）。"
|
||||
)
|
||||
messages = [{"role": r, "content": c} for r, c in msgs]
|
||||
for attempt in (1, 2):
|
||||
content = await self._chat_once(ws, messages)
|
||||
try:
|
||||
return self._parse_json(content)
|
||||
except ValueError as e:
|
||||
if attempt == 1:
|
||||
# 4.6：把原始输出与错误回喂重写一次
|
||||
messages = messages + [
|
||||
{"role": "assistant", "content": content},
|
||||
{"role": "user",
|
||||
"content": f"你的输出不是合法 JSON（{e}）。请重新只输出合法 JSON 对象。"},
|
||||
]
|
||||
continue
|
||||
raise ArchitectError(f"Architect 输出非合法 JSON，重试后仍失败: {e}") from e
|
||||
raise ArchitectError("未预期：_chat_with_retry 未返回") # 不可达
|
||||
|
||||
async def _chat_once(self, ws: Workspace, messages: List[Dict[str, Any]]) -> str:
|
||||
if ws.exhausted():
|
||||
raise ArchitectCircuitBreaker(
|
||||
f"预算熔断：api_tokens={ws.budget()['api_input_tokens'] + ws.budget()['api_output_tokens']}"
|
||||
f"/{ws.budget()['api_token_cap']}, round={ws.meta()['round']}/{ws.budget()['rounds_cap']}"
|
||||
)
|
||||
client = self._get_client()
|
||||
headers = {"Authorization": f"Bearer {self.api_key}"} if self.api_key else {}
|
||||
body = {
|
||||
"model": self.model,
|
||||
"messages": messages,
|
||||
"temperature": self.temperature,
|
||||
"max_tokens": self.max_tokens,
|
||||
"response_format": {"type": "json_object"},
|
||||
}
|
||||
try:
|
||||
resp = await client.post(
|
||||
f"{self.base_url}/chat/completions",
|
||||
headers=headers,
|
||||
json=body,
|
||||
)
|
||||
resp.raise_for_status()
|
||||
except Exception as e:
|
||||
raise ArchitectError(f"Architect API 调用失败: {type(e).__name__}: {e}") from e
|
||||
data = resp.json()
|
||||
usage = data.get("usage", {})
|
||||
ws.add_budget(usage.get("prompt_tokens", 0), usage.get("completion_tokens", 0))
|
||||
try:
|
||||
return data["choices"][0]["message"]["content"]
|
||||
except (KeyError, IndexError) as e:
|
||||
raise ArchitectError(f"Architect 响应缺少 choices/content: {e}") from e
|
||||
|
||||
@staticmethod
|
||||
def _parse_json(content: str) -> Dict[str, Any]:
|
||||
"""从模型输出解析 JSON：剥除 markdown 代码围栏，取首个 JSON 对象。"""
|
||||
text = content.strip()
|
||||
_bt = "\x60" # 反引号（避免与构建脚本的字符串定界冲突）
|
||||
fence = _bt * 3
|
||||
text = text.replace(fence + "json", fence).replace(fence, "").strip()
|
||||
try:
|
||||
obj = json.loads(text)
|
||||
if isinstance(obj, dict):
|
||||
return obj
|
||||
raise ValueError("顶层不是 object")
|
||||
except json.JSONDecodeError:
|
||||
pass
|
||||
start = text.find("{")
|
||||
end = text.rfind("}")
|
||||
if start != -1 and end != -1 and end > start:
|
||||
try:
|
||||
obj = json.loads(text[start:end + 1])
|
||||
if isinstance(obj, dict):
|
||||
return obj
|
||||
except json.JSONDecodeError:
|
||||
pass
|
||||
raise ValueError("无法解析为 JSON 对象")
|
||||
|
||||
|
||||
def build_architect(cfg: Dict[str, Any],
|
||||
get_env: Callable[[str], Optional[str]] = None) -> ArchitectClient:
|
||||
"""cfg 为 config.architect 段。get_env 可注入（默认读 os.environ）。"""
|
||||
import os
|
||||
_env = get_env or os.environ.get
|
||||
key = cfg.get("api_key") or _env(cfg.get("api_key_env", "DEEPSEEK_API_KEY"))
|
||||
return ArchitectClient(
|
||||
model=cfg.get("model", "deepseek-chat"),
|
||||
base_url=cfg.get("base_url", "https://api.deepseek.com/v1"),
|
||||
api_key=key,
|
||||
temperature=float(cfg.get("temperature", 0.2)),
|
||||
timeout_s=float(cfg.get("timeout_s", 60)),
|
||||
)
|
||||
@@ -0,0 +1,169 @@
|
||||
"""T3 ArchitectClient 单测（封闭：httpx.MockTransport 注入，D11）。"""
|
||||
import json
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from router_system.architect import (
|
||||
ArchitectCircuitBreaker,
|
||||
ArchitectClient,
|
||||
ArchitectError,
|
||||
build_architect,
|
||||
)
|
||||
from router_system.workspace import Workspace
|
||||
|
||||
BRIEF_JSON = json.dumps({
|
||||
"goal": "实现快排",
|
||||
"constraints": ["标准库"],
|
||||
"tags": ["code"],
|
||||
"acceptance": [{"id": "a1", "check": "排序正确", "machine_checkable": True}],
|
||||
"plan": [{"id": "s1", "task": "实现", "deps": [], "done_criteria": "可运行"}],
|
||||
}, ensure_ascii=False)
|
||||
|
||||
DECIDE_JSON = json.dumps({"reply": "改用断言", "patch_plan": [{"id": "s2", "task": "修"}]}, ensure_ascii=False)
|
||||
REVIEW_JSON = json.dumps({"verdict": "done", "notes": "通过", "fix_issues": []}, ensure_ascii=False)
|
||||
|
||||
|
||||
def _make_client(handler, api_key="test-key", **kw):
|
||||
transport = httpx.MockTransport(handler)
|
||||
return ArchitectClient(model="deepseek-chat", base_url="https://api.deepseek.com/v1",
|
||||
api_key=api_key, transport=transport, **kw)
|
||||
|
||||
|
||||
def _resp_json(content, usage=None):
|
||||
return httpx.Response(200, json={
|
||||
"choices": [{"message": {"content": content}}],
|
||||
"usage": usage or {"prompt_tokens": 100, "completion_tokens": 20},
|
||||
})
|
||||
|
||||
|
||||
def _ws(**kw):
|
||||
return Workspace.new(request_id="a1b2c3d4e5f6", query=kw.get("query", "写个快排"),
|
||||
api_token_cap=kw.get("cap", 8000), rounds_cap=kw.get("rounds", 6))
|
||||
|
||||
|
||||
# ---------- brief 成功 ----------
|
||||
def test_brief_success_records_budget():
|
||||
calls = []
|
||||
def handler(request):
|
||||
calls.append(request.url.path)
|
||||
return _resp_json(BRIEF_JSON)
|
||||
client = _make_client(handler)
|
||||
ws = _ws()
|
||||
brief = asyncio_run(client.brief("写个快排", ws))
|
||||
assert brief["goal"] == "实现快排"
|
||||
assert brief["plan"][0]["id"] == "s1"
|
||||
assert calls == ["/v1/chat/completions"]
|
||||
# token 计量回写
|
||||
assert ws.budget()["api_input_tokens"] == 100
|
||||
assert ws.budget()["api_output_tokens"] == 20
|
||||
|
||||
|
||||
# ---------- 缺 key ----------
|
||||
def test_no_key_raises():
|
||||
client = ArchitectClient(model="deepseek-chat", api_key=None)
|
||||
ws = _ws()
|
||||
with pytest.raises(ArchitectError):
|
||||
asyncio_run(client.brief("hi", ws))
|
||||
|
||||
|
||||
# ---------- 坏 JSON 重试一次成功 ----------
|
||||
def test_bad_json_retry_once_success():
|
||||
seq = [{"body": "不是json{{", "ok": False}, {"body": BRIEF_JSON, "ok": True}]
|
||||
calls = []
|
||||
def handler(request):
|
||||
calls.append(1)
|
||||
item = seq[len(calls) - 1]
|
||||
return _resp_json(item["body"])
|
||||
client = _make_client(handler)
|
||||
ws = _ws()
|
||||
brief = asyncio_run(client.brief("hi", ws))
|
||||
assert len(calls) == 2
|
||||
assert brief["goal"] == "实现快排"
|
||||
|
||||
|
||||
# ---------- 坏 JSON 两次失败 ----------
|
||||
def test_bad_json_twice_raises():
|
||||
def handler(request):
|
||||
return _resp_json("垃圾输出{")
|
||||
client = _make_client(handler)
|
||||
ws = _ws()
|
||||
with pytest.raises(ArchitectError):
|
||||
asyncio_run(client.brief("hi", ws))
|
||||
|
||||
|
||||
# ---------- API 错误（非 2xx） ----------
|
||||
def test_http_error_raises():
|
||||
def handler(request):
|
||||
return httpx.Response(500, text="server error")
|
||||
client = _make_client(handler)
|
||||
ws = _ws()
|
||||
with pytest.raises(ArchitectError):
|
||||
asyncio_run(client.brief("hi", ws))
|
||||
|
||||
|
||||
# ---------- 缺 choices ----------
|
||||
def test_missing_choices_raises():
|
||||
def handler(request):
|
||||
return httpx.Response(200, json={"usage": {}})
|
||||
client = _make_client(handler)
|
||||
ws = _ws()
|
||||
with pytest.raises(ArchitectError):
|
||||
asyncio_run(client.brief("hi", ws))
|
||||
|
||||
|
||||
# ---------- 熔断：预算触顶，不再调用 transport ----------
|
||||
def test_circuit_breaker_before_transport():
|
||||
called = []
|
||||
def handler(request):
|
||||
called.append(1)
|
||||
return _resp_json(BRIEF_JSON)
|
||||
client = _make_client(handler)
|
||||
ws = _ws(cap=1)
|
||||
ws.add_budget(input_tokens=1, output_tokens=0) # used=1 >= cap=1
|
||||
with pytest.raises(ArchitectCircuitBreaker):
|
||||
asyncio_run(client.brief("hi", ws))
|
||||
assert called == [] # 未触达 API
|
||||
|
||||
|
||||
# ---------- decide / final_review ----------
|
||||
def test_decide():
|
||||
def handler(request):
|
||||
return _resp_json(DECIDE_JSON)
|
||||
client = _make_client(handler)
|
||||
ws = _ws()
|
||||
ws.apply_brief({"goal": "x", "constraints": [], "tags": ["code"],
|
||||
"acceptance": [], "plan": [{"id": "s1", "task": "t", "deps": [], "done_criteria": "c"}]})
|
||||
ws.add_issue("s1", "a://f.py#L1", "obs", "exp", "try", "ask")
|
||||
out = asyncio_run(client.decide(ws))
|
||||
assert out["reply"] == "改用断言"
|
||||
|
||||
|
||||
def test_final_review_done():
|
||||
def handler(request):
|
||||
return _resp_json(REVIEW_JSON)
|
||||
client = _make_client(handler)
|
||||
ws = _ws()
|
||||
ws.apply_brief({"goal": "x", "constraints": [], "tags": ["code"],
|
||||
"acceptance": [], "plan": [{"id": "s1", "task": "t", "deps": [], "done_criteria": "c"}]})
|
||||
out = asyncio_run(client.final_review(ws))
|
||||
assert out["verdict"] == "done"
|
||||
|
||||
|
||||
# ---------- build_architect 工厂 ----------
|
||||
def test_build_architect_reads_env(monkeypatch):
|
||||
cfg = {"model": "deepseek-chat", "api_key_env": "DEEPSEEK_API_KEY"}
|
||||
client = build_architect(cfg, get_env=lambda name: "sk-fake")
|
||||
assert client.api_key == "sk-fake"
|
||||
|
||||
|
||||
def test_build_architect_no_key():
|
||||
cfg = {"api_key_env": "DEEPSEEK_API_KEY"}
|
||||
client = build_architect(cfg, get_env=lambda name: None)
|
||||
assert client.api_key is None
|
||||
|
||||
|
||||
# ---------- 小工具 ----------
|
||||
def asyncio_run(coro):
|
||||
import asyncio
|
||||
return asyncio.run(coro)
|
||||
+1
-1
@@ -69,7 +69,7 @@ P0 完成后的能力：干净的后端抽象 + 可量化的评测 + 可追溯
|
||||
|---|------|------|--------|
|
||||
| T1 | 环境与基线确认（126 测试全绿；README 环境备忘） | ✅ 完成 | （并入 T2 commit） |
|
||||
| T2 | 运维层：hw_profile + llama_server 进程管理 | ✅ 完成 | T2 |
|
||||
| T3 | ArchitectClient（DeepSeek API，JSON 约束） | ⬜ | |
|
||||
| T3 | ArchitectClient（DeepSeek API，JSON 约束） | ✅ 完成 | T3 |
|
||||
| T4 | Workspace（交流文本 schema/校验/渲染/rollup） | ✅ 完成 | T4 |
|
||||
| T5 | WorkerLoop + 接地验证 | ⬜ | |
|
||||
| T6 | CollaborativePipeline 编排 | ⬜ | |
|
||||
|
||||
Reference in New Issue
Block a user