From c8743821307f13d1b8e13d811ca9ec2e386b8a26 Mon Sep 17 00:00:00 2001 From: tzt <14718231+flying-travel@user.noreply.gitee.com> Date: Sun, 30 Aug 2026 21:15:19 +0800 Subject: [PATCH] =?UTF-8?q?feat(v2):=20T7=20=E7=BD=91=E5=85=B3=E6=89=A9?= =?UTF-8?q?=E5=B1=95=20+=20T9=20token=20=E8=AE=A1=E9=87=8F=E4=B8=8E?= =?UTF-8?q?=E8=B4=A6=E5=8D=95=EF=BC=88/chat=20v2,/chat/legacy,/runs,/revie?= =?UTF-8?q?w,/metrics=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- gateway/api.py | 355 ++++++++++++++++++++++++++++----------- router_system/v2stats.py | 80 +++++++++ router_system/worker.py | 19 ++- tests/test_gateway.py | 44 ++++- tests/test_trace.py | 81 +++++++++ 任务拆解与执行计划.md | 4 +- 6 files changed, 482 insertions(+), 101 deletions(-) create mode 100644 router_system/v2stats.py create mode 100644 tests/test_trace.py diff --git a/gateway/api.py b/gateway/api.py index 38cd2aa..da12dcf 100644 --- a/gateway/api.py +++ b/gateway/api.py @@ -1,94 +1,261 @@ -"""FastAPI 网关：对外提供 /chat /health /metrics 接口。 - -启动： - uvicorn gateway.api:app --host 0.0.0.0 --port 8000 -或： - python -m gateway.api - -依赖：fastapi, uvicorn, pydantic（见 requirements.txt） -""" -from __future__ import annotations - -from typing import List, Optional - -from pydantic import BaseModel, Field - -from router_system.config import load_config -from router_system.router import Router, build_router - -# ---- 全局单例 ---- -_router: Optional[Router] = None - - -def get_router() -> Router: - global _router - if _router is None: - _router = build_router() - return _router - - -# ---- 请求/响应模型 ---- -class QueryRequest(BaseModel): - query: str = Field(..., min_length=1, max_length=8000, description="用户查询") - - -class QueryResponse(BaseModel): - response: str - domain: str - difficulty: str - confidence: float - upgraded: bool - quality_score: float - model_used: str - route: List[str] - latency_ms: float - cache_hit: bool - cache_level: Optional[str] - cost_est: float - error: Optional[str] = None - - -class HealthResponse(BaseModel): - status: str - domains: List[str] - classifier: str - judge: str - fallback: str - - -# ---- FastAPI 应用 ---- -try: - from fastapi import FastAPI - - app = FastAPI( - title="Multi-Expert Router API", - description="多专业小模型 + 路由模型系统（MVP）", - version="0.1.0", - ) - - @app.get("/health", response_model=HealthResponse, tags=["system"]) - async def health(): - return get_router().health() - - @app.post("/chat", response_model=QueryResponse, tags=["chat"]) - async def chat(req: QueryRequest): - result = await get_router().route(req.query) - return QueryResponse(**result.to_dict()) - - @app.get("/metrics", tags=["system"]) - async def metrics(): - r = get_router() - return { - "router": r.stats.summary(), - "cache": r.cache.stats(), - } - -except ImportError: - # fastapi 未安装时，提供 CLI 入口提示 - app = None - print("[gateway] 未安装 fastapi，请执行: pip install -r requirements.txt") - - -if __name__ == "__main__": - import uvicorn - uvicorn.run("gateway.api:app", host="0.0.0.0", port=8000, reload=False) +"""FastAPI 网关：对外提供 v2（端云协同）与 v1（legacy 路由）两套接口。 + +v2 端点（《实现方案_v2》5.3）： + POST /chat 端云协同协作管线 + POST /chat/legacy 原 v1 L0 行为（离线降级，测试封闭） + GET /runs/{request_id}/workspace 交流文本 + GET /runs/{request_id}/artifacts/{name} 工件下载 + GET /review/queue、POST /review/{id} 人工检验 + GET /metrics 含 v2 token/快路径统计 + +启动：uvicorn gateway.api:app --host 0.0.0.0 --port 8000 +""" +from __future__ import annotations + +from pathlib import Path +from typing import List, Optional + +from pydantic import BaseModel, Field + +from router_system.config import load_config +from router_system.router import Router, build_router + +# ---- v2 依赖（惰性导入，缺依赖时降级提示） ---- +try: + from router_system.architect import build_architect + from router_system.knowledge import KnowledgeBase + from router_system.pipeline import CollaborativePipeline, build_pipeline + from router_system.review import ReviewQueue + from router_system.v2stats import V2Stats + from router_system.worker import build_worker + _V2_OK = True +except Exception: # pragma: no cover + _V2_OK = False + +# ---- v1 全局单例 ---- +_router: Optional[Router] = None + +# ---- v2 全局单例 ---- +_pipeline: Optional["CollaborativePipeline"] = None +_v2stats = V2Stats() if _V2_OK else None +_review = None + + +def get_router() -> Router: + global _router + if _router is None: + _router = build_router() + return _router + + +def get_review() -> "ReviewQueue": + global _review + if _review is None and _V2_OK: + cfg = load_config().get("review", {}) + _review = ReviewQueue(db_path=cfg.get("queue_db", "data/review.sqlite3")) + return _review + + +def build_v2_pipeline(worker_cfg_override: Optional[dict] = None): + """从配置构建 v2 协作管线（architect + worker + pipeline）。 + + worker_cfg_override 可注入（测试/演示用 mock）。无 API key 时 /chat 会走 + 本地降级路径（不崩溃）。 + """ + global _pipeline + if _pipeline is None: + cfg = load_config() + kb = KnowledgeBase() + architect = build_architect(cfg.get("architect", {})) + wcfg = dict(cfg.get("worker", {})) + if worker_cfg_override: + wcfg.update(worker_cfg_override) + worker = build_worker(wcfg, kb=kb) + _pipeline = build_pipeline(cfg, architect, worker) + return _pipeline + + +def set_pipeline(pipe) -> None: + """测试注入替身管线。""" + global _pipeline + _pipeline = pipe + + +def get_pipeline(): + global _pipeline + if _pipeline is None: + return build_v2_pipeline() + return _pipeline + + +# ---- 请求/响应模型 ---- +class QueryRequest(BaseModel): + query: str = Field(..., min_length=1, max_length=8000, description="用户查询") + domain_group: Optional[str] = Field( + None, description="大领域组（两级路由第一级）：tech | professional | lifestyle | general；不指定则自动检测" + ) + + +class HealthResponse(BaseModel): + status: str + domains: List[str] + classifier: str + judge: str + fallback: str + + +# ---- FastAPI 应用 ---- +try: + from fastapi import FastAPI, HTTPException + + app = FastAPI( + title="端云协同 LLM 协作系统", + description="大模型(Architect) + 本地小模型(Worker) 通过交流文本协作；v1 保留为 legacy 路由", + version="2.0.0", + ) + + @app.get("/health", response_model=HealthResponse, tags=["system"]) + async def health(): + return get_router().health() + + # ---------------- v2：/chat ---------------- + @app.post("/chat", tags=["chat"]) + async def chat(req: QueryRequest): + try: + result = await get_pipeline().run(req.query) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + # 记账 + 入队 + if _v2stats is not None: + _v2stats.record(result) + _maybe_enqueue(result) + return { + "response": result.response, + "request_id": result.request_id, + "status": result.status, + "fast_path": result.fast_path, + "rounds_used": result.rounds_used, + "api_input_tokens": result.api_input_tokens, + "api_output_tokens": result.api_output_tokens, + "cost_est": result.cost_est, + "model_used": result.model_used, + "latency_ms": round(result.latency_ms, 2), + "route": result.route, + "workspace_path": result.workspace_path, + "error": result.error, + } + + # ---------------- v1：/chat/legacy ---------------- + @app.post("/chat/legacy", tags=["chat"]) + async def chat_legacy(req: QueryRequest): + """原 v1 L0 专家系统行为（离线可跑，测试封闭）。""" + try: + result = await get_router().route(req.query, domain_group=req.domain_group) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + return { + "response": result.response, + "domain": result.domain, + "difficulty": result.difficulty, + "confidence": round(result.confidence, 4), + "upgraded": result.upgraded, + "quality_score": round(result.quality_score, 4), + "model_used": result.model_used, + "route": result.route, + "latency_ms": round(result.latency_ms, 2), + "cache_hit": result.cache_hit, + "cache_level": result.cache_level, + "cost_est": round(result.cost_est, 6), + "error": result.error, + "subdomain": result.subdomain, + "subdomain2": result.subdomain2, + "domain_group": result.domain_group, + "request_id": result.request_id, + } + + @app.get("/traces/{request_id}", tags=["system"]) + async def get_trace(request_id: str): + trace = get_router().trace_store.get(request_id) + if trace is None: + raise HTTPException(status_code=404, detail=f"未找到请求 {request_id} 的推理链") + return trace + + # ---------------- v2：workspace / artifacts ---------------- + @app.get("/runs/{request_id}/workspace", tags=["v2"]) + async def get_workspace(request_id: str): + p = Path("runs") / request_id / "workspace.json" + if not p.exists(): + raise HTTPException(status_code=404, detail=f"未找到运行 {request_id}") + import json + return json.loads(p.read_text(encoding="utf-8")) + + @app.get("/runs/{request_id}/artifacts/{name}", tags=["v2"]) + async def get_artifact(request_id: str, name: str): + p = Path("runs") / request_id / "artifacts" / name + if not p.exists(): + raise HTTPException(status_code=404, detail=f"未找到工件 {name}") + return {"name": name, "content": p.read_text(encoding="utf-8")} + + # ---------------- v2：人工检验 ---------------- + @app.get("/review/queue", tags=["v2"]) + async def review_queue(status: Optional[str] = None, limit: int = 50): + return get_review().list(status=status, limit=limit) + + @app.post("/review/{review_id}", tags=["v2"]) + async def review_submit(review_id: int, verdict: str, correction: Optional[str] = None): + try: + ok = get_review().submit(review_id, verdict, correction=correction) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + if not ok: + raise HTTPException(status_code=404, detail=f"审核记录不存在或已审核: {review_id}") + return {"ok": True, "review_id": review_id, "verdict": verdict} + + # ---------------- metrics ---------------- + @app.get("/metrics", tags=["system"]) + async def metrics(): + out = { + "router": get_router().stats.summary(), + "cache": get_router().cache.stats(), + } + if _V2_OK and _v2stats is not None: + out["v2"] = _v2stats.summary() + out["review"] = { + "pending": get_review().count(status="pending"), + "total": get_review().count(), + } + return out + +except ImportError: + app = None + print("[gateway] 未安装 fastapi，请执行: pip install -r requirements.txt") + + +def _maybe_enqueue(result) -> None: + """按 review 配置抽样/强制入队（异步场景用同步快速调用）。""" + if not _V2_OK or get_review() is None: + return + if result.status not in ("done", "escalated", "fast_path"): + return + cfg = load_config().get("review", {}) + sample_rate = float(cfg.get("sample_rate", 0.10)) + force_tags = cfg.get("force_tags", ["safety"]) + # 从交流文本读取 tags（若可） + tags: List[str] = [] + if result.workspace_path: + try: + import json + ws = json.loads(Path(result.workspace_path).read_text(encoding="utf-8")) + tags = (ws.get("brief") or {}).get("tags") or [] + except Exception: + tags = [] + if ReviewQueue.should_enqueue(tags, sample_rate=sample_rate, force_tags=force_tags): + get_review().enqueue( + request_id=result.request_id, query=result.query, response=result.response, + tags=tags, reason="auto", workspace_path=result.workspace_path, + ) + + +if __name__ == "__main__": + import uvicorn + uvicorn.run("gateway.api:app", host="0.0.0.0", port=8000, reload=False) diff --git a/router_system/v2stats.py b/router_system/v2stats.py new file mode 100644 index 0000000..6044192 --- /dev/null +++ b/router_system/v2stats.py @@ -0,0 +1,80 @@ +"""v2 统计聚合器（V2Stats）—— token 计量与账单（纯标准库）。 + +T9：每请求 API token 记账；聚合快路径命中率、回合数分布、熔断次数、累计 token/成本。 +配合 /metrics 对外透出（论文 E1 token 经济学数据来源之一）。 +""" +from __future__ import annotations + +import threading +from typing import Any, Dict, List, Optional + + +class V2Stats: + """线程安全的 v2 运行统计。""" + + def __init__(self): + self._lock = threading.Lock() + self._total = 0 + self._fast_path = 0 + self._breach = 0 + self._by_status: Dict[str, int] = {} + self._rounds: List[int] = [] + self._api_input_tokens = 0 + self._api_output_tokens = 0 + self._api_cost_usd = 0.0 + self._recent: List[Dict[str, Any]] = [] + + def record(self, result) -> None: + """记录一次 PipelineResult。""" + with self._lock: + self._total += 1 + if getattr(result, "fast_path", False): + self._fast_path += 1 + status = getattr(result, "status", "?") + self._by_status[status] = self._by_status.get(status, 0) + 1 + self._rounds.append(getattr(result, "rounds_used", 0)) + if "breach" in " ".join(getattr(result, "route", [])): + self._breach += 1 + self._api_input_tokens += getattr(result, "api_input_tokens", 0) + self._api_output_tokens += getattr(result, "api_output_tokens", 0) + self._api_cost_usd += getattr(result, "cost_est", 0.0) + self._recent.append({ + "request_id": getattr(result, "request_id", ""), + "status": status, + "fast_path": getattr(result, "fast_path", False), + "api_input_tokens": getattr(result, "api_input_tokens", 0), + "api_output_tokens": getattr(result, "api_output_tokens", 0), + "rounds_used": getattr(result, "rounds_used", 0), + }) + if len(self._recent) > 200: + self._recent = self._recent[-200:] + + def summary(self) -> Dict[str, Any]: + with self._lock: + total = self._total + rounds = self._rounds + return { + "total_requests": total, + "fast_path_rate": round(self._fast_path / total, 4) if total else 0.0, + "status_distribution": dict(self._by_status), + "breach_count": self._breach, + "rounds_used": { + "avg": round(sum(rounds) / len(rounds), 2) if rounds else 0.0, + "max": max(rounds) if rounds else 0, + "distribution": _histogram(rounds), + }, + "api_tokens": { + "input": self._api_input_tokens, + "output": self._api_output_tokens, + "total": self._api_input_tokens + self._api_output_tokens, + "cost_est_usd": round(self._api_cost_usd, 6), + }, + } + + +def _histogram(values: List[int], max_bucket: int = 10) -> Dict[str, int]: + out: Dict[str, int] = {} + for v in values: + key = str(v) if v <= max_bucket else f">{max_bucket}" + out[key] = out.get(key, 0) + 1 + return out diff --git a/router_system/worker.py b/router_system/worker.py index 4179db1..462ecdb 100644 --- a/router_system/worker.py +++ b/router_system/worker.py @@ -154,9 +154,14 @@ class WorkerLoop: def build_worker(cfg: Dict[str, Any], kb: Any = None, generate: Optional[Callable[[str], Awaitable[str]]] = None) -> WorkerLoop: - """cfg 为 config.worker 段。generate 缺省时用 llama-server 端点客户端（惰性）。""" + """cfg 为 config.worker 段。generate 缺省时按 backend 选择： + mock（零运行时演示）| llama_server（真实本地模型，惰性连接）。""" + backend = cfg.get("backend", "llama_server") if generate is None: - generate = _make_llama_generate(cfg) + if backend == "mock": + generate = _mock_generate() + else: + generate = _make_llama_generate(cfg) verifier = Verifier(code_timeout_s=float(cfg.get("code_timeout_s", 10))) return WorkerLoop( generate=generate, @@ -167,6 +172,16 @@ def build_worker(cfg: Dict[str, Any], kb: Any = None, ) +def _mock_generate() -> Callable[[str], Awaitable[str]]: + """零运行时 mock 生成器：返回一段确定性文本（演示/测试，不连真实模型）。""" + + async def _gen(prompt: str) -> str: + return ("（mock worker）以下是对当前步骤的实现说明：" + "步骤已完成，内容足够长且非占位，可供接地验证通过。") + + return _gen + + def _make_llama_generate(cfg: Dict[str, Any]) -> Callable[[str], Awaitable[str]]: """返回调用本地 llama-server（OpenAI 兼容 /v1/chat/completions）的生成器。""" base_url = cfg.get("base_url", f"http://127.0.0.1:{cfg.get('port', 8901)}/v1") diff --git a/tests/test_gateway.py b/tests/test_gateway.py index 39ca51f..c6721bc 100644 --- a/tests/test_gateway.py +++ b/tests/test_gateway.py @@ -1,4 +1,4 @@ -"""???????? fastapi + httpx??""" +"""FastAPI 网关测试：v1 legacy 端点保持 + v2 端点（封闭，注入 mock 管线）。""" import pytest pytest.importorskip("fastapi") @@ -6,6 +6,7 @@ pytest.importorskip("httpx") from fastapi.testclient import TestClient +import gateway.api as ga from gateway.api import app @@ -14,6 +15,14 @@ def client(): return TestClient(app) +@pytest.fixture() +def v2_client(client): + # 用 mock worker 构建真实 v2 管线并注入（无需 API key / 真实模型） + pipe = ga.build_v2_pipeline(worker_cfg_override={"backend": "mock"}) + ga.set_pipeline(pipe) + return client + + def test_health(client): resp = client.get("/health") assert resp.status_code == 200 @@ -22,8 +31,8 @@ def test_health(client): assert "code" in data["domains"] -def test_chat(client): - resp = client.post("/chat", json={"query": "? Python ???????"}) +def test_chat_legacy(client): + resp = client.post("/chat/legacy", json={"query": "用 Python 写一个快速排序函数"}) assert resp.status_code == 200 data = resp.json() assert data["response"] @@ -31,6 +40,17 @@ def test_chat(client): assert "route" in data +def test_chat_v2(v2_client): + resp = v2_client.post("/chat", json={"query": "请介绍快速排序算法"}) + assert resp.status_code == 200 + data = resp.json() + assert data["response"] + assert data["request_id"] + assert "fast_path" in data + assert "route" in data + assert data["status"] in ("fast_path", "done", "escalated", "failed") + + def test_chat_empty_query(client): resp = client.post("/chat", json={"query": ""}) assert resp.status_code == 422 @@ -42,3 +62,21 @@ def test_metrics(client): data = resp.json() assert "router" in data assert "cache" in data + assert "v2" in data + assert "review" in data + + +def test_workspace_not_found(client): + resp = client.get("/runs/nonexistent/workspace") + assert resp.status_code == 404 + + +def test_review_flow(client): + q = ga.get_review() + rid = q.enqueue("req-x", "q", "ans", tags=["safety"], reason="test") + assert q.count() >= 1 + resp = client.get("/review/queue") + assert resp.status_code == 200 + resp2 = client.post(f"/review/{rid}", params={"verdict": "approve"}) + assert resp2.status_code == 200 + assert resp2.json()["ok"] is True diff --git a/tests/test_trace.py b/tests/test_trace.py new file mode 100644 index 0000000..e81666c --- /dev/null +++ b/tests/test_trace.py @@ -0,0 +1,81 @@ +"""推理链轨迹存储 / 查询测试（T3：整体项目部分拆解·先行实现）。""" +import pytest + +from router_system.router import build_router +from router_system.trace import TraceStore + + +def test_trace_store_put_get(): + ts = TraceStore() + ts.put("abc", {"query": "q1", "route": ["a", "b"]}) + t = ts.get("abc") + assert t["query"] == "q1" + assert ts.get("not-exist") is None + + +def test_trace_store_ring_eviction(): + ts = TraceStore(max_entries=3) + for i in range(5): + ts.put(f"id{i}", {"i": i}) + assert ts.size() == 3 + assert ts.get("id0") is None # 最旧被淘汰 + assert ts.get("id4") is not None + + +def test_trace_store_overwrite(): + ts = TraceStore() + ts.put("a", {"v": 1}) + ts.put("a", {"v": 2}) + assert ts.get("a")["v"] == 2 + assert ts.size() == 1 + + +@pytest.mark.asyncio +async def test_router_records_trace(): + r = build_router() + res = await r.route("求解方程 x^2 - 5x + 6 = 0") + assert res.request_id + trace = r.trace_store.get(res.request_id) + assert trace is not None + assert trace["domain"] == "math" + assert trace["domain_group"] == "tech" + assert "plan:multi" in " ".join(trace["route"]) + assert trace["quality_score"] > 0 + + +@pytest.mark.asyncio +async def test_trace_cache_hit_roundtrip(): + r = build_router() + q = "加班费怎么计算" + r1 = await r.route(q) + r2 = await r.route(q) # 缓存命中 + assert r2.cache_hit is True + trace = r.trace_store.get(r2.request_id) + assert trace["cache_hit"] is True + assert trace["cache_level"] == "exact" + + +@pytest.mark.asyncio +async def test_trace_subdomain_fields(): + r = build_router() + res = await r.route("基金定投的收益率怎么计算") + trace = r.trace_store.get(res.request_id) + assert trace["subdomain"] == "investing" + assert trace["subdomain2"] == "investing" + assert trace["domain_group"] == "professional" + + +def test_gateway_trace_endpoint(): + from fastapi.testclient import TestClient + from gateway.api import app + c = TestClient(app, raise_server_exceptions=False) + chat = c.post("/chat/legacy", json={"query": "请用 Python 实现快速排序的迭代版本，并分析其时间与空间复杂度"}) + rid = chat.json().get("request_id") + assert rid + t = c.get(f"/traces/{rid}") + assert t.status_code == 200 + body = t.json() + assert body["domain"] == "code" + assert "subdomain:algorithm" in " ".join(body["route"]) + miss = c.get("/traces/不存在的id") + assert miss.status_code == 404 diff --git a/任务拆解与执行计划.md b/任务拆解与执行计划.md index 1d50d35..cc497cb 100644 --- a/任务拆解与执行计划.md +++ b/任务拆解与执行计划.md @@ -73,9 +73,9 @@ P0 完成后的能力：干净的后端抽象 + 可量化的评测 + 可追溯 | T4 | Workspace（交流文本 schema/校验/渲染/rollup） | ✅ 完成 | T4 | | T5 | WorkerLoop + 接地验证 | ✅ 完成 | T5 | | T6 | CollaborativePipeline 编排 | ✅ 完成 | T6 | -| T7 | 网关扩展（/chat 切 v2，/chat/legacy） | ⬜ | | +| T7 | 网关扩展（/chat 切 v2，/chat/legacy） | ✅ 完成 | T7 | | T8 | 人工检验队列 ReviewQueue | ✅ 完成 | T8 | -| T9 | token 计量与账单 | ⬜ | | +| T9 | token 计量与账单 | ✅ 完成 | T9 | | T10 | rollup + prefix cache 调优 | ⬜ | | | T11 | 打包分发 setup_runtime.py | ⬜ | | | T12 | 实验脚本 bench_tokens.py + 数据集 | ⬜ | |