diff --git a/contracts/trace/README.md b/contracts/trace/README.md index 32a64c7e..69e67dfa 100644 --- a/contracts/trace/README.md +++ b/contracts/trace/README.md @@ -18,6 +18,19 @@ SAA 那一轨的扩展段由 Java 线另一个 adapter 产,字段与本契约的 消费链路按 H 族图说 H2 是「Event System → TracingMiddleware → OpenTelemetry span → AgentScope Studio」。trace 记录最终落进已部署的 MySQL 加对象存储,不上重型可观测中间件;观测要早建,是为了 spike 调试和成本对账当下就用得上,而不是要先铺一套独立基建。`traceId` 贯穿整条生成任务链路,既是反查键,也是成本关联键——它对接 `tier2-verdict.schema.json` 的 `evidence.traceId`,让一次验收终判能反查到它对应的全过程轨迹。成本侧由 `cost.py` 读 new-api 的 `logs.quota` 权威口径,按 `traceId` 关联后把人民币金额 best-effort 回填进 `cost.cost_rmb`(trace 阶段只记 token,不携金额估算)。 +## 两份 schema 的关系 + +本目录维护两份 trace schema,对应两条异构生成线: + +| schema 文件 | 对应生成线 | 特有扩展段 | +|---|---|---| +| `tier2-trace-event.schema.json` | tier2 富游戏自治线(Python / AgentScope ReAct) | `ext`:reasoning / action / observation 三段(ReAct 推理→动作→观察) | +| `saa-trace-event.schema.json` | SAA 16 节点廉价生成线(Java / Spring AI Alibaba) | `ext`:nodeIndex / nodeName / nodeStatus(节点位置与执行状态) | + +核心五字段(`traceId` / `step` / `cost` / `verdict` / `timestamp`)在两份 schema 里**完全同名同义**——这是 H1「接口对称」子集,让两条线写进同一张 trace 表时无需 join 也能按 `traceId` 对账。`ext` 扩展段各写各的、字段不要求对齐(H1「内容不对称」):tier2 的推理/动作/观察三段 SAA 轨不存在,SAA 的节点编号/名称/状态 tier2 轨也不存在,**没有的字段绝不编造**。 + +谁消费哪份:目前 tier2 侧 adapter(`tier2/gen-worker/observability/trace.py` 的 `TraceAdapter`)已产出并落盘(`JsonlFileSink` 写进 `_tier2-gen//trace.jsonl`);SAA 侧 adapter 在 Java 后端待产线接线(控制面 phase-1)。 + ## 与 H 族图说的对应 设计权威是 `docs/architecture/架构/生成引擎/tier2细节图说-H-观测与成本.md`。 diff --git a/contracts/trace/saa-trace-event.schema.json b/contracts/trace/saa-trace-event.schema.json new file mode 100644 index 00000000..df16bc22 --- /dev/null +++ b/contracts/trace/saa-trace-event.schema.json @@ -0,0 +1,88 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://wanxiang.ai/contracts/trace/saa-trace-event.schema.json", + "title": "SaaTraceEvent", + "description": "SAA(Spring AI Alibaba)16 节点廉价生成线的统一 trace 事件契约(H 族 · A2)。【这是什么】一条 trace 记录 = 一次生成轨迹里的一步,由 SAA Java adapter 产出。核心五字段(traceId / step / cost / verdict / timestamp)与 tier2-trace-event.schema.json 完全同名同义(H1 接口对称子集);ext 扩展段是 SAA 这一轨独有的,记录当前节点在 16 节点图上的位置与执行状态(H1 内容不对称:各轨各写各的,没有的字段绝不编造)。【为什么独立成约】两条异构生成线写进同一张 trace 表,靠的是「接口对称、内容不对称」。本 schema 定义 SAA 轨写进统一表的记录形状;tier2 那一轨的扩展段在 tier2-trace-event.schema.json 里,字段与本 schema 的 ext 不同名、也不要求对齐。【谁产】SAA Java 线的 trace adapter(game-cloud/game-module-aigc 或 studio 模块,待接线);本 schema 是字段口径约束,不是采集机制。【状态】spec 已立,SAA Java adapter 待产线接线(控制面 phase-1)。详见 contracts/trace/README.md 与 docs/architecture/架构/生成引擎/tier2细节图说-H-观测与成本.md 图 H1。", + "type": "object", + "required": ["traceId", "step", "timestamp", "ext"], + "additionalProperties": false, + "properties": { + "traceId": { + "description": "公共核心子集字段(两条生成线同名同义必填)。一次生成的轨迹主键,贯穿整条生成任务链路;是对账 / 成本关联键,对接 tier2-verdict.schema.json 的 evidence.traceId。SAA 侧由 Java adapter 在构造时钉定,本次轨迹所有 step 共用同一个 traceId。", + "type": "string", + "minLength": 1 + }, + "step": { + "description": "公共核心子集字段。第几步(SAA 轨 = 第几个执行节点,0 起;对应 ext.nodeIndex。tier2 轨语义为 ReAct 第几轮 reply / 工具调用步,两者语义不同但字段名对称)。", + "type": "integer", + "minimum": 0 + }, + "cost": { + "description": "公共核心子集字段。这一步折成人民币的成本载体;仅在 SAA 节点真正发起模型调用时非空,其余节点步为 null。cost_rmb 由 cost.py 按 new-api quota 权威口径事后 best-effort 回填(成本权威源是 new-api 计费行,trace 不携金额估算)。", + "type": ["object", "null"], + "$ref": "#/$defs/cost" + }, + "verdict": { + "description": "公共核心子集字段。这一步 / 这道门的裁决(可空)。SAA 轨当前在节点执行终态时填入(success / error / skipped),与 ext.nodeStatus 语义一致;门级裁决由裁决引擎 judge 另行回填,不在 adapter 映射阶段产生。", + "type": ["string", "null"] + }, + "timestamp": { + "description": "公共核心子集字段。本步节点执行的发生时刻(ISO8601 字符串)。节点缺时间戳时为兜底空串。", + "type": "string" + }, + "ext": { + "description": "SAA 扩展段(不对称段,各轨各写各的;tier2 轨的扩展段字段与此不同名也不要求对齐)。记录当前节点在 SAA 16 节点图上的位置(nodeIndex / nodeName)与执行状态(nodeStatus)。additionalProperties:false 锁死段名,防止 adapter 之外注入未约定字段。", + "$ref": "#/$defs/ext" + } + }, + "$defs": { + "cost": { + "type": ["object", "null"], + "description": "成本载体(节点发起模型调用时非空)。tokens 由 adapter 实填(从模型调用 response 抓 input_tokens / output_tokens);cost_rmb 由 cost.py 按 new-api quota 事后回填(权威口径,trace 不携金额估算)。", + "additionalProperties": false, + "required": ["tokens"], + "properties": { + "tokens": { + "description": "本次模型调用的 token 用量。in / out 由 adapter 实填;cached 预留缓存命中 token(当前 adapter 未填)。", + "type": "object", + "additionalProperties": false, + "required": ["in", "out"], + "properties": { + "in": { "description": "输入 token 数。", "type": "integer", "minimum": 0 }, + "out": { "description": "输出 token 数。", "type": "integer", "minimum": 0 }, + "cached": { "description": "缓存命中 token 数(可选,预留;当前 adapter 未实填)。", "type": "integer", "minimum": 0 } + } + }, + "cost_rmb": { + "description": "本步折算人民币(可选)。adapter 不填,由 cost.py 读 new-api logs.quota 按 traceId 关联后 best-effort 回填。", + "type": "number", + "minimum": 0 + } + } + }, + "ext": { + "type": "object", + "description": "SAA 扩展段对象。三个必填字段描述当前节点在 16 节点图上的位置与执行状态。additionalProperties:false 锁死字段集。", + "additionalProperties": false, + "required": ["nodeIndex", "nodeName", "nodeStatus"], + "properties": { + "nodeIndex": { + "description": "当前节点在 SAA 16 节点执行图中的位置(0 起整数;与顶层 step 同义但独立落盘,便于按节点序号切片分析)。", + "type": "integer", + "minimum": 0, + "maximum": 15 + }, + "nodeName": { + "description": "SAA 节点的可读名称(如 game_concept_generator / code_writer / asset_patcher 等;由 Java adapter 按 SAANode.name 填入,便于在 trace 里直接定位是哪个节点步)。", + "type": "string", + "minLength": 1 + }, + "nodeStatus": { + "description": "节点执行状态(终态才填;running 表示节点仍在执行中——仅在 adapter 需要记录中间态时使用)。该值同时被填进顶层 verdict 作这一步裁决。", + "type": "string", + "enum": ["pending", "running", "success", "error", "skipped"] + } + } + } + } +} diff --git a/tier2/gen-worker/observability/trace.py b/tier2/gen-worker/observability/trace.py index 9d538801..e9c9a1e1 100644 --- a/tier2/gen-worker/observability/trace.py +++ b/tier2/gen-worker/observability/trace.py @@ -28,6 +28,8 @@ contracts/trace 现行边界:统一 trace 契约的正式 schema 落处 = contra from __future__ import annotations +import json +from pathlib import Path from typing import Any, Callable, Optional @@ -377,6 +379,74 @@ class TraceAdapter: } +class JsonlFileSink: + """把统一 trace 事件(TraceStep dict)追加写到 JSONL 文件的落盘 sink。 + + 用法: + sink = JsonlFileSink(workdir / "trace.jsonl") + adapter = TraceAdapter(trace_id="...", sink=sink) + + 为什么要它: + - 五字段 schema(traceId / step / cost / verdict / timestamp)原样落盘,供成本对账、replay 用; + - 每条事件是独立 JSON 行(JSONL),追加写、无需整文件加锁,适合长跑生成; + - 落盘路径 = 生成产物 workdir / "trace.jsonl",与 game_id 一一对应,随产物一起留存。 + + best-effort 铁律(与 TraceAdapter 一致): + - 目录不存在则自动创建; + - 写失败只 print 告警,**绝不抛异常**、绝不阻断主生成流程(trace 是旁路观测)。 + """ + + def __init__(self, output_path: "Path | str") -> None: + """ + Args: + output_path: trace.jsonl 落盘路径(全路径);父目录不存在时在首次写前自动创建。 + """ + self._path = Path(output_path) + # 预建父目录(best-effort;目录已存在不报错)。 + try: + self._path.parent.mkdir(parents=True, exist_ok=True) + except Exception as exc: + # 预建失败(如路径不可写)只告警,每次 __call__ 时再尝试一次。 + print( + f"[tier2-trace] 预建 trace.jsonl 目录失败(best-effort,写时再试):" + f"{self._path.parent} — {exc}", + flush=True, + ) + + def __call__(self, trace_step: dict) -> None: + """把一条 TraceStep dict 序列化为 JSON 追加写一行到 trace.jsonl。 + + 五字段(traceId / step / cost / verdict / timestamp)由 TraceAdapter.ingest 已填进 + trace_step(TraceStep 是 dict 子类),此处原样序列化落盘,保留 ext 扩展段。 + + 写失败(磁盘满 / 权限错等)只 print 告警,绝不抛 —— 一次落库失败不得废掉已成功的生成产物。 + """ + try: + # 防御性再建一次目录(首次 __init__ 预建可能因时序失败)。 + self._path.parent.mkdir(parents=True, exist_ok=True) + line = json.dumps(trace_step, ensure_ascii=False) + with self._path.open("a", encoding="utf-8") as f: + f.write(line + "\n") + except Exception as exc: + print( + f"[tier2-trace] trace.jsonl 写失败(best-effort,不阻断生成,path={self._path}):{exc}", + flush=True, + ) + + +def make_jsonl_sink(output_path: "Path | str") -> "JsonlFileSink": + """构造 JSONL 落盘 sink,目标路径 output_path;可直接塞进 TraceAdapter(sink=...)。 + + 用法示例(在编排器 run_studio 里): + sink = make_jsonl_sink(run._workdir(game_id) / "trace.jsonl") + tracer = Tier2TraceMiddleware(trace_id=game_id, sink=sink) + + :param output_path: trace.jsonl 落盘路径(全路径)。 + :return: JsonlFileSink 实例(Callable[[dict], None])。 + """ + return JsonlFileSink(output_path) + + def with_studio( trace_id: str, *, diff --git a/tier2/gen-worker/service/admin_routes.py b/tier2/gen-worker/service/admin_routes.py new file mode 100644 index 00000000..3dd38c3b --- /dev/null +++ b/tier2/gen-worker/service/admin_routes.py @@ -0,0 +1,272 @@ +"""service/admin_routes.py —— tier2 管理面只读 API(phase-1 · U-C3a)。 + +三个只读 GET 端点,供运营回看生成配置与轨迹: + GET /control/roles —— 各生成角色当前用的模型(从 worker/config.py 常量读)。 + GET /control/traces/{id} —— 读 run._workdir(traceId)/trace.jsonl 返回全部步骤。 + GET /control/cost/{id} —— 对 trace.jsonl 做 new-api quota 口径成本对账。 + +设计决策(附中文注释说明): + ① 全部 GET 只读,不触发任何生成、不写文件、不起真服务; + ② worker/config.py 顶层 import agentscope(mini-desktop 上有;6c6g/测试环境无), + 本模块对 config 一律惰性 import + try/except,确保 6c6g py_compile / import 不炸; + ③ /control/traces/{id} 文件不存在返 404(REST 语义:资源不存在即 404),响应体带 found=false + 方便客户端分支一致处理,优于返回 200+found=false(后者歧义:200 暗示"有了"); + ④ /control/cost/{id} 取价不可达 → pricingSource="degraded" + 200(不抛 500), + 对齐 best-effort 铁律(成本观测不得中断主链); + ⑤ _workdir_for_trace 与 run._workdir 同口径独立实现(无需 import run),便于轻量测试打桩。 +""" + +from __future__ import annotations + +import json +import os +from pathlib import Path + +from fastapi import APIRouter +from fastapi.responses import JSONResponse + +# 路由前缀 /control;标签 admin 供 OpenAPI docs 分组展示。 +router = APIRouter(prefix="/control", tags=["admin"]) + + +# ───────────────────────────────────────────────────────────────────────────── +# 辅助:角色 → 模型映射 +# ───────────────────────────────────────────────────────────────────────────── + +def _get_roles() -> list[dict]: + """读各生成角色当前配的模型名。 + + 优先路:import worker.config 取真实编译常量(只在 mini-desktop 上成功,那里装了 agentscope)。 + 回落路:worker.config 因缺 agentscope 而 import 失败时,直接读 worker.genconfig(无 agentscope 依赖)。 + 两条路产出相同形状:[{"role": ..., "model": ...}, ...]。 + 角色说明: + writer —— tier2 单写 ReAct agent(默认 MiniMax-M3;env TIER2_MODEL 可临时压)。 + cheap_flash —— 便宜档主力(deepseek-v4-flash,经 new-api OpenAI 兼容路)。 + cheap_pro —— 便宜档强档(deepseek-v4-pro,退路树 Q1 用)。 + baseline —— 跨族对照(当前 deepseek-v4-pro;网关开通 Opus/Fable 后改 generation.yaml 切换)。 + """ + try: + # 优先路:worker.config 顶层 import agentscope(mini-desktop 真跑时走这条)。 + from worker import config as _cfg # noqa: PLC0415 — 惰性 import + return [ + {"role": "writer", "model": _cfg.model_name_from_env()}, + {"role": "cheap_flash", "model": _cfg.DEEPSEEK_FLASH}, + {"role": "cheap_pro", "model": _cfg.DEEPSEEK_PRO}, + {"role": "baseline", "model": _cfg.BASELINE_CROSSCHECK}, + ] + except Exception: + # 回落路:agentscope 未装(6c6g / 测试环境)→ genconfig 直读,无 agentscope 依赖。 + from worker import genconfig as _gc # noqa: PLC0415 + return [ + {"role": "writer", + "model": os.environ.get("TIER2_MODEL", + _gc.get("model", "default_model_name", "MiniMax-M3"))}, + {"role": "cheap_flash", + "model": _gc.get("model", "deepseek_flash", "deepseek-v4-flash")}, + {"role": "cheap_pro", + "model": _gc.get("model", "deepseek_pro", "deepseek-v4-pro")}, + {"role": "baseline", + "model": _gc.get("model", "baseline_crosscheck", "deepseek-v4-pro")}, + ] + + +# ───────────────────────────────────────────────────────────────────────────── +# 辅助:workdir 路径计算(与 run._workdir 同口径,无需 import run) +# ───────────────────────────────────────────────────────────────────────────── + +def _workdir_for_trace(trace_id: str) -> Path: + """计算 trace_id 对应的 tier2 工程目录,口径与 worker/run._workdir 完全一致。 + + 为何独立实现而不直接 import worker.run: + run.py 顶层 import archetypes(agentscope 链),保持本模块尽量轻量(6c6g / 测试环境友好); + 路径计算是纯 pathlib 操作,独立复制最小逻辑更清晰。 + 路径: + 本文件在 tier2/gen-worker/service/admin_routes.py → 上溯 4 级到仓根: + service/ → gen-worker/ → tier2/ → 仓根(games-development-ai) + GEN_DIR = 仓根/game-runtime/games/_tier2-gen(tier2 专用生成目录)。 + 此函数是测试桩点:测试用 unittest.mock.patch 替换它指向临时目录即可,无需真实工程目录存在。 + """ + repo_root = Path(__file__).resolve().parents[3] + gen_dir = repo_root / "game-runtime" / "games" / "_tier2-gen" + return gen_dir / trace_id + + +# ───────────────────────────────────────────────────────────────────────────── +# 辅助:从 trace 步骤提取 tokens_by_model +# ───────────────────────────────────────────────────────────────────────────── + +def _extract_tokens_by_model(steps: list[dict]) -> dict[str, dict[str, int]]: + """从 trace 步骤列表提取 tokens_by_model 聚合(按模型分 in/out/cached token 累加)。 + + 关联规则(与 to_trace_step 映射规则一致): + - ext.raw.model_name 非空 → 更新「当前模型名」(对应 ModelCallStartEvent)。 + - cost.tokens.in / out 非零 → 记入当前模型的 bucket(对应 ModelCallEndEvent)。 + 跨多轮模型调用:同一模型的 token 累加;未见模型名时用 "unknown" 兜底。 + + :param steps: trace.jsonl 解析后的步骤列表。 + :return: {model_name: {"in": Σin, "out": Σout, "cached": Σcached}}。 + """ + tokens_by_model: dict[str, dict[str, int]] = {} + current_model = "unknown" # 每遇 ModelCallStart 类事件更新 + for step in steps: + # 更新当前模型名:从 ext.raw.model_name 读(ModelCallStartEvent 标记)。 + ext = step.get("ext") or {} + raw = ext.get("raw") or {} + model_name_in_step = raw.get("model_name") + if model_name_in_step: + current_model = str(model_name_in_step) + # 读 token 计数:cost.tokens.in / out(ModelCallEndEvent / Tier2ModelCallPhaseEvent 填)。 + cost = step.get("cost") or {} + tokens = cost.get("tokens") if isinstance(cost, dict) else None + if not isinstance(tokens, dict): + continue + in_tok = int(tokens.get("in") or 0) + out_tok = int(tokens.get("out") or 0) + # cached 字段在 trace 当前格式里一般不出现(由 RecordingChatModel 记录,非 trace 字段), + # 预留:若将来 trace 扩展落 tokens.cached 可直接读到。 + cached_tok = int(tokens.get("cached") or 0) + if in_tok == 0 and out_tok == 0: + continue # 零 token 步骤跳过(非 ModelCallEnd) + bucket = tokens_by_model.setdefault(current_model, {"in": 0, "out": 0, "cached": 0}) + bucket["in"] += in_tok + bucket["out"] += out_tok + bucket["cached"] += cached_tok + return tokens_by_model + + +# ───────────────────────────────────────────────────────────────────────────── +# 端点一:GET /control/roles +# ───────────────────────────────────────────────────────────────────────────── + +@router.get("/roles") +def get_roles() -> dict: + """GET /control/roles —— 各生成角色当前配的模型(只读,无副作用)。 + + 数据源:worker/config.py 编译常量(role→model 静态映射);agentscope 不可达时回落 genconfig。 + 返回:{"roles": [{"role": "writer", "model": "MiniMax-M3"}, ...]}。 + """ + return {"roles": _get_roles()} + + +# ───────────────────────────────────────────────────────────────────────────── +# 端点二:GET /control/traces/{trace_id} +# ───────────────────────────────────────────────────────────────────────────── + +@router.get("/traces/{trace_id}") +def get_traces(trace_id: str): + """GET /control/traces/{traceId} —— 读 trace.jsonl 返回全部步骤。 + + 数据源:run._workdir(traceId)/trace.jsonl(JsonlFileSink 写的 JSONL,五字段: + traceId / step / cost / verdict / timestamp + ext 扩展段)。 + + 返回(200):{"traceId":..., "steps":[...], "count":n, "found":true}。 + 文件不存在(404):{"traceId":..., "steps":[], "count":0, "found":false}。 + + REST 选型理由(404 而非 200+found=false): + 资源不存在 = 404 是 HTTP 语义约定,让客户端无需解析 found 字段即可判断; + 响应体仍带 found=false + 空 steps 以方便调用方分支一致解析。 + + 健壮性:坏行跳过不崩(best-effort,对齐 JsonlFileSink 的 best-effort 铁律)。 + """ + workdir = _workdir_for_trace(trace_id) + trace_file = workdir / "trace.jsonl" + if not trace_file.exists(): + # 资源不存在 → 404;带 found=false 供调用方分支处理。 + return JSONResponse( + status_code=404, + content={"traceId": trace_id, "steps": [], "count": 0, "found": False}, + ) + steps: list[dict] = [] + for raw_line in trace_file.read_text(encoding="utf-8").splitlines(): + line = raw_line.strip() + if not line: + continue + try: + steps.append(json.loads(line)) + except Exception: + # 坏行跳过:不崩、不计入 count(best-effort;trace 写失败不得影响读)。 + pass + return {"traceId": trace_id, "steps": steps, "count": len(steps), "found": True} + + +# ───────────────────────────────────────────────────────────────────────────── +# 端点三:GET /control/cost/{trace_id} +# ───────────────────────────────────────────────────────────────────────────── + +@router.get("/cost/{trace_id}") +def get_cost(trace_id: str): + """GET /control/cost/{traceId} —— 对 trace.jsonl 做 new-api quota 口径成本对账。 + + 数据源: + ① 读 trace.jsonl → 提取 tokens_by_model(按模型分 in/out token 累加)。 + ② 调 newapi_pricing.fetch_pricing_params() 活取计费三件套(pricing/qpu/usd_rate)。 + ③ cost_for_run(tokens, pricing, qpu, usd_rate) 按 new-api quota 公式折¥。 + + 返回(200): + {"traceId":..., "costRmb":float, "byModel":{model:{quota,usd,rmb,...}}, "pricingSource":"live"} + + 降级处理(best-effort 铁律,成本观测不得阻断): + - trace 不存在 → 404(与 /traces 同语义)。 + - 无 token 记录 → pricingSource="degraded", costRmb=0.0, byModel={} 返回 200。 + - 取价不可达(网络/6c6g) → pricingSource="degraded", costRmb=0.0, byModel={} 返回 200。 + - cost 模块 import 失败 → 同上 degraded 路径(防御)。 + """ + workdir = _workdir_for_trace(trace_id) + trace_file = workdir / "trace.jsonl" + if not trace_file.exists(): + # trace 文件不存在 → 404(与 /traces 端点语义一致)。 + return JSONResponse( + status_code=404, + content={"traceId": trace_id, "costRmb": 0.0, "byModel": {}, + "pricingSource": "notfound"}, + ) + + # ① 读 trace.jsonl 并解析步骤(坏行跳过,best-effort)。 + steps: list[dict] = [] + for raw_line in trace_file.read_text(encoding="utf-8").splitlines(): + line = raw_line.strip() + if not line: + continue + try: + steps.append(json.loads(line)) + except Exception: + pass # 坏行跳过 + + # ② 提取 tokens_by_model。 + tokens_by_model = _extract_tokens_by_model(steps) + if not tokens_by_model: + # 无 token 记录(trace 全是非 ModelCallEnd 步骤,或还没生成过)→ degraded。 + print(f"[tier2-admin] GET /cost/{trace_id}: trace 无 token 记录,返回 degraded。", + flush=True) + return {"traceId": trace_id, "costRmb": 0.0, "byModel": {}, "pricingSource": "degraded"} + + # ③ 惰性 import cost 模块(observability 不依赖 agentscope,但防御性兜住任何 import 错误)。 + try: + from observability.newapi_pricing import fetch_pricing_params # noqa: PLC0415 + from observability.cost import cost_for_run # noqa: PLC0415 + except Exception as exc: + # 极端环境下 import 失败(如依赖缺失)→ degraded,不抛 500。 + print(f"[tier2-admin] GET /cost/{trace_id}: 导入 cost/pricing 模块失败(degraded):{exc}", + flush=True) + return {"traceId": trace_id, "costRmb": 0.0, "byModel": {}, "pricingSource": "degraded"} + + # ④ 活取 new-api 计费参数(best-effort:取不到 → degraded,不抛)。 + pricing_params = fetch_pricing_params() # 内部 best-effort,失败返 None + if pricing_params is None: + # 取价不可达(6c6g / 网关下线 / httpx 超时)→ degraded 标识,返 200 不抛 500。 + print(f"[tier2-admin] GET /cost/{trace_id}: new-api 取价不可达,返回 degraded。", flush=True) + return {"traceId": trace_id, "costRmb": 0.0, "byModel": {}, "pricingSource": "degraded"} + + # ⑤ 按 new-api quota 公式折¥(cost_for_run 纯函数,全显式传参,不读全局)。 + cost_result = cost_for_run( + tokens_by_model, + pricing_params["pricing"], + pricing_params["qpu"], + pricing_params["usd_rate"], + ) + return { + "traceId": trace_id, + "costRmb": cost_result["cost_rmb"], + "byModel": cost_result.get("by_model", {}), + "pricingSource": "live", + } diff --git a/tier2/gen-worker/service/app.py b/tier2/gen-worker/service/app.py index f7f115c8..6e3a4559 100644 --- a/tier2/gen-worker/service/app.py +++ b/tier2/gen-worker/service/app.py @@ -238,7 +238,7 @@ def build_app(*, title: str = SERVICE_TITLE) -> "FastAPI": flush=True, ) - return create_app( + app = create_app( storage=RedisStorage(**storage_params), message_bus=RedisMessageBus(**bus_params), workspace_manager=LocalWorkspaceManager(basedir=ws_basedir), @@ -252,6 +252,14 @@ def build_app(*, title: str = SERVICE_TITLE) -> "FastAPI": title=title, ) + # ── 挂管理面只读路由(phase-1 U-C3a;GET /control/roles · /control/traces/{id} · /control/cost/{id})── + # 惰性 import 红线:admin_routes 顶层 import fastapi。create_app 既已建出 FastAPI app, + # fastapi 在能调 build_app 的环境(mini-desktop)必然在场;在 build_app 体内 import,保证 6c6g + # 仅 `import service.app`(不调 build_app)时不牵 fastapi、不破坏顶层惰性红线。 + from .admin_routes import router as admin_router # noqa: PLC0415 —— 惰性 import 红线 + app.include_router(admin_router) + return app + def main() -> None: """部署入口:起 uvicorn 跑 tier2 Agent Service(只在 mini-desktop 装好 agentscope[full]+redis 后跑)。 diff --git a/tier2/gen-worker/tests/test_admin_routes.py b/tier2/gen-worker/tests/test_admin_routes.py new file mode 100644 index 00000000..29452782 --- /dev/null +++ b/tier2/gen-worker/tests/test_admin_routes.py @@ -0,0 +1,179 @@ +"""test_admin_routes.py —— U-C3a 管理面只读 API 单测(TDD:先红后绿)。 + +用 FastAPI TestClient(进程内,**不起真 uvicorn 服务**、不连 Redis/agentscope、不跑 gen)验三个只读端点: + GET /control/roles —— 200 + 各角色模型(至少含 writer)。 + GET /control/traces/{id} —— 临时样本 trace.jsonl 打桩 _workdir → 200 + steps 数对 + 五字段; + 坏行 best-effort 跳过;不存在 id → 404 + found:false。 + GET /control/cost/{id} —— 样本 trace + 打桩取价:live 含 costRmb;取价不可达 → degraded 标识,不崩。 + +跑法(借 cheap-worker venv 解释器,绝不 import/读写 cheap-worker 源码): + PYTHONPATH=tier2/gen-worker /bin/python -m pytest tier2/gen-worker/tests/test_admin_routes.py -v +""" + +from __future__ import annotations + +import json +import os +import sys +from pathlib import Path +from unittest import mock + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +# 包内/直跑兼容:把 gen-worker/ 加进 sys.path,使 service / observability / worker 可 import +# (PYTHONPATH 已指 gen-worker 时这步是冗余保险;直接 python 跑本文件也能解析包)。 +_HERE = os.path.dirname(os.path.abspath(__file__)) +_GW = os.path.dirname(_HERE) # gen-worker/ +if _GW not in sys.path: + sys.path.insert(0, _GW) + +from service.admin_routes import router # noqa: E402 —— 被测路由 + + +# ── 合法样本 trace.jsonl 内容(五字段 + ext)────────────────────────────────────── +# 覆盖:ReplyStart(reply 边界)/ ModelCallStart(带 model_name,供成本按模型归集)/ +# ModelCallEnd(带 token,供成本折算)/ ToolResultEnd(verdict=success)。 +_SAMPLE_TRACE_LINES = [ + {"traceId": "t-sample", "step": 0, "cost": None, "verdict": None, + "timestamp": "2026-06-27T10:00:00", + "ext": {"raw": {"event": "ReplyStartEvent", "agent_name": "writer"}}}, + {"traceId": "t-sample", "step": 1, "cost": None, "verdict": None, + "timestamp": "2026-06-27T10:00:01", + "ext": {"raw": {"event": "ModelCallStartEvent", "model_name": "MiniMax-M3"}}}, + {"traceId": "t-sample", "step": 2, "cost": {"tokens": {"in": 1200, "out": 800}}, "verdict": None, + "timestamp": "2026-06-27T10:00:02", + "ext": {"raw": {"event": "ModelCallEndEvent", "model_call_end": True}}}, + {"traceId": "t-sample", "step": 3, "cost": None, "verdict": "success", + "timestamp": "2026-06-27T10:00:03", + "ext": {"observation": {"phase": "observation", "state": "success"}}}, +] + + +@pytest.fixture +def client() -> TestClient: + """最小 FastAPI app,只挂被测 admin router(不调 build_app、不连 Redis/agentscope)。""" + app = FastAPI() + app.include_router(router) + return TestClient(app) + + +def _write_sample_trace(tmp_path: Path) -> Path: + """在临时目录写样本工程 workdir + trace.jsonl(末尾附一坏行,验 best-effort 跳过)。 + + 返回样本 workdir(tmp_path/'t-sample'),其下 trace.jsonl = 五字段合法行 + 1 坏行。 + """ + workdir = tmp_path / "t-sample" + workdir.mkdir(parents=True, exist_ok=True) + lines = [json.dumps(o, ensure_ascii=False) for o in _SAMPLE_TRACE_LINES] + lines.append("{ 这是一行坏 JSON —— best-effort 应跳过、不计入 count ") # 坏行 + (workdir / "trace.jsonl").write_text("\n".join(lines) + "\n", encoding="utf-8") + return workdir + + +# ════════════════════════════════════════════════════════════════════════════ +# 端点一:GET /control/roles +# ════════════════════════════════════════════════════════════════════════════ +def test_roles_returns_writer(client): + """GET /control/roles → 200 + roles 列表含 writer 角色且其 model 非空。""" + resp = client.get("/control/roles") + assert resp.status_code == 200, resp.text + data = resp.json() + assert "roles" in data, f"响应缺 roles 字段:{data}" + roles = {r["role"]: r["model"] for r in data["roles"]} + assert "writer" in roles, f"roles 应含 writer:{roles}" + assert roles["writer"], f"writer 的 model 不应为空:{roles}" + print(f" [ok] roles={roles}") + + +# ════════════════════════════════════════════════════════════════════════════ +# 端点二:GET /control/traces/{id} +# ════════════════════════════════════════════════════════════════════════════ +def test_traces_found_with_sample(client, tmp_path): + """样本 trace.jsonl 打桩 _workdir → 200 + steps 数对(坏行跳过)+ 五字段在。""" + _write_sample_trace(tmp_path) + # 打桩:_workdir_for_trace(tid) → tmp_path/tid(端点据此读 /trace.jsonl)。 + with mock.patch("service.admin_routes._workdir_for_trace", + side_effect=lambda tid: tmp_path / tid): + resp = client.get("/control/traces/t-sample") + assert resp.status_code == 200, resp.text + data = resp.json() + assert data["found"] is True, f"found 应为 True:{data}" + # 坏行被跳过 → count == 合法样本行数。 + assert data["count"] == len(_SAMPLE_TRACE_LINES), \ + f"steps 数应为 {len(_SAMPLE_TRACE_LINES)}(坏行跳过),实际 {data['count']}" + assert len(data["steps"]) == data["count"] + # 五字段齐(traceId/step/cost/verdict/timestamp)。 + core = {"traceId", "step", "cost", "verdict", "timestamp"} + for s in data["steps"]: + assert core <= set(s.keys()), f"步骤缺五字段,缺 {core - set(s.keys())}:{s}" + # 抽验内容:第 2 步 token、第 3 步 verdict。 + assert data["steps"][2]["cost"]["tokens"]["in"] == 1200 + assert data["steps"][3]["verdict"] == "success" + print(f" [ok] count={data['count']}(坏行已跳过),五字段齐,token/verdict 对") + + +def test_traces_not_found_returns_404(client, tmp_path): + """不存在的 traceId → 404 + found:false + 空 steps。""" + with mock.patch("service.admin_routes._workdir_for_trace", + side_effect=lambda tid: tmp_path / tid): + resp = client.get("/control/traces/does-not-exist") + assert resp.status_code == 404, resp.text + data = resp.json() + assert data["found"] is False, f"found 应为 False:{data}" + assert data["steps"] == [], f"steps 应为空:{data}" + assert data["count"] == 0 + print(f" [ok] 不存在 traceId → 404 found=false") + + +# ════════════════════════════════════════════════════════════════════════════ +# 端点三:GET /control/cost/{id} +# ════════════════════════════════════════════════════════════════════════════ +def test_cost_live_with_stub_pricing(client, tmp_path): + """样本 trace + 打桩取价(live)→ 200 + costRmb>0 + pricingSource=live + byModel 含模型。""" + _write_sample_trace(tmp_path) + # 桩 new-api 计费三件套(模拟网关返回,避免真网络;cost_for_run 是纯函数,折算确定)。 + stub_params = { + "pricing": {"MiniMax-M3": {"model_ratio": 2.0, "completion_ratio": 4.0, "cache_ratio": 0.1}}, + "qpu": 500000.0, + "usd_rate": 7.3, + } + with mock.patch("service.admin_routes._workdir_for_trace", + side_effect=lambda tid: tmp_path / tid), \ + mock.patch("observability.newapi_pricing.fetch_pricing_params", + return_value=stub_params): + resp = client.get("/control/cost/t-sample") + assert resp.status_code == 200, resp.text + data = resp.json() + assert "costRmb" in data, f"响应应含 costRmb:{data}" + assert data["pricingSource"] == "live", f"pricingSource 应为 live:{data}" + assert data["costRmb"] > 0, f"样本有 token,costRmb 应 > 0:{data}" + assert "MiniMax-M3" in data["byModel"], f"byModel 应含 MiniMax-M3:{data}" + print(f" [ok] live costRmb={data['costRmb']} byModel={list(data['byModel'])}") + + +def test_cost_degraded_when_pricing_unreachable(client, tmp_path): + """取价不可达(fetch_pricing_params→None)→ 200 + pricingSource=degraded + 含 costRmb,不崩。""" + _write_sample_trace(tmp_path) + with mock.patch("service.admin_routes._workdir_for_trace", + side_effect=lambda tid: tmp_path / tid), \ + mock.patch("observability.newapi_pricing.fetch_pricing_params", + return_value=None): + resp = client.get("/control/cost/t-sample") + assert resp.status_code == 200, resp.text # 取价失败不抛 500 + data = resp.json() + assert data["pricingSource"] == "degraded", f"pricingSource 应为 degraded:{data}" + assert "costRmb" in data, f"degraded 也应带 costRmb 字段:{data}" + print(f" [ok] degraded(取价不可达)costRmb={data['costRmb']} 不崩") + + +def test_cost_trace_not_found_returns_404(client, tmp_path): + """cost 对不存在的 traceId → 404(与 /traces 同语义,不崩)。""" + with mock.patch("service.admin_routes._workdir_for_trace", + side_effect=lambda tid: tmp_path / tid): + resp = client.get("/control/cost/no-such-trace") + assert resp.status_code == 404, resp.text + data = resp.json() + assert data["pricingSource"] == "notfound", f"应标 notfound:{data}" + print(f" [ok] cost 不存在 traceId → 404 notfound") diff --git a/tier2/gen-worker/tests/test_jsonl_sink.py b/tier2/gen-worker/tests/test_jsonl_sink.py new file mode 100644 index 00000000..48950326 --- /dev/null +++ b/tier2/gen-worker/tests/test_jsonl_sink.py @@ -0,0 +1,130 @@ +"""test_jsonl_sink.py —— U-C1a JSONL sink 落盘单测(TDD:先红后绿)。 + +跑:python3 tests/test_jsonl_sink.py(无 pytest,__main__ 跑全部断言)。 +验 JsonlFileSink: + 1. emit 三条 trace step → trace.jsonl 落盘; + 2. 每行是合法 JSON; + 3. 五字段(traceId / step / cost / verdict / timestamp)齐全; + 4. 目录不存在 → 自动创建; + 5. 写失败(不可写路径) → 只告警、不抛,不中断调用方。 +""" +import sys +import os +import json +import tempfile +from pathlib import Path + +_HERE = os.path.dirname(os.path.abspath(__file__)) +_GW = os.path.dirname(_HERE) # gen-worker/ +sys.path.insert(0, _GW) + +from observability.trace import JsonlFileSink, make_jsonl_sink, TraceAdapter # noqa: E402 + + +# ── 测试一:核心落盘 + 五字段完整性 ────────────────────────────────────────────── +def test_jsonl_sink_creates_file_and_lines(): + """emit 三条 trace step → trace.jsonl 落盘、每行合法 JSON、五字段齐、内容对。""" + with tempfile.TemporaryDirectory() as tmpdir: + out = Path(tmpdir) / "trace.jsonl" + sink = JsonlFileSink(out) + + # 用 TraceAdapter + dict 桩事件(无需 AgentScope 真环境)驱动 sink。 + adapter = TraceAdapter(trace_id="test-trace-001", sink=sink) + fake_events = [ + # ReplyStartEvent → 落 raw.reply_boundary / agent_name + {"type": "ReplyStartEvent", "created_at": "2026-06-29T10:00:00", + "reply_id": "r1", "name": "writer"}, + # ModelCallEndEvent → cost.tokens 非空 + {"type": "ModelCallEndEvent", "created_at": "2026-06-29T10:00:01", + "input_tokens": 100, "output_tokens": 50}, + # ToolResultEndEvent → verdict=success + {"type": "ToolResultEndEvent", "created_at": "2026-06-29T10:00:02", + "state": "success"}, + ] + for e in fake_events: + adapter.ingest(e) + + # 断言①:文件已落盘 + assert out.exists(), f"trace.jsonl 应存在,但未落盘:{out}" + + lines = out.read_text(encoding="utf-8").strip().split("\n") + # 断言②:三条事件 → 三行 + assert len(lines) == 3, f"应有 3 行,实际 {len(lines)} 行:\n" + "\n".join(lines) + + # 断言③:每行合法 JSON + 五字段齐 + CORE_FIELDS = {"traceId", "step", "cost", "verdict", "timestamp"} + for i, line in enumerate(lines): + obj = json.loads(line) # 若非合法 JSON 此处抛 + missing = CORE_FIELDS - set(obj.keys()) + assert not missing, f"第 {i} 行缺五字段 {missing}:{line}" + assert obj["traceId"] == "test-trace-001", ( + f"第 {i} 行 traceId 错(期望 test-trace-001,实际 {obj['traceId']})") + assert obj["step"] == i, f"第 {i} 行 step 错(期望 {i},实际 {obj['step']})" + + # 断言④:ModelCallEnd 步 cost 非空且 token 字段正确 + second = json.loads(lines[1]) + assert second["cost"] is not None, "ModelCallEnd 步 cost 应非空" + assert second["cost"]["tokens"]["in"] == 100, "input_tokens 应为 100" + assert second["cost"]["tokens"]["out"] == 50, "output_tokens 应为 50" + + # 断言⑤:ToolResultEnd 步 verdict=success + third = json.loads(lines[2]) + assert third["verdict"] == "success", ( + f"ToolResultEnd 步 verdict 应为 success,实际 {third['verdict']}") + + print(f" [ok] lines={len(lines)}, traceId={json.loads(lines[0])['traceId']}, " + f"cost.tokens.in={second['cost']['tokens']['in']}, verdict={third['verdict']}") + + +# ── 测试二:嵌套目录自动创建 ───────────────────────────────────────────────────── +def test_jsonl_sink_auto_creates_dir(): + """目录不存在 → sink 自动创建嵌套目录;emit 成功落盘。""" + with tempfile.TemporaryDirectory() as tmpdir: + out = Path(tmpdir) / "subdir" / "nested" / "trace.jsonl" + sink = make_jsonl_sink(out) + step = {"traceId": "t2", "step": 0, "cost": None, "verdict": None, + "timestamp": "2026-06-29T00:00:00", + "ext": {"raw": {"event": "ReplyStartEvent"}}} + sink(step) + + assert out.exists(), f"嵌套目录下 trace.jsonl 应自动创建:{out}" + obj = json.loads(out.read_text(encoding="utf-8").strip()) + assert obj["traceId"] == "t2", f"traceId 错:{obj['traceId']}" + print(f" [ok] auto-mkdir:{out}") + + +# ── 测试三:写失败只告警、不抛 ──────────────────────────────────────────────────── +def test_jsonl_sink_write_failure_no_raise(): + """写失败(不可写路径) → 只 print 告警,绝不抛,不中断调用方。""" + # /proc/nonexistent/ 在 macOS 下不存在且不可写,触发写失败场景。 + bad_path = Path("/proc/nonexistent_for_test_sink/trace.jsonl") + sink = JsonlFileSink(bad_path) + try: + sink({"traceId": "t3", "step": 0, "cost": None, "verdict": None, + "timestamp": "", "ext": {"raw": {"event": "X"}}}) + print(" [ok] write-fail → 只告警,不抛") + except Exception as e: + raise AssertionError(f"sink 写失败时不应抛异常,但抛了:{e}") from e + + +# ── 入口 ──────────────────────────────────────────────────────────────────────── +if __name__ == "__main__": + tests = [ + ("test_jsonl_sink_creates_file_and_lines", test_jsonl_sink_creates_file_and_lines), + ("test_jsonl_sink_auto_creates_dir", test_jsonl_sink_auto_creates_dir), + ("test_jsonl_sink_write_failure_no_raise", test_jsonl_sink_write_failure_no_raise), + ] + passed = failed = 0 + for name, fn in tests: + try: + print(f"[RUN] {name}") + fn() + print(f"[PASS] {name}") + passed += 1 + except Exception as exc: + import traceback + print(f"[FAIL] {name}: {exc}") + traceback.print_exc() + failed += 1 + print(f"\n结果:{passed} 通过 / {failed} 失败") + sys.exit(0 if failed == 0 else 1) diff --git a/tier2/gen-worker/worker/agent_loop/studio.py b/tier2/gen-worker/worker/agent_loop/studio.py index f0cd79ff..b0d0b9f6 100644 --- a/tier2/gen-worker/worker/agent_loop/studio.py +++ b/tier2/gen-worker/worker/agent_loop/studio.py @@ -51,11 +51,13 @@ except ImportError: # pragma: no cover —— 直接 python studio.py 兜底 try: from observability.cost import tokens_by_model_from_records, cost_for_run from observability.newapi_pricing import fetch_pricing_params, DEFAULT_QUOTA_PER_UNIT, DEFAULT_USD_RATE + from observability.trace import make_jsonl_sink except Exception: # pragma: no cover —— 直跑/路径未就位兜底 sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from observability.cost import tokens_by_model_from_records, cost_for_run # type: ignore from observability.newapi_pricing import ( # type: ignore fetch_pricing_params, DEFAULT_QUOTA_PER_UNIT, DEFAULT_USD_RATE) + from observability.trace import make_jsonl_sink # type: ignore # 框架接缝(2.0.2):Agent + ReActConfig + UserMsg + AgentState + BYPASS 权限。 from agentscope.agent import Agent, ReActConfig @@ -365,9 +367,11 @@ async def run_studio( toolkit = build_toolkit(session) breaker = CircuitBreakerMiddleware() # 四道熔断 + 软刹(on_reply / on_system_prompt) # trace 接线(H1/H2):traceId 用 game_id(贯穿本次生成,对接 verdict.evidence.traceId / 成本关联键)。 - # sink=None → 只留内存 steps(spike 期收口由编排器读 adapter.summary;真落库 sink 随控制面 phase-1 接)。 + # JSONL sink → 把每条 trace step(五字段 schema:traceId/step/cost/verdict/timestamp + ext) + # 追加写到产物 workdir/trace.jsonl,供成本对账 / replay 用;sink 失败 best-effort 只告警不阻断主链。 # 单实例对应单 agent;它与 breaker 都挂 on_reply,由框架按 middlewares 列表序串成洋葱链(trace 在外、纯旁路)。 - tracer = Tier2TraceMiddleware(trace_id=game_id) + _trace_jsonl_path = run._workdir(game_id) / "trace.jsonl" + tracer = Tier2TraceMiddleware(trace_id=game_id, sink=make_jsonl_sink(_trace_jsonl_path)) writer_model = config.build_model( mname, max_tokens=max_tokens, thinking_budget=thinking_budget) @@ -629,7 +633,8 @@ async def run_studio( # RunRecord 接线(B2)从此抽 cost_yuan / tokens_by_model 填 G4 采集字段。 "cost": cost, # trace 收口摘要(H1/H2,新增):{traceId, steps, dropped, middlewareDropped}。 - # spike 期 trace step 留在 tracer.adapter.steps 内存(sink=None);真落库 sink 随控制面 phase-1 接。 + # trace step 既留 tracer.adapter.steps 内存(供本摘要 / replay),又经 JsonlFileSink 落盘 + # 到 workdir/trace.jsonl(控制面 phase-1 已接,见上文 make_jsonl_sink 接线点)。 "trace": trace_summary, "wall_s": wall_s, "file_tree": session.file_tree(),