From ba28dee1f05da52d4c1aba01fda03365d2437acb Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Fri, 16 Oct 2026 10:45:23 +0800 Subject: [PATCH] =?UTF-8?q?=E7=94=9F=E6=88=90=E8=BF=87=E7=A8=8B=E6=8A=BD?= =?UTF-8?q?=E6=88=90=20answer=5Fops=E3=80=81=E5=A4=8D=E6=A0=B8=E6=8A=BD?= =?UTF-8?q?=E6=88=90=20review=5Fof=5Frun=EF=BC=8C=E5=8A=A9=E6=89=8B?= =?UTF-8?q?=E5=BB=BA=E5=9B=BE=E5=92=8C=E5=A4=8D=E6=A0=B8=E7=9A=84=E7=94=A8?= =?UTF-8?q?=E9=87=8F=E4=B9=9F=E8=AE=B0=E4=B8=8B=EF=BC=88=E7=BB=86=E6=A1=88?= =?UTF-8?q?=2001=20=E5=81=9A=E6=B3=95=202=E3=80=813=EF=BC=89=EF=BC=9A?= =?UTF-8?q?=E8=AF=84=E6=B5=8B=E5=92=8C=E9=97=AE=E6=95=B0=E6=8D=AE=E9=A1=B5?= =?UTF-8?q?=E8=B5=B0=E5=90=8C=E4=B8=80=E6=9D=A1=E7=94=9F=E6=88=90=E8=B7=AF?= =?UTF-8?q?=E5=BE=84=EF=BC=8C=E8=BF=99=E4=B8=80=E8=BD=AE=E7=9A=84=E8=8A=B1?= =?UTF-8?q?=E8=B4=B9=E4=B8=8D=E5=86=8D=E4=BD=8E=E6=8A=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - api/copilot.py：event_stream 原样搬成 async generator answer_ops(payload, *, catalog_view=None)，交出的 op dict 不变；开流前的准备（数据源、目录检查器、助手模型、历史、查看服务）用自己的短会话，在第一条操作之前做完。SSE 路由只包一层：先取第一条，开流前的拒绝（模型没配好、限定的数据源都不在）照旧是 HTTP 错误 - catalog_view 一路传给查看服务和自查的 SQL 检查器（目录视图接缝的「生成过程从参数读」） - providers/usage.py：Meter 是一个 LangChain 回调，和 get_usage_metadata_callback 同一个机制（登记过的上下文变量），按模型累计 tokens、次数，金额按价格目录；估不出的记 unpriced_calls。没直接用 get_usage_metadata_callback：with 块跨 yield 会把变量留在调用方上下文里，调用方随后发起的运行会继承它、被重复计一遍；metered 只在每一步 __anext__ 期间挂上，嵌套时并进外层 - 生成过程读到 done / look / ask 就关模型的流（后面的闲话不要），流末尾才报的用量拿不到：这一次照样计次数和已报的 token，记成估不出，不悄悄少算（遗留：真实接入上建图那几次多半是下限） - providers/budget.py：BudgetFallback.astream 用 aclosing 包住里面那条流——以前调用方关流时里面那条要等垃圾回收才关，模型那次调用的收尾（用量回调）落在算完账之后；测试去掉它回跑会少计一次 - answer_ops 末尾多一条 {op: usage, by_model, input_tokens, output_tokens, calls, cost_usd, unpriced_calls}；/copilot/review 拆出 review_of_run（问数据页、评测共用），结果带 usage（没调模型时全 0）和 retry_instruction（要不要带着原图重来、说什么，按 chat/turn.py 的决定算；ReviewIn 多一个 attempt） - 新增 app/chat/turn.py（问数据一轮的决定，复核这边先用上；final.next、对照测试、前端照做在下一个提交）；新增 tests/test_copilot_answer_ops.py（八种结局的 SSE 输出和抽出来之前录下的 fixtures/copilot_answer_ops_baseline.json 逐条相同，末尾多一条 usage；直接迭代和 SSE 同一串；开流前的拒绝仍是 HTTP 错误）、tests/test_usage_meter.py；读 SSE 的十一个测试文件的收流帮手跳过末尾那条 usage --- backend/app/api/copilot.py | 211 ++-- backend/app/chat/turn.py | 129 +++ backend/app/providers/budget.py | 16 +- backend/app/providers/usage.py | 204 ++++ .../fixtures/copilot_answer_ops_baseline.json | 1030 +++++++++++++++++ backend/tests/test_catalog_patch.py | 3 +- backend/tests/test_conversations.py | 3 +- backend/tests/test_copilot_answer_ops.py | 165 +++ backend/tests/test_copilot_evidence_guide.py | 3 +- backend/tests/test_copilot_look.py | 3 +- backend/tests/test_copilot_merge.py | 3 +- backend/tests/test_copilot_patch_layout.py | 3 +- backend/tests/test_copilot_prompt_evidence.py | 3 +- backend/tests/test_copilot_scope.py | 3 +- backend/tests/test_copilot_selfcheck.py | 3 +- backend/tests/test_copilot_sqlcheck.py | 3 +- backend/tests/test_copilot_truncation.py | 9 +- backend/tests/test_usage_meter.py | 283 +++++ 18 files changed, 1983 insertions(+), 94 deletions(-) create mode 100644 backend/app/chat/turn.py create mode 100644 backend/app/providers/usage.py create mode 100644 backend/tests/fixtures/copilot_answer_ops_baseline.json create mode 100644 backend/tests/test_copilot_answer_ops.py create mode 100644 backend/tests/test_usage_meter.py diff --git a/backend/app/api/copilot.py b/backend/app/api/copilot.py index 2f8766a5..469e5864 100644 --- a/backend/app/api/copilot.py +++ b/backend/app/api/copilot.py @@ -110,11 +110,12 @@ def _tool_catalog(rows: list[Any], *, builtins: frozenset[str] | None = None) -> async def _look_service(session: AsyncSession, payload: GenerateIn, sources: list[Any], *, graph: dict[str, Any] | None, kinds: tuple[str, ...] = ("table", "capability"), - ) -> Any: - """这一轮的查看服务（copilot_context.look_service）。挑选模型用同一个助手模型，第一次要挑时才拿。""" + catalog_view: Any = None) -> Any: + """这一轮的查看服务（copilot_context.look_service）。挑选模型用同一个助手模型，第一次要挑时才拿。catalog_view：表详情 + 里给哪一份数据目录（data/catalog_view.py），不给是当前目录。""" spec = await copilot_model_spec(session, payload, max_tokens=copilot_context.PICK_MAX_TOKENS) return await copilot_context.look_service(sources, spec=spec, graph=graph, kinds=kinds, - answer=payload.intent == "answer") + answer=payload.intent == "answer", catalog_view=catalog_view) NODE_REFERENCE = """\ @@ -1756,45 +1757,79 @@ def _repair_request( @router.post("/generate-stream") -async def generate_stream(payload: GenerateIn, session: AsyncSession = Depends(get_session)): +async def generate_stream(payload: GenerateIn): """流式版生成：SSE 逐操作推送，前端边收边把节点摆上画布。 - 生成过程（copilot_gen）读到 look 就停下、查看、接着写，查看记录作为 look 事件发给前端（前端存进这一轮的 - meta.looks）。结束时（done 或流断）发一条 final：服务端把累积的图排版、校验后整体给出——过程可见性来自操作流， - 最终质量仍由和手工编排相同的 layout + validate 保证。模型问了用户（ask）的，这一轮到此为止，不自查修正、不自动运行。 + 生成过程整个在 answer_ops 里（评测直接迭代它，和问数据页走同一条路），这里只包一层 SSE。开流前的拒绝（模型没配好、 + 对话读不了、限定的数据源都不在）发生在 answer_ops 交出第一条操作之前：先取第一条，拒绝照旧是 HTTP 错误，不进流。 """ from fastapi.responses import StreamingResponse - from app.data.sqlcheck import load_checkers + ops = answer_ops(payload) + first = await ops.__anext__() + async def event_stream(): + yield _sse(first) + async for op in ops: + yield _sse(op) + + return StreamingResponse( + event_stream(), + media_type="text/event-stream", + headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}, + ) + + +async def answer_ops(payload: GenerateIn, *, catalog_view: Any = None): + """生成过程：按需求起草、按需查看、自查修正、排版校验，逐条交出操作（op dict）。问数据页（经 SSE）和评测（直接迭代） + 走的是这同一条路（细案 01 做法 2）。 + + 读到 look 就停下、查看、接着写，查看记录作为 look 操作交出（前端存进这一轮的 meta.looks）。结束时（done 或流断）交一条 + final：累积的图排版、校验后整体给出——过程可见性来自操作流，最终质量仍由和手工编排相同的 layout + validate 保证。模型 + 问了用户（ask）的，这一轮到此为止，不自查修正、不自动运行。 + + 开流前的准备（取数据源、读目录检查器、取助手模型、拼历史、组查看服务）用自己的短会话，在交出第一条操作之前做完： + 拒绝在这里抛 HTTPException（调用方没迭代到第一条就知道）。catalog_view 是这一轮读哪一份数据目录（data/catalog_view.py： + 查看的表详情、自查的 SQL 检查器都按它），不给是当前目录。 + + 最后多交一条 usage（providers/usage.py）：{op: usage, by_model, input_tokens, output_tokens, calls, cost_usd, + unpriced_calls}——这一轮助手建图（含挑表、接着写、自查修正）花了多少。前面每一条和抽出来之前逐条一样。运行不在这里， + 不计。 + """ from app.api.deps import restricted + from app.data.catalog_view import CatalogView + from app.data.sqlcheck import load_checkers + from app.db.base import SessionLocal + from app.providers import usage - # 带了对话的，先核对这个人能不能用它（D-d）：历史、上一轮的图都从它来 - await _ensure_conversation_access(session, payload.conversation_id) - # 开登录后的 user 角色（D-c）：图只能用 chat_policy 放行的节点和工具，提示词先说、自查按同一个判断挡 - chat_user = restricted() - sources = await _sources(session, payload.datasource_ids) - scope = {r.name for r in sources} if payload.datasource_ids else None - # 基于数据目录的 SQL 检查器：在开流之前读好目录（流里不再碰这个会话），自查和收尾共用 - checkers = await load_checkers(session, sources) - nodes, edges = _start_graph(payload) - # 用户摆好的坐标，收尾排版时原样留着 - pinned = _pinned_positions(payload.base_graph) + view = CatalogView.parse(catalog_view) + async with SessionLocal() as session: + # 带了对话的，先核对这个人能不能用它（D-d）：历史、上一轮的图都从它来 + await _ensure_conversation_access(session, payload.conversation_id) + # 开登录后的 user 角色（D-c）：图只能用 chat_policy 放行的节点和工具，提示词先说、自查按同一个判断挡 + chat_user = restricted() + sources = await _sources(session, payload.datasource_ids) + scope = {r.name for r in sources} if payload.datasource_ids else None + # 基于数据目录的 SQL 检查器：在开流之前读好目录（流里不再碰这个会话），自查和收尾共用 + checkers = await load_checkers(session, sources, catalog_view=view) + nodes, edges = _start_graph(payload) + # 用户摆好的坐标，收尾排版时原样留着 + pinned = _pinned_positions(payload.base_graph) - try: - model, model_id = await _op_model(session, payload) - except ProviderNotConfigured as e: - raise HTTPException(400, _unconfigured(e)) from e + try: + model, model_id = await _op_model(session, payload) + except ProviderNotConfigured as e: + raise HTTPException(400, _unconfigured(e)) from e - human = _with_history(_graph_request(payload), await _history_section(session, payload.conversation_id)) - # 兜底用：模型只发了改动操作时，把它们重放到这张图上 - prev_graph = await _previous_graph(session, payload.conversation_id) - # 比对工具绑定的基准：画布上是改之前的图；问数据页是上一轮的图（沿用它的节点 id） - baseline = copy.deepcopy( - (payload.base_graph or {}).get("nodes") or (prev_graph or {}).get("nodes") or []) - # 查看服务：读目录、读能力登记在开流之前做完。现有工作流（画布上的原图、问数据追问时上一轮的图）查过的表列在概览里 - looks = await _look_service(session, payload, sources, graph=payload.base_graph if ( - payload.base_graph or {}).get("nodes") else prev_graph) + human = _with_history(_graph_request(payload), await _history_section(session, payload.conversation_id)) + # 兜底用：模型只发了改动操作时，把它们重放到这张图上 + prev_graph = await _previous_graph(session, payload.conversation_id) + # 比对工具绑定的基准：画布上是改之前的图；问数据页是上一轮的图（沿用它的节点 id） + baseline = copy.deepcopy( + (payload.base_graph or {}).get("nodes") or (prev_graph or {}).get("nodes") or []) + # 查看服务：读目录、读能力登记在开流之前做完。现有工作流（画布上的原图、问数据追问时上一轮的图）查过的表列在概览里 + looks = await _look_service(session, payload, sources, graph=payload.base_graph if ( + payload.base_graph or {}).get("nodes") else prev_graph, catalog_view=view) # 这一份 system 一直用到最后：接着写、自查修正都沿用，不换 system = _graph_system(sources, scope, await copilot_context.section(looks), looks, chat_user=chat_user) applier = GraphApplier(nodes, edges) @@ -1803,8 +1838,8 @@ async def generate_stream(payload: GenerateIn, session: AsyncSession = Depends(g checker=GraphChecker(applier, scope=scope, sources=sources, baseline=baseline, checkers=checkers, chat_user=chat_user)) - async def event_stream(): - yield _sse({"op": "model", "model": model_id}) + async def steps(): + yield {"op": "model", "model": model_id} patches = 0 try: async for op in _with_heartbeat(gen.draft(human), started=started): @@ -1813,23 +1848,23 @@ async def generate_stream(payload: GenerateIn, session: AsyncSession = Depends(g # reply 先压着（applier.replied）：前端收到 reply 就收尾，提案得排在它前面转出 if (checked := await catalog_patch_op(op, sources)) is not None: patches += 1 - yield _sse(checked) + yield checked continue - yield _sse(op) + yield op if applier.replied is not None: # 这句话不需要工作流：后面那套排版校验对着一张空图跑，只会得到"图是空的，先拖一个节点进来" - yield _sse(applier.replied) + yield applier.replied return if gen.asked is not None and not applier.effective: # 先问清楚才能动手，什么都还没改：这一轮到此为止。照样发一条 final 收尾，带上问题（asked）和没动过的图： # 前端据此把这一轮记成「助手反问」、问题进历史（以前流就这么断了，问数据页一直停在「进行中」）， # 不运行、不重排。图给一份对象（画布上原来的图、问数据上一轮的图，都没有就是空图），老前端照样认得 - yield _sse(_asked_final(gen.asked, payload.base_graph, prev_graph)) + yield _asked_final(gen.asked, payload.base_graph, prev_graph) return # 只发了目录修改提案，既没回答也没改图（用户只是在说一条数据事实）：补一句回答收尾， # 不能按下面的「模型没有给出任何修改」报错——提案已经摆在用户面前了 if patches and all(o.get("op") == "plan" for o in applier.seen): - yield _sse({"op": "reply", "text": PATCH_ONLY_REPLY}) + yield {"op": "reply", "text": PATCH_ONLY_REPLY} return # 流正常结束，却没有 done、也没有一个操作能认（格式跑偏，每一行都被跳过）： # 往下走就是拿原图去自查、交付，界面上写「无需修改」——而模型什么都没说 @@ -1837,12 +1872,11 @@ async def generate_stream(payload: GenerateIn, session: AsyncSession = Depends(g raise CopilotIncomplete("模型没有给出任何修改，也没有说明原因", hint="画布未修改。可以重试，或把需求描述得更具体") except CopilotIncomplete as e: - yield _sse({"op": "error", "message": f"助手本轮未完成：{e}", "hint": e.hint, "detail": e.detail}) + yield {"op": "error", "message": f"助手本轮未完成：{e}", "hint": e.hint, "detail": e.detail} return except Exception as e: # noqa: BLE001 reason, hint = explain_error(e) - yield _sse({"op": "error", "message": f"助手本轮未完成：{reason}", - "hint": hint, "detail": raw_error(e)}) + yield {"op": "error", "message": f"助手本轮未完成：{reason}", "hint": hint, "detail": raw_error(e)} return # 兜底：一个操作都没落地，而上一轮有图。 @@ -1858,7 +1892,7 @@ async def generate_stream(payload: GenerateIn, session: AsyncSession = Depends(g # 以前这些问题照样交付、照样自动开跑——跑到那一步才炸（开发库里四次运行死在 # 循环 / 分支条件里套了 {{ }}），前面的步骤白跑，报错还停在半路 async for op in _with_heartbeat(gen.repair(), phase="repairing"): - yield _sse(op) + yield op remaining = _blocking_issues(nodes, edges, scope, sources=sources, baseline=baseline, checkers=checkers, chat_user=chat_user) # 工具绑定变化不挡运行，但要在自查这一步就说出来：工具被改没了的图照样 @@ -1871,11 +1905,9 @@ async def generate_stream(payload: GenerateIn, session: AsyncSession = Depends(g # 改不好也照实交付、列出问题，但不自动运行：明知跑不起来的图，开跑只会 # 在半路报一个用户看不懂的错 autorun = False - yield _sse({"op": "check", "status": "failed", "issues": remaining, - "warnings": warnings}) + yield {"op": "check", "status": "failed", "issues": remaining, "warnings": warnings} elif remaining is not None: - yield _sse({"op": "check", "status": "passed", "repaired": gen.repaired, - "warnings": warnings}) + yield {"op": "check", "status": "passed", "repaired": gen.repaired, "warnings": warnings} # 收尾：排版 + 校验，把最终图整体交付。改图时旧节点留在原处，只排新节点 try: @@ -1916,13 +1948,12 @@ async def generate_stream(payload: GenerateIn, session: AsyncSession = Depends(g except Exception as e: # noqa: BLE001 final = {"op": "error", "message": f"模型返回的工作流结构有误：{graph_error(e)}", "hint": "请重试，或把需求描述得更具体", "detail": raw_error(e)} - yield _sse(final) + yield final - return StreamingResponse( - event_stream(), - media_type="text/event-stream", - headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}, - ) + spent = usage.meter() + async for op in usage.metered(steps(), spent): + yield op + yield {"op": "usage", **spent.summary()} # -------------------------------------------------------------------------- @@ -2774,13 +2805,31 @@ class ReviewIn(BaseModel): #: 默认只在扫出异常信号时才叫模型。打开它就每次都复核——排查时有用， #: 平时不开：干净的运行没什么可复核的，代价却是每次问数都多一次调用 always: bool = False + #: 这一轮已经自动试了几次（从 0 数，同 chat/turn.retry_after_review）：结果里的 retry_instruction 据此给不给。 + #: 不传按第一次算 + attempt: int = Field(default=0, ge=0) @router.post("/review") async def review_run( payload: ReviewIn, session: AsyncSession = Depends(get_session) ) -> dict[str, Any]: - """接住一次编排的产出，判断它能不能就这么交给用户。 + """接住一次编排的产出，判断它能不能就这么交给用户（规则见 review_of_run）。这里只核对能不能读这次运行。""" + from app.api.deps import PRIVATE_RUN, can_read_run, current_principal + from app.db.models import Run + + run = await session.get(Run, payload.run_id) + if not run: + raise HTTPException(404, "运行记录不存在，可能已被删除") + if not await can_read_run(current_principal(), run): + raise HTTPException(403, PRIVATE_RUN) + return await review_of_run(session, run, question=payload.question, provider=payload.provider, + model=payload.model, always=payload.always, attempt=payload.attempt) + + +async def review_of_run(session: AsyncSession, run: Any, *, question: str, provider: str | None = None, + model: str | None = None, always: bool = False, attempt: int = 0) -> dict[str, Any]: + """复核一次运行（问数据页经 /copilot/review、评测直接调，同一个函数）。 「跑完了」和「答得对」是两回事。运行状态只说明流程走完了，而过程中检索 降级、工具报错、agent 步数用满这些事，原先一件都不会进到答案里——用户拿到 @@ -2788,47 +2837,57 @@ async def review_run( 规则层先扫一遍（engine/review.py:scan），没扫出东西就到此为止、直接返回， 不花一次模型调用；扫出来了才叫模型重组答案并把异常说破。 - """ - from app.api.deps import PRIVATE_RUN, can_read_run, current_principal - from app.db.models import Run, RunEvent - from app.engine import review as rv - run = await session.get(Run, payload.run_id) - if not run: - raise HTTPException(404, "运行记录不存在，可能已被删除") - if not await can_read_run(current_principal(), run): - raise HTTPException(403, PRIVATE_RUN) + 结果是 ReviewResult.as_dict()，另带两项： + - retry_instruction：要不要带着原图重来一次（chat/turn.retry_after_review，按 attempt 算这一轮还能不能再试）——要的话 + 是交给助手的那句话（chat/turn.retry_instruction），不要是 null。问数据页照它走； + - usage：复核这一次模型调用花了多少（providers/usage.py 的汇总，没调模型时是全 0），这一轮的花费 = 助手 + 运行 + 复核。 + """ + from app.chat import turn + from app.db.models import RunEvent + from app.engine import review as rv + from app.providers import usage rows = list((await session.execute( - select(RunEvent).where(RunEvent.run_id == payload.run_id).order_by(RunEvent.seq) + select(RunEvent).where(RunEvent.run_id == run.id).order_by(RunEvent.seq) )).scalars()) output = run.output if isinstance(run.output, dict) else None signals = rv.scan(rows, output) answer = rv.answer_text(output) + spent = usage.meter() - if not signals and not payload.always: - return rv.ReviewResult(verdict="ok", signals=[]).as_dict() + def done(result: Any) -> dict[str, Any]: + out = result.as_dict() + out["retry_instruction"] = ( + turn.retry_instruction(question, turn.review_reason(out), turn.has_nodes(run.graph)) + if turn.retry_after_review(out, attempt) else None) + out["usage"] = spent.summary() + return out + + if not signals and not always: + return done(rv.ReviewResult(verdict="ok", signals=[])) try: spec = await copilot_model_spec(session, GenerateIn( - instruction="review", provider=payload.provider, model=payload.model, + instruction="review", provider=provider, model=model, ), max_tokens=4096) - model, _ = await get_chat_model(session, spec) + chat_model, _ = await get_chat_model(session, spec) except (ProviderNotConfigured, HTTPException) as e: # 没配模型不该让这一轮挂掉：规则层已经知道出了什么事，照样说得出话 - return rv.ReviewResult( + return done(rv.ReviewResult( verdict="annotated", note="；".join(s.detail for s in signals) or ( _unconfigured(e) if isinstance(e, ProviderNotConfigured) else str(e.detail) ), retry=rv.should_retry(signals), signals=signals, severity=rv.worst(signals), - ).as_dict() + )) - result = await rv.review( - model, question=payload.question, answer=answer, signals=signals, - # 报告撰写节点产出的答案逐段对应着证据文档：只能加说明，不能改写 - locked=bool(output and output.get("_evidence")), - ) - return result.as_dict() + with usage.metering(spent): + result = await rv.review( + chat_model, question=question, answer=answer, signals=signals, + # 报告撰写节点产出的答案逐段对应着证据文档：只能加说明，不能改写 + locked=bool(output and output.get("_evidence")), + ) + return done(result) diff --git a/backend/app/chat/turn.py b/backend/app/chat/turn.py new file mode 100644 index 00000000..993c0ad5 --- /dev/null +++ b/backend/app/chat/turn.py @@ -0,0 +1,129 @@ +"""问数据一轮里的几个「决定」（细案 01 做法 1，总方案 2.1）。 + +一轮问数据是：建图（助手的生成过程，api/copilot.answer_ops）→ 要不要运行 → 运行（输入怎么填）→ 复核 → 要不要带着原图 +重来一次、重来时说什么。以前这几个决定都写在问数据页里（frontend/src/store/chat.ts 的 plan / launch / settle / retry）， +后端跑不了一轮问数据：评测要么照着抄一份（迟早走样，量出来的不是用户拿到的那一轮），要么做不了；这也违反「界面层不放 +逻辑」。现在决定写在这里，是纯函数： + +- next_step：生成过程交出结局之后做什么； +- fill_input：用户的问题填进这张图的入口字段； +- retry_after_review：复核之后要不要带着原图重来一次； +- retry_instruction：重来时交给助手的那句话（两种措辞）。 + +接口把决定交给问数据页：/copilot/generate-stream 的 final 带 next（run | ready | fail），/copilot/review 的结果带 +retry_instruction（要重来时是那句话，不重来是 null）。问数据页照后端给的走；老后端没有这两个字段时退回它自己那份同样的 +规则（tests/test_chat_turn_policy.py 逐条对照两边的常量和措辞）。评测（第 3 波）用同一组函数把一轮串起来。 + +只搬「决定」：重连、断点续看、取消、复核等多久这些仍在问数据页（把整轮编排搬到后端要重做这些，单独立项）。 +""" +from __future__ import annotations + +import re +from collections.abc import Mapping +from typing import Any + +#: 一轮最多自动试几次（含第一次）：第二次还不行多半是问题本身没说清，该让用户介入（同 chat.ts 的 MAX_ATTEMPTS） +MAX_ATTEMPTS = 2 + +#: final 之后的三种走法（final.next）：运行、只建不跑（摆着等用户点运行）、判失败（没过检查的图不拿去跑） +NEXT_RUN, NEXT_READY, NEXT_FAIL = "run", "ready", "fail" +NEXT_STEPS = (NEXT_RUN, NEXT_READY, NEXT_FAIL) +#: 别的结局：直接回答了（reply）、助手反问了等用户答（asked）、没自查过的老服务端交来没过检查的图时重来一次（retry） +NEXT_REPLY, NEXT_ASKED, NEXT_RETRY = "reply", "asked", "retry" + +#: 问题往哪个入口字段放：名字像「问题」的优先（同 chat.ts fillInput 的正则，大小写不敏感） +QUESTION_FIELD = re.compile(r"question|query|topic|input|问题|需求|主题", re.IGNORECASE) + +#: 带着原图重来时交给助手的话：原图交回去改（调大步数、改参数这种事就是改一个字段，不是推倒重来） +RETRY_WITH_BASE = ("这个工作流上一次运行没有成功：{reason}。\n\n" + "请针对这个原因改它（比如调大最大步数、改正工具参数、换个查询方式），" + "改完继续回答原来的问题：{question}") +#: 没有图可交回（上一次连图都没建出来）时：原问题后面缀上原因，从零再建 +RETRY_FROM_SCRATCH = "{question}\n\n（上一次没有成功：{reason}。请针对这个原因调整，重新给出完整的工作流。）" + + +def _blockers(final: Mapping[str, Any]) -> list[str]: + return [str(i.get("message") or "") for i in final.get("issues") or [] + if isinstance(i, Mapping) and i.get("level") == "error"] + + +def next_step(terminal: Mapping[str, Any] | None, *, self_checked: bool = True) -> str: + """生成过程交出结局之后做什么。terminal 是最后那条结局操作（final / reply / error）；流结束时只有反问（ask）、没有 + 别的结局的给 None。 + + - reply：直接回答，这一轮到此为止；error：判失败；只有反问：等用户回答； + - final：有 error 级问题时，服务端已经自查过（收到过 check 的结论）就判失败——没过的问题已经交回模型在同一轮里改过 + 了，再整轮重来是再烧一次完整生成、改的还是同一批问题；没自查过（老服务端）才重来一次（retry）；模型说不运行 + （autorun=false：用户要的是流程本身、或者问了用户）只建不跑；其余运行。 + """ + if terminal is None: + return NEXT_ASKED + kind = terminal.get("op") + if kind == "reply": + return NEXT_REPLY + if kind != "final": + return NEXT_FAIL + if _blockers(terminal): + return NEXT_FAIL if self_checked else NEXT_RETRY + if terminal.get("autorun") is False: + return NEXT_READY + return NEXT_RUN + + +def blocked_reason(final: Mapping[str, Any]) -> str: + """没过检查的图重来时的原因（只有老服务端那条路用得到）。""" + return "生成的工作流没有通过校验：" + "；".join(_blockers(final)) + + +def fill_input(graph: Mapping[str, Any] | None, question: str, history: str = "") -> dict[str, Any]: + """用户的问题填进这张图的入口字段（同 chat.ts 的 fillInput）。 + + 入口没声明字段：{question, history}（下游可能用 input.question）。声明了：每个字段先填默认值（没有就空串），问题放进 + 名字像「问题」的那个，其次第一个必填的，再不然第一个；history（前几轮压成的文字）照样带上——图跑起来时模型得知道 + 「它们」指的是上一轮那批数据。 + """ + nodes = (graph or {}).get("nodes") or [] + entry = next((n for n in nodes if isinstance(n, Mapping) and n.get("type") == "input"), None) + config = ((entry or {}).get("data") or {}).get("config") or {} + fields = [f for f in (config.get("fields") or []) if isinstance(f, Mapping)] + if not fields: + return {"question": question, "history": history} + payload: dict[str, Any] = {"history": history} + for f in fields: + if f.get("name"): + payload[str(f["name"])] = f["default"] if f.get("default") is not None else "" + named = next((f for f in fields if QUESTION_FIELD.search(str(f.get("name") or ""))), None) + target = named or next((f for f in fields if f.get("required")), None) or fields[0] + if target.get("name"): + payload[str(target["name"])] = question + return payload + + +def review_reason(review: Mapping[str, Any]) -> str: + """复核判不可用时重来的原因：各条信号的说明，没有就用给用户看的那句说明。""" + details = [str(s.get("detail") or "") for s in review.get("signals") or [] if isinstance(s, Mapping)] + return "；".join(d for d in details if d) or str(review.get("note") or "") + + +def retry_after_review(review: Mapping[str, Any] | None, attempt: int) -> bool: + """复核之后要不要带着原图重来一次：复核说换个配置重跑大概率就好（retry）、答案不可信（severity=broken）、这一轮还没 + 试满（attempt 从 0 数，attempt + 1 < MAX_ATTEMPTS）。""" + if not review: + return False + return bool(review.get("retry")) and review.get("severity") == "broken" and attempt + 1 < MAX_ATTEMPTS + + +def retry_instruction(question: str, reason: str, has_base: bool) -> str: + """重来时交给助手的话。has_base：有上一次建出的图可以交回去改。""" + template = RETRY_WITH_BASE if has_base else RETRY_FROM_SCRATCH + return template.format(question=question, reason=reason) + + +def has_nodes(graph: Any) -> bool: + """一张图有没有节点（有才交回去改）。""" + return isinstance(graph, Mapping) and bool(graph.get("nodes")) + + +__all__ = ["MAX_ATTEMPTS", "NEXT_ASKED", "NEXT_FAIL", "NEXT_READY", "NEXT_REPLY", "NEXT_RETRY", "NEXT_RUN", "NEXT_STEPS", + "QUESTION_FIELD", "RETRY_FROM_SCRATCH", "RETRY_WITH_BASE", "blocked_reason", "fill_input", "has_nodes", + "next_step", "retry_after_review", "retry_instruction", "review_reason"] diff --git a/backend/app/providers/budget.py b/backend/app/providers/budget.py index 3d52402a..78b31f31 100644 --- a/backend/app/providers/budget.py +++ b/backend/app/providers/budget.py @@ -18,6 +18,7 @@ engine/nodes/llm.py（cite_fields 的抽取）、data/recipe_ai.py 各有一份 from __future__ import annotations from collections.abc import Awaitable, Callable, Hashable +from contextlib import aclosing from typing import Any #: 服务方嫌额度太大时报错里会出现的字样：超过模型的输出上限（DeepSeek chat 系只到 8K）， @@ -167,19 +168,24 @@ class BudgetFallback: return await (await self._current()).ainvoke(messages, *args, **kwargs) async def astream(self, messages: Any, *args: Any, **kwargs: Any): + # 里面那条流用 aclosing 包着：调用方读到一半不读了（助手读到 done、look 就关流），关这一层时里面那条也当场关掉、 + # 模型那一次调用当场收尾（LangChain 的回调这时把已经收到的用量交出来，providers/usage.py）。不包的话里面那条要等 + # 垃圾回收才关，收尾落在调用方算完账之后 used = await self._current() started = False try: - async for chunk in used.astream(messages, *args, **kwargs): - started = True - yield chunk + async with aclosing(used.astream(messages, *args, **kwargs)) as stream: + async for chunk in stream: + started = True + yield chunk return except Exception as e: if started or not self._can_fall_back(used) or not self._switch.rejected(e): raise self._switch.fall_back() - async for chunk in (await self._current()).astream(messages, *args, **kwargs): - yield chunk + async with aclosing((await self._current()).astream(messages, *args, **kwargs)) as stream: + async for chunk in stream: + yield chunk def _derived(self, make: Callable[[Any], Any]) -> BudgetFallback: """对大小两个模型做同一件事（绑工具、要结构化输出），包成共用开关的一层。小的还没建时，建好再做。""" diff --git a/backend/app/providers/usage.py b/backend/app/providers/usage.py new file mode 100644 index 00000000..37563e6b --- /dev/null +++ b/backend/app/providers/usage.py @@ -0,0 +1,204 @@ +"""模型用量计量：助手建图（含挑表、接着写、自查修正）和复核这两段花了多少（细案 01 做法 3）。 + +**为什么要。** 运行按 llm.end 事件记用量，问数据一轮里另外两块——助手建图、复核——一直不记：这一轮的花费是低报的， +评测也算不出按角色的成本。在每个调用点手写累计（生成、挑表、退回小额度重来、复核各一处）迟早漏掉一处，所以用 LangChain +的回调在模型这一层统一截：和 langchain_core.callbacks.get_usage_metadata_callback 同一个机制（一个登记过的上下文变量， +模型每次调用时把它挂成回调），按模型累计 input / output tokens 和调用次数。 + +**为什么不直接用 get_usage_metadata_callback。** 它是一个 with 块，块里设上下文变量、出块时按 token 复原。生成过程是一个 +async generator：with 块跨着 yield 的话，变量在两次 yield 之间留在调用方的上下文里——调用方这时去发起运行，运行的任务 +会继承它，运行里的调用就被重复计一遍；调用方中途不迭代了、在别的任务里收尾，复原还会报「token 不属于这个上下文」。 +这里只在每一步 __anext__ 期间设上变量、交出操作之前复原（metered）：生成过程里起的任务（心跳那一层每一步一个任务） +在创建时拷走上下文，照样计到；调用方自己的代码一行都不在计量里。 + +**金额。** 按 providers/catalog.estimate_cost 算，和运行的 llm.end 一个口径。估不出的调用记 unpriced_calls：模型不在价格 +目录里，或者用量没报全而目录价又不是 0。估不出的不当成 0 元，调用方据此写「≥」或「金额无法估算」。用量没报全有两种： +回复没带用量（有的网关流式不给）；读到一半就不读了——生成过程读到 done、look、ask 那一行就关掉模型的流（后面的闲话不要、 +也不再为它付钱），用量在流的末尾才报全（Anthropic 的输出 token 在最后一条，OpenAI 兼容系开了 stream_usage 也在最后）， +关掉的那一次只拿得到已经报了的部分。这一次照样计次数、计已报的 token，记成估不出，不悄悄少算。 + +模型名按发起调用时的参数取（invocation_params 的 model / model_name，就是接入里配的模型 id，价格目录按它查），取不到再 +看回复里的 model_name。 +""" +from __future__ import annotations + +import threading +from collections.abc import AsyncIterator, Iterator +from contextlib import contextmanager +from contextvars import ContextVar +from typing import Any +from uuid import UUID + +from langchain_core.callbacks import BaseCallbackHandler +from langchain_core.tracers.context import register_configure_hook + +#: 取不到模型名的调用记在这个名下 +UNKNOWN_MODEL = "unknown" + + +class Meter(BaseCallbackHandler): + """一次计量：按模型累计用量。线程安全（同步模型的回调可能在线程里来）。""" + + #: 同步模型在线程里调用时也要回调 + run_inline = True + + def __init__(self) -> None: + super().__init__() + self._lock = threading.Lock() + self._pending: dict[UUID, str] = {} + self._models: dict[str, dict[str, int]] = {} + + # ------------------------------------------------------------------ 回调 + + def _started(self, run_id: UUID, serialized: dict[str, Any] | None, kwargs: dict[str, Any]) -> None: + params = kwargs.get("invocation_params") or {} + model = params.get("model") or params.get("model_name") or params.get("model_id") + if not model and isinstance(serialized, dict): + model = (serialized.get("kwargs") or {}).get("model") or (serialized.get("kwargs") or {}).get("model_name") + with self._lock: + self._pending[run_id] = str(model) if model else "" + + def on_chat_model_start(self, serialized: dict[str, Any], messages: Any, *, run_id: UUID, + **kwargs: Any) -> None: + self._started(run_id, serialized, kwargs) + + def on_llm_start(self, serialized: dict[str, Any], prompts: Any, *, run_id: UUID, **kwargs: Any) -> None: + self._started(run_id, serialized, kwargs) + + def on_llm_end(self, response: Any, *, run_id: UUID, **kwargs: Any) -> None: + message = None + try: + message = response.generations[0][0].message + except (AttributeError, IndexError, TypeError): + pass + usage = getattr(message, "usage_metadata", None) or None + meta = getattr(message, "response_metadata", None) or {} + with self._lock: + model = self._pending.pop(run_id, "") or str(meta.get("model_name") or meta.get("model") or "") \ + or UNKNOWN_MODEL + row = self._models.setdefault(model, {"input_tokens": 0, "output_tokens": 0, "calls": 0, "unreported": 0}) + row["calls"] += 1 + if usage: + row["input_tokens"] += int(usage.get("input_tokens") or 0) + row["output_tokens"] += int(usage.get("output_tokens") or 0) + else: + row["unreported"] += 1 + + def on_llm_error(self, error: BaseException, *, run_id: UUID, **kwargs: Any) -> None: + """流读到一半停了（生成过程读到 done / look / ask 就关流，或者连接断了）：LangChain 照样交来已经收到的那几段。 + 有收到东西的照计一次、记已报的 token，算用量没报全；一段都没收到的（请求刚发出去就被拒了）不计。""" + from langchain_core.outputs import ChatGenerationChunk + + response = kwargs.get("response") + partial = None + try: + first = response.generations[0][0] if response is not None and response.generations[0] else None + partial = first.message if isinstance(first, ChatGenerationChunk) else None + except (AttributeError, IndexError, TypeError): + partial = None + with self._lock: + model = self._pending.pop(run_id, "") + if partial is None: + return + meta = getattr(partial, "response_metadata", None) or {} + model = model or str(meta.get("model_name") or meta.get("model") or "") or UNKNOWN_MODEL + row = self._models.setdefault(model, {"input_tokens": 0, "output_tokens": 0, "calls": 0, "unreported": 0}) + row["calls"] += 1 + row["unreported"] += 1 + reported = getattr(partial, "usage_metadata", None) or {} + row["input_tokens"] += int(reported.get("input_tokens") or 0) + row["output_tokens"] += int(reported.get("output_tokens") or 0) + + # ------------------------------------------------------------------ 合并、汇总 + + def merge(self, other: Meter) -> None: + """把另一次计量并进来（嵌套的计量出块时并进外层）。""" + with other._lock: + rows = {k: dict(v) for k, v in other._models.items()} + with self._lock: + for model, row in rows.items(): + mine = self._models.setdefault(model, dict.fromkeys(row, 0)) + for key, value in row.items(): + mine[key] = mine.get(key, 0) + value + + def summary(self) -> dict[str, Any]: + """{by_model: {模型: {input_tokens, output_tokens, calls, cost_usd, unpriced_calls}}, input_tokens, output_tokens, + calls, cost_usd, unpriced_calls}。cost_usd 只是估得出的那部分。""" + from app.providers.catalog import estimate_cost, find_pricing + + with self._lock: + rows = {k: dict(v) for k, v in self._models.items()} + by_model: dict[str, dict[str, Any]] = {} + for model, row in sorted(rows.items()): + pricing = find_pricing(model) + free = pricing is not None and not any(pricing.values()) + if pricing is None: + unpriced = row["calls"] + else: + unpriced = 0 if free else row["unreported"] + by_model[model] = {"input_tokens": row["input_tokens"], "output_tokens": row["output_tokens"], + "calls": row["calls"], + "cost_usd": round(estimate_cost(model, row["input_tokens"], row["output_tokens"]), 6), + "unpriced_calls": unpriced} + total = {key: sum(r[key] for r in by_model.values()) + for key in ("input_tokens", "output_tokens", "calls", "unpriced_calls")} + return {"by_model": by_model, **total, "cost_usd": round(sum(r["cost_usd"] for r in by_model.values()), 6)} + + +#: 计量挂在这个上下文变量上（inheritable：模型调用时 LangChain 按它把 Meter 挂成回调，子任务拷走上下文时一并带走） +_CURRENT: ContextVar[Meter | None] = ContextVar("agentlab_usage_meter", default=None) +register_configure_hook(_CURRENT, inheritable=True) + + +def meter() -> Meter: + """新的一次计量。用 metering（一段代码）或 metered（一个生成过程）把它挂上。""" + return Meter() + + +def current() -> Meter | None: + return _CURRENT.get() + + +@contextmanager +def metering(m: Meter) -> Iterator[Meter]: + """这一段代码里的模型调用计进 m。只给不跨 yield 的一段用（复核那一次调用）；生成过程用 metered。 + + 嵌套的：出块时并进外层的那一次，外层照样看得到。复原用 set 而不是按 token reset：在别的任务里收尾也不报错。""" + outer = _CURRENT.get() + _CURRENT.set(m) + try: + yield m + finally: + _CURRENT.set(outer) + if outer is not None and outer is not m: + outer.merge(m) + + +async def metered(source: AsyncIterator[Any], m: Meter) -> AsyncIterator[Any]: + """迭代一个生成过程，它的模型调用计进 m：每一步 __anext__ 期间挂上，交出操作之前复原（见模块开头）。 + + 开始时外层已经有一次计量的（评测把一整题的助手用量记成一份）：结束时把 m 并进去，同 metering。""" + enclosing = _CURRENT.get() + iterator = source.__aiter__() + try: + while True: + outer = _CURRENT.get() + _CURRENT.set(m) + try: + item = await iterator.__anext__() + except StopAsyncIteration: + return + finally: + _CURRENT.set(outer) + yield item + finally: + if enclosing is not None and enclosing is not m: + enclosing.merge(m) + + +def empty() -> dict[str, Any]: + """一次调用都没有时的汇总（同 Meter().summary()）。""" + return Meter().summary() + + +__all__ = ["Meter", "UNKNOWN_MODEL", "current", "empty", "meter", "metered", "metering"] diff --git a/backend/tests/fixtures/copilot_answer_ops_baseline.json b/backend/tests/fixtures/copilot_answer_ops_baseline.json new file mode 100644 index 00000000..f1de6562 --- /dev/null +++ b/backend/tests/fixtures/copilot_answer_ops_baseline.json @@ -0,0 +1,1030 @@ +{ + "graph_ok": [ + { + "op": "model", + "model": "mock-fast" + }, + { + "op": "plan", + "summary": "数一数" + }, + { + "op": "add_node", + "node": { + "id": "start", + "type": "input", + "label": "start", + "config": { + "fields": [ + { + "name": "question" + } + ] + } + } + }, + { + "op": "add_node", + "node": { + "id": "init", + "type": "transform", + "label": "init", + "config": { + "mode": "expression", + "expression": "[1, 2]", + "assign_to": "items" + } + } + }, + { + "op": "add_edge", + "edge": { + "source": "start", + "target": "init" + } + }, + { + "op": "add_node", + "node": { + "id": "lp", + "type": "loop", + "label": "lp", + "config": { + "mode": "while", + "condition": "len(vars.items) > 5", + "max_iterations": 2 + } + } + }, + { + "op": "add_edge", + "edge": { + "source": "init", + "target": "lp" + } + }, + { + "op": "add_node", + "node": { + "id": "body", + "type": "transform", + "label": "body", + "config": { + "mode": "expression", + "expression": "1" + } + } + }, + { + "op": "add_edge", + "edge": { + "source": "lp", + "target": "body", + "sourceHandle": "body" + } + }, + { + "op": "add_edge", + "edge": { + "source": "body", + "target": "lp" + } + }, + { + "op": "add_node", + "node": { + "id": "out", + "type": "output", + "label": "out", + "config": { + "fields": [ + { + "name": "结果", + "value": "{{ vars.items }}" + } + ] + } + } + }, + { + "op": "add_edge", + "edge": { + "source": "lp", + "target": "out", + "sourceHandle": "done" + } + }, + { + "op": "done", + "explanation": "数一数", + "run": true + }, + { + "op": "check", + "status": "passed", + "repaired": 0, + "warnings": [] + }, + { + "op": "final", + "graph": { + "nodes": [ + { + "id": "start", + "type": "input", + "position": { + "x": 80.0, + "y": 0.0 + }, + "data": { + "label": "start", + "description": "", + "config": { + "fields": [ + { + "name": "question" + } + ] + } + } + }, + { + "id": "init", + "type": "transform", + "position": { + "x": 414.0, + "y": 0.0 + }, + "data": { + "label": "init", + "description": "", + "config": { + "mode": "expression", + "expression": "[1, 2]", + "assign_to": "items" + } + } + }, + { + "id": "lp", + "type": "loop", + "position": { + "x": 748.0, + "y": 0.0 + }, + "data": { + "label": "lp", + "description": "", + "config": { + "mode": "while", + "condition": "len(vars.items) > 5", + "max_iterations": 2 + } + } + }, + { + "id": "body", + "type": "transform", + "position": { + "x": 1082.0, + "y": -66.0 + }, + "data": { + "label": "body", + "description": "", + "config": { + "mode": "expression", + "expression": "1" + } + } + }, + { + "id": "out", + "type": "output", + "position": { + "x": 1082.0, + "y": 66.0 + }, + "data": { + "label": "out", + "description": "", + "config": { + "fields": [ + { + "name": "结果", + "value": "{{ vars.items }}" + } + ] + } + } + } + ], + "edges": [ + { + "id": "start:->init", + "source": "start", + "target": "init", + "sourceHandle": null, + "targetHandle": null, + "label": "" + }, + { + "id": "init:->lp", + "source": "init", + "target": "lp", + "sourceHandle": null, + "targetHandle": null, + "label": "" + }, + { + "id": "lp:body->body", + "source": "lp", + "target": "body", + "sourceHandle": "body", + "targetHandle": null, + "label": "" + }, + { + "id": "body:->lp", + "source": "body", + "target": "lp", + "sourceHandle": null, + "targetHandle": null, + "label": "" + }, + { + "id": "lp:done->out", + "source": "lp", + "target": "out", + "sourceHandle": "done", + "targetHandle": null, + "label": "" + } + ], + "viewport": {}, + "defaults": {} + }, + "issues": [], + "explanation": "数一数", + "autorun": true, + "layout": { + "mode": "full", + "placed": [ + "start", + "init", + "lp", + "body", + "out" + ] + }, + "tool_changes": [] + } + ], + "graph_fixed": [ + { + "op": "model", + "model": "mock-fast" + }, + { + "op": "plan", + "summary": "数一数" + }, + { + "op": "add_node", + "node": { + "id": "start", + "type": "input", + "label": "start", + "config": { + "fields": [ + { + "name": "question" + } + ] + } + } + }, + { + "op": "add_node", + "node": { + "id": "init", + "type": "transform", + "label": "init", + "config": { + "mode": "expression", + "expression": "[1, 2]", + "assign_to": "items" + } + } + }, + { + "op": "add_edge", + "edge": { + "source": "start", + "target": "init" + } + }, + { + "op": "add_node", + "node": { + "id": "lp", + "type": "loop", + "label": "lp", + "config": { + "mode": "while", + "condition": "{{ vars.items | length }} > 5", + "max_iterations": 2 + } + } + }, + { + "op": "add_edge", + "edge": { + "source": "init", + "target": "lp" + } + }, + { + "op": "add_node", + "node": { + "id": "body", + "type": "transform", + "label": "body", + "config": { + "mode": "expression", + "expression": "1" + } + } + }, + { + "op": "add_edge", + "edge": { + "source": "lp", + "target": "body", + "sourceHandle": "body" + } + }, + { + "op": "add_edge", + "edge": { + "source": "body", + "target": "lp" + } + }, + { + "op": "add_node", + "node": { + "id": "out", + "type": "output", + "label": "out", + "config": { + "fields": [ + { + "name": "结果", + "value": "{{ vars.items }}" + } + ] + } + } + }, + { + "op": "add_edge", + "edge": { + "source": "lp", + "target": "out", + "sourceHandle": "done" + } + }, + { + "op": "done", + "explanation": "数一数", + "run": true + }, + { + "op": "check", + "status": "repairing", + "round": 1, + "issues": [ + { + "level": "error", + "node_id": "lp", + "edge_id": null, + "message": "循环条件有误：表达式中不支持 | 过滤器（那是模板的写法）：取长度请写 len(x)，转换大小写请写 upper(x) / lower(x)（原文：{{ vars.items | length }} > 5）", + "field": "condition", + "code": null, + "fix": null + } + ] + }, + { + "op": "update_node", + "id": "lp", + "config": { + "mode": "while", + "condition": "len(vars.items) > 5", + "max_iterations": 2 + } + }, + { + "op": "check", + "status": "passed", + "repaired": 1, + "warnings": [] + }, + { + "op": "final", + "graph": { + "nodes": [ + { + "id": "start", + "type": "input", + "position": { + "x": 80.0, + "y": 0.0 + }, + "data": { + "label": "start", + "description": "", + "config": { + "fields": [ + { + "name": "question" + } + ] + } + } + }, + { + "id": "init", + "type": "transform", + "position": { + "x": 414.0, + "y": 0.0 + }, + "data": { + "label": "init", + "description": "", + "config": { + "mode": "expression", + "expression": "[1, 2]", + "assign_to": "items" + } + } + }, + { + "id": "lp", + "type": "loop", + "position": { + "x": 748.0, + "y": 0.0 + }, + "data": { + "label": "lp", + "description": "", + "config": { + "mode": "while", + "condition": "len(vars.items) > 5", + "max_iterations": 2 + } + } + }, + { + "id": "body", + "type": "transform", + "position": { + "x": 1082.0, + "y": -66.0 + }, + "data": { + "label": "body", + "description": "", + "config": { + "mode": "expression", + "expression": "1" + } + } + }, + { + "id": "out", + "type": "output", + "position": { + "x": 1082.0, + "y": 66.0 + }, + "data": { + "label": "out", + "description": "", + "config": { + "fields": [ + { + "name": "结果", + "value": "{{ vars.items }}" + } + ] + } + } + } + ], + "edges": [ + { + "id": "start:->init", + "source": "start", + "target": "init", + "sourceHandle": null, + "targetHandle": null, + "label": "" + }, + { + "id": "init:->lp", + "source": "init", + "target": "lp", + "sourceHandle": null, + "targetHandle": null, + "label": "" + }, + { + "id": "lp:body->body", + "source": "lp", + "target": "body", + "sourceHandle": "body", + "targetHandle": null, + "label": "" + }, + { + "id": "body:->lp", + "source": "body", + "target": "lp", + "sourceHandle": null, + "targetHandle": null, + "label": "" + }, + { + "id": "lp:done->out", + "source": "lp", + "target": "out", + "sourceHandle": "done", + "targetHandle": null, + "label": "" + } + ], + "viewport": {}, + "defaults": {} + }, + "issues": [], + "explanation": "数一数", + "autorun": true, + "layout": { + "mode": "full", + "placed": [ + "start", + "init", + "lp", + "body", + "out" + ] + }, + "tool_changes": [] + } + ], + "check_failed": [ + { + "op": "model", + "model": "mock-fast" + }, + { + "op": "plan", + "summary": "数一数" + }, + { + "op": "add_node", + "node": { + "id": "start", + "type": "input", + "label": "start", + "config": { + "fields": [ + { + "name": "question" + } + ] + } + } + }, + { + "op": "add_node", + "node": { + "id": "init", + "type": "transform", + "label": "init", + "config": { + "mode": "expression", + "expression": "[1, 2]", + "assign_to": "items" + } + } + }, + { + "op": "add_edge", + "edge": { + "source": "start", + "target": "init" + } + }, + { + "op": "add_node", + "node": { + "id": "lp", + "type": "loop", + "label": "lp", + "config": { + "mode": "while", + "condition": "foo(vars.items)", + "max_iterations": 2 + } + } + }, + { + "op": "add_edge", + "edge": { + "source": "init", + "target": "lp" + } + }, + { + "op": "add_node", + "node": { + "id": "body", + "type": "transform", + "label": "body", + "config": { + "mode": "expression", + "expression": "1" + } + } + }, + { + "op": "add_edge", + "edge": { + "source": "lp", + "target": "body", + "sourceHandle": "body" + } + }, + { + "op": "add_edge", + "edge": { + "source": "body", + "target": "lp" + } + }, + { + "op": "add_node", + "node": { + "id": "out", + "type": "output", + "label": "out", + "config": { + "fields": [ + { + "name": "结果", + "value": "{{ vars.items }}" + } + ] + } + } + }, + { + "op": "add_edge", + "edge": { + "source": "lp", + "target": "out", + "sourceHandle": "done" + } + }, + { + "op": "done", + "explanation": "数一数", + "run": true + }, + { + "op": "check", + "status": "repairing", + "round": 1, + "issues": [ + { + "level": "error", + "node_id": "lp", + "edge_id": null, + "message": "循环条件有误：不支持函数 foo()，可用的函数：len、str、int、float、bool、abs、min、max、sum、round、sorted、any、all、lower、upper、contains、startswith、endswith、matches、get、cell（原文：foo(vars.items)）", + "field": "condition", + "code": null, + "fix": null + } + ] + }, + { + "op": "update_node", + "id": "lp", + "config": { + "mode": "while", + "condition": "bar(vars.items)", + "max_iterations": 2 + } + }, + { + "op": "check", + "status": "repairing", + "round": 2, + "issues": [ + { + "level": "error", + "node_id": "lp", + "edge_id": null, + "message": "循环条件有误：不支持函数 bar()，可用的函数：len、str、int、float、bool、abs、min、max、sum、round、sorted、any、all、lower、upper、contains、startswith、endswith、matches、get、cell（原文：bar(vars.items)）", + "field": "condition", + "code": null, + "fix": null + } + ] + }, + { + "op": "update_node", + "id": "lp", + "config": { + "mode": "while", + "condition": "bar(vars.items)", + "max_iterations": 2 + } + }, + { + "op": "check", + "status": "failed", + "issues": [ + { + "level": "error", + "node_id": "lp", + "edge_id": null, + "message": "循环条件有误：不支持函数 bar()，可用的函数：len、str、int、float、bool、abs、min、max、sum、round、sorted、any、all、lower、upper、contains、startswith、endswith、matches、get、cell（原文：bar(vars.items)）", + "field": "condition", + "code": null, + "fix": null + } + ], + "warnings": [] + }, + { + "op": "final", + "graph": { + "nodes": [ + { + "id": "start", + "type": "input", + "position": { + "x": 80.0, + "y": 0.0 + }, + "data": { + "label": "start", + "description": "", + "config": { + "fields": [ + { + "name": "question" + } + ] + } + } + }, + { + "id": "init", + "type": "transform", + "position": { + "x": 414.0, + "y": 0.0 + }, + "data": { + "label": "init", + "description": "", + "config": { + "mode": "expression", + "expression": "[1, 2]", + "assign_to": "items" + } + } + }, + { + "id": "lp", + "type": "loop", + "position": { + "x": 748.0, + "y": 0.0 + }, + "data": { + "label": "lp", + "description": "", + "config": { + "mode": "while", + "condition": "bar(vars.items)", + "max_iterations": 2 + } + } + }, + { + "id": "body", + "type": "transform", + "position": { + "x": 1082.0, + "y": -66.0 + }, + "data": { + "label": "body", + "description": "", + "config": { + "mode": "expression", + "expression": "1" + } + } + }, + { + "id": "out", + "type": "output", + "position": { + "x": 1082.0, + "y": 66.0 + }, + "data": { + "label": "out", + "description": "", + "config": { + "fields": [ + { + "name": "结果", + "value": "{{ vars.items }}" + } + ] + } + } + } + ], + "edges": [ + { + "id": "start:->init", + "source": "start", + "target": "init", + "sourceHandle": null, + "targetHandle": null, + "label": "" + }, + { + "id": "init:->lp", + "source": "init", + "target": "lp", + "sourceHandle": null, + "targetHandle": null, + "label": "" + }, + { + "id": "lp:body->body", + "source": "lp", + "target": "body", + "sourceHandle": "body", + "targetHandle": null, + "label": "" + }, + { + "id": "body:->lp", + "source": "body", + "target": "lp", + "sourceHandle": null, + "targetHandle": null, + "label": "" + }, + { + "id": "lp:done->out", + "source": "lp", + "target": "out", + "sourceHandle": "done", + "targetHandle": null, + "label": "" + } + ], + "viewport": {}, + "defaults": {} + }, + "issues": [ + { + "level": "error", + "node_id": "lp", + "edge_id": null, + "message": "循环条件有误：不支持函数 bar()，可用的函数：len、str、int、float、bool、abs、min、max、sum、round、sorted、any、all、lower、upper、contains、startswith、endswith、matches、get、cell（原文：bar(vars.items)）", + "field": "condition", + "code": null, + "fix": null + } + ], + "explanation": "数一数", + "autorun": false, + "layout": { + "mode": "full", + "placed": [ + "start", + "init", + "lp", + "body", + "out" + ] + }, + "tool_changes": [] + } + ], + "reply": [ + { + "op": "model", + "model": "mock-fast" + }, + { + "op": "reply", + "text": "上一轮已经查过了：一共 3 家门店。" + } + ], + "ask_only": [ + { + "op": "model", + "model": "mock-fast" + }, + { + "op": "plan", + "summary": "先问清楚" + }, + { + "op": "ask", + "question": "要看哪个月的？", + "options": [ + "九月", + "十月" + ] + }, + { + "op": "final", + "graph": { + "nodes": [], + "edges": [] + }, + "issues": [], + "explanation": "", + "autorun": false, + "layout": { + "mode": "keep", + "placed": [] + }, + "tool_changes": [], + "asked": { + "question": "要看哪个月的？", + "options": [ + "九月", + "十月" + ] + } + } + ], + "error": [ + { + "op": "model", + "model": "mock-fast" + }, + { + "op": "error", + "message": "助手本轮未完成：接入返回了 502", + "hint": "", + "detail": "RuntimeError: 接入返回了 502" + } + ], + "no_ops": [ + { + "op": "model", + "model": "mock-fast" + }, + { + "op": "error", + "message": "助手本轮未完成：模型没有给出任何修改，也没有说明原因", + "hint": "画布未修改。可以重试，或把需求描述得更具体", + "detail": "" + } + ], + "truncated": [ + { + "op": "model", + "model": "mock-fast" + }, + { + "op": "plan", + "summary": "想了很久" + }, + { + "op": "error", + "message": "助手本轮未完成：模型的输出超出了长度上限，修改没有写完", + "hint": "画布未修改。可以重试，或把需求拆成几步分别提出", + "detail": "finish_reason=length" + } + ] +} diff --git a/backend/tests/test_catalog_patch.py b/backend/tests/test_catalog_patch.py index e9c1a49d..15fb7c1b 100644 --- a/backend/tests/test_catalog_patch.py +++ b/backend/tests/test_catalog_patch.py @@ -268,7 +268,8 @@ async def _stream(monkeypatch, model, source, base_graph=None): async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: async with client.stream("POST", "/api/copilot/generate-stream", json=body) as r: async for line in r.aiter_lines(): - if line.startswith("data: "): + # 末尾那条 usage 是这一轮助手的用量（answer_ops），不是生成的操作 + if line.startswith("data: ") and not line.startswith('data: {"op": "usage"'): events.append(json.loads(line[6:])) return events diff --git a/backend/tests/test_conversations.py b/backend/tests/test_conversations.py index df0d1fb9..55f03d55 100644 --- a/backend/tests/test_conversations.py +++ b/backend/tests/test_conversations.py @@ -215,7 +215,8 @@ async def _stream_graph(client, monkeypatch, conv_id: str, lines: list[str]) -> }) as r: last = {} async for line in r.aiter_lines(): - if line.startswith("data: "): + # 末尾那条 usage 是这一轮助手的用量（answer_ops），不是生成的操作 + if line.startswith("data: ") and not line.startswith('data: {"op": "usage"'): last = json.loads(line[6:]) return last diff --git a/backend/tests/test_copilot_answer_ops.py b/backend/tests/test_copilot_answer_ops.py new file mode 100644 index 00000000..4bbd5566 --- /dev/null +++ b/backend/tests/test_copilot_answer_ops.py @@ -0,0 +1,165 @@ +"""生成过程抽成可直接调用的 answer_ops（细案 01 做法 2）：SSE 路由只包一层，同一个剧本下输出逐条不变。 + +评测和问数据页要走同一条生成路径：评测直接迭代 answer_ops 拿 model / look / check / final / reply / error 这些操作， +问数据页照旧收 SSE。这里守： + +- 几种结局（建图交付、直接回答、只问不改、自查改不好、模型出错、没给操作、输出被截断）的 SSE 输出，和抽出来之前 + 逐条一样（剧本见 SCENARIOS，抽出来之前录下的输出在 fixtures/copilot_answer_ops_baseline.json），末尾只多一条 usage； +- 直接迭代 answer_ops 和 SSE 收到的是同一串； +- 开流前的拒绝（模型没配好、限定的数据源都不在）照旧是 HTTP 错误，不进流。 + +剧本、需求都是虚构的。 +""" +from __future__ import annotations + +import json +from pathlib import Path + +import pytest +from httpx import ASGITransport, AsyncClient +from langchain_core.messages import AIMessageChunk + +from app.main import app + +BASELINE = Path(__file__).parent / "fixtures" / "copilot_answer_ops_baseline.json" +#: 和时间有关、每次都不一样的字段：比较前去掉 +_TIMING = ("elapsed_ms", "ms", "duration_ms") + + +@pytest.fixture +async def client(): + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + yield c + + +def _node(nid: str, ntype: str, config: dict) -> dict: + return {"op": "add_node", "node": {"id": nid, "type": ntype, "label": nid, "config": config}} + + +def _edge(a: str, b: str, **extra) -> dict: + return {"op": "add_edge", "edge": {"source": a, "target": b, **extra}} + + +def _loop_graph(condition: str) -> list[dict]: + return [ + {"op": "plan", "summary": "数一数"}, + _node("start", "input", {"fields": [{"name": "question"}]}), + _node("init", "transform", {"mode": "expression", "expression": "[1, 2]", "assign_to": "items"}), + _edge("start", "init"), + _node("lp", "loop", {"mode": "while", "condition": condition, "max_iterations": 2}), + _edge("init", "lp"), + _node("body", "transform", {"mode": "expression", "expression": "1"}), + _edge("lp", "body", sourceHandle="body"), + _edge("body", "lp"), + _node("out", "output", {"fields": [{"name": "结果", "value": "{{ vars.items }}"}]}), + _edge("lp", "out", sourceHandle="done"), + {"op": "done", "explanation": "数一数", "run": True}, + ] + + +def _fix(condition: str) -> list[dict]: + return [{"op": "plan", "summary": "改条件"}, + {"op": "update_node", "id": "lp", "config": {"mode": "while", "condition": condition, "max_iterations": 2}}, + {"op": "done", "explanation": "改好了"}] + + +def _chunks(ops: list) -> list[AIMessageChunk]: + return [op if isinstance(op, AIMessageChunk) else AIMessageChunk(content=json.dumps(op, ensure_ascii=False) + "\n") + for op in ops] + + +#: 剧本：(第一次调用吐的, 自查修正时吐的)。None 表示这一次调用直接抛错 +SCENARIOS: dict[str, tuple[list | None, list | None]] = { + "graph_ok": (_loop_graph("len(vars.items) > 5"), []), + "graph_fixed": (_loop_graph("{{ vars.items | length }} > 5"), _fix("len(vars.items) > 5")), + "check_failed": (_loop_graph("foo(vars.items)"), _fix("bar(vars.items)")), + "reply": ([{"op": "reply", "text": "上一轮已经查过了：一共 3 家门店。"}], []), + "ask_only": ([{"op": "plan", "summary": "先问清楚"}, + {"op": "ask", "question": "要看哪个月的？", "options": ["九月", "十月"]}], []), + "error": (None, []), + "no_ops": (["模型的随口注释，不是 JSON"], []), + "truncated": ([{"op": "plan", "summary": "想了很久"}, + AIMessageChunk(content="", response_metadata={"finish_reason": "length"})], []), +} + + +class _Scripted: + """照 SCENARIOS 吐：human 里有「上线前自查」时吐修正那一份。""" + + def __init__(self, first: list | None, fix: list | None) -> None: + self.first, self.fix = first, fix + + async def astream(self, messages): + human = next(text for role, text in messages if role == "human") + lines = self.fix if "上线前自查" in human else self.first + if lines is None: + raise RuntimeError("接入返回了 502") + for chunk in _chunks(lines): + yield chunk + + +def _use(monkeypatch, model) -> None: + import app.api.copilot as copilot + + async def _model(*_a, **_k): + return model, "mock-fast" + + monkeypatch.setattr(copilot, "get_chat_model", _model) + # 心跳和时间有关：拍子调大，剧本里不会有 + monkeypatch.setattr(copilot, "_HEARTBEAT_SECONDS", 3600.0) + + +def _clean(value): + if isinstance(value, dict): + return {k: _clean(v) for k, v in value.items() if k not in _TIMING} + if isinstance(value, list): + return [_clean(v) for v in value] + return value + + +async def _sse(client, body: dict) -> list[dict]: + out: list[dict] = [] + async with client.stream("POST", "/api/copilot/generate-stream", json=body) as r: + assert r.status_code == 200, await r.aread() + async for line in r.aiter_lines(): + if line.startswith("data: "): + out.append(json.loads(line[6:])) + return [_clean(op) for op in out if op.get("op") != "heartbeat"] + + +BODY = {"instruction": "数一数门店", "intent": "answer"} + + +@pytest.mark.parametrize("name", list(SCENARIOS)) +async def test_the_sse_output_is_unchanged_but_for_a_trailing_usage(client, monkeypatch, name): + _use(monkeypatch, _Scripted(*SCENARIOS[name])) + got = await _sse(client, BODY) + expected = json.loads(BASELINE.read_text(encoding="utf-8"))[name] + assert got[-1]["op"] == "usage", got[-1] + assert got[:-1] == expected + + +@pytest.mark.parametrize("name", ["graph_ok", "reply", "error"]) +async def test_iterating_answer_ops_directly_gives_the_same_ops(client, monkeypatch, name): + from app.api.copilot import GenerateIn, answer_ops + + _use(monkeypatch, _Scripted(*SCENARIOS[name])) + sse = await _sse(client, BODY) + direct = [_clean(op) async for op in answer_ops(GenerateIn(**BODY)) if op.get("op") != "heartbeat"] + assert direct == sse + + +async def test_refusals_before_the_stream_stay_http_errors(client, monkeypatch): + import app.api.copilot as copilot + from app.providers.factory import ProviderNotConfigured + + async def _unconfigured(*_a, **_k): + raise ProviderNotConfigured("provider 'gw' 需要填 base_url") + + monkeypatch.setattr(copilot, "get_chat_model", _unconfigured) + r = await client.post("/api/copilot/generate-stream", json=BODY) + assert r.status_code == 400 and "模型接入「gw」" in r.json()["detail"] + + _use(monkeypatch, _Scripted(*SCENARIOS["reply"])) + r = await client.post("/api/copilot/generate-stream", json={**BODY, "datasource_ids": ["no-such-source"]}) + assert r.status_code == 400 and r.json()["code"] == "datasource_scope_empty" diff --git a/backend/tests/test_copilot_evidence_guide.py b/backend/tests/test_copilot_evidence_guide.py index 9473da07..8a2a184e 100644 --- a/backend/tests/test_copilot_evidence_guide.py +++ b/backend/tests/test_copilot_evidence_guide.py @@ -213,7 +213,8 @@ async def test_guessed_columns_are_sent_back_and_fixed_before_running(sales, mon async with client.stream("POST", "/api/copilot/generate-stream", json={"instruction": "上周销售额多少", "intent": "answer"}) as r: async for line in r.aiter_lines(): - if line.startswith("data: "): + # 末尾那条 usage 是这一轮助手的用量（answer_ops），不是生成的操作 + if line.startswith("data: ") and not line.startswith('data: {"op": "usage"'): events.append(json.loads(line[6:])) checks = [e for e in events if e["op"] == "check"] assert [c["status"] for c in checks] == ["repairing", "passed"], checks diff --git a/backend/tests/test_copilot_look.py b/backend/tests/test_copilot_look.py index 86ade6a8..fdc96299 100644 --- a/backend/tests/test_copilot_look.py +++ b/backend/tests/test_copilot_look.py @@ -99,7 +99,8 @@ async def _events(client, monkeypatch, model, body: dict, path: str = "/api/copi async with client.stream("POST", path, json=body) as r: assert r.status_code == 200, await r.aread() async for line in r.aiter_lines(): - if line.startswith("data: "): + # 末尾那条 usage 是这一轮助手的用量（answer_ops），不是生成的操作 + if line.startswith("data: ") and not line.startswith('data: {"op": "usage"'): out.append(json.loads(line[6:])) return out diff --git a/backend/tests/test_copilot_merge.py b/backend/tests/test_copilot_merge.py index 9f8c97da..e425652f 100644 --- a/backend/tests/test_copilot_merge.py +++ b/backend/tests/test_copilot_merge.py @@ -110,7 +110,8 @@ async def _stream(client, monkeypatch, model, body: dict) -> list[dict]: async with client.stream("POST", "/api/copilot/generate-stream", json=body) as r: assert r.status_code == 200, await r.aread() async for line in r.aiter_lines(): - if line.startswith("data: "): + # 末尾那条 usage 是这一轮助手的用量（answer_ops），不是生成的操作 + if line.startswith("data: ") and not line.startswith('data: {"op": "usage"'): events.append(json.loads(line[6:])) return events diff --git a/backend/tests/test_copilot_patch_layout.py b/backend/tests/test_copilot_patch_layout.py index 27525fbd..647d9e01 100644 --- a/backend/tests/test_copilot_patch_layout.py +++ b/backend/tests/test_copilot_patch_layout.py @@ -64,7 +64,8 @@ async def _events(client, monkeypatch, ops: list[dict], base: dict | None) -> li out: list[dict] = [] async with client.stream("POST", "/api/copilot/generate-stream", json=body) as r: async for line in r.aiter_lines(): - if line.startswith("data: "): + # 末尾那条 usage 是这一轮助手的用量（answer_ops），不是生成的操作 + if line.startswith("data: ") and not line.startswith('data: {"op": "usage"'): out.append(json.loads(line[6:])) return out diff --git a/backend/tests/test_copilot_prompt_evidence.py b/backend/tests/test_copilot_prompt_evidence.py index fe4250e1..ee6f9beb 100644 --- a/backend/tests/test_copilot_prompt_evidence.py +++ b/backend/tests/test_copilot_prompt_evidence.py @@ -59,7 +59,8 @@ async def generate(client, monkeypatch, model: Scripted) -> list[dict]: events: list[dict] = [] async with client.stream("POST", "/api/copilot/generate-stream", json={"instruction": "写一份订单周报"}) as r: async for line in r.aiter_lines(): - if line.startswith("data: "): + # 末尾那条 usage 是这一轮助手的用量（answer_ops），不是生成的操作 + if line.startswith("data: ") and not line.startswith('data: {"op": "usage"'): events.append(json.loads(line[6:])) return events diff --git a/backend/tests/test_copilot_scope.py b/backend/tests/test_copilot_scope.py index 0a6f9322..48d738a5 100644 --- a/backend/tests/test_copilot_scope.py +++ b/backend/tests/test_copilot_scope.py @@ -96,7 +96,8 @@ async def _stream(client, monkeypatch, model, body: dict) -> tuple[int, list[dic if r.status_code != 200: return r.status_code, [json.loads(await r.aread())] async for line in r.aiter_lines(): - if line.startswith("data: "): + # 末尾那条 usage 是这一轮助手的用量（answer_ops），不是生成的操作 + if line.startswith("data: ") and not line.startswith('data: {"op": "usage"'): events.append(json.loads(line[6:])) return 200, events diff --git a/backend/tests/test_copilot_selfcheck.py b/backend/tests/test_copilot_selfcheck.py index 6e56e193..add5b950 100644 --- a/backend/tests/test_copilot_selfcheck.py +++ b/backend/tests/test_copilot_selfcheck.py @@ -85,7 +85,8 @@ async def _generate(client, monkeypatch, model: _Scripted) -> list[dict]: async with client.stream("POST", "/api/copilot/generate-stream", json={"instruction": "数一数"}) as r: async for line in r.aiter_lines(): - if line.startswith("data: "): + # 末尾那条 usage 是这一轮助手的用量（answer_ops），不是生成的操作 + if line.startswith("data: ") and not line.startswith('data: {"op": "usage"'): events.append(json.loads(line[6:])) return events diff --git a/backend/tests/test_copilot_sqlcheck.py b/backend/tests/test_copilot_sqlcheck.py index 2263ccc3..2dd9f809 100644 --- a/backend/tests/test_copilot_sqlcheck.py +++ b/backend/tests/test_copilot_sqlcheck.py @@ -195,7 +195,8 @@ async def _stream(monkeypatch, model, source): json={"instruction": "上周订单金额多少", "intent": "answer", "datasource_ids": [source]}) as r: async for line in r.aiter_lines(): - if line.startswith("data: "): + # 末尾那条 usage 是这一轮助手的用量（answer_ops），不是生成的操作 + if line.startswith("data: ") and not line.startswith('data: {"op": "usage"'): events.append(json.loads(line[6:])) return events diff --git a/backend/tests/test_copilot_truncation.py b/backend/tests/test_copilot_truncation.py index c81a73e0..b3f0733c 100644 --- a/backend/tests/test_copilot_truncation.py +++ b/backend/tests/test_copilot_truncation.py @@ -87,7 +87,8 @@ async def _events(client, monkeypatch, model, base: dict | None = BASE) -> list[ async with client.stream("POST", "/api/copilot/generate-stream", json={"instruction": "并行算同比和环比", "base_graph": base}) as r: async for line in r.aiter_lines(): - if line.startswith("data: "): + # 末尾那条 usage 是这一轮助手的用量（answer_ops），不是生成的操作 + if line.startswith("data: ") and not line.startswith('data: {"op": "usage"'): out.append(json.loads(line[6:])) return out @@ -263,7 +264,8 @@ async def test_a_rejected_budget_falls_back_to_the_old_one(client, monkeypatch, async with client.stream("POST", "/api/copilot/generate-stream", json={"instruction": "加一支同比", "base_graph": BASE}) as r: async for line in r.aiter_lines(): - if line.startswith("data: "): + # 末尾那条 usage 是这一轮助手的用量（answer_ops），不是生成的操作 + if line.startswith("data: ") and not line.startswith('data: {"op": "usage"'): events.append(json.loads(line[6:])) assert 8192 in asked assert big.calls == 1 and small.calls == 1 @@ -284,7 +286,8 @@ async def test_other_rejections_are_not_retried(client, monkeypatch): async with client.stream("POST", "/api/copilot/generate-stream", json={"instruction": "改一下", "base_graph": BASE}) as r: async for line in r.aiter_lines(): - if line.startswith("data: "): + # 末尾那条 usage 是这一轮助手的用量（answer_ops），不是生成的操作 + if line.startswith("data: ") and not line.startswith('data: {"op": "usage"'): events.append(json.loads(line[6:])) assert small.calls == 0 assert events[-1]["op"] == "error" diff --git a/backend/tests/test_usage_meter.py b/backend/tests/test_usage_meter.py new file mode 100644 index 00000000..c69259f6 --- /dev/null +++ b/backend/tests/test_usage_meter.py @@ -0,0 +1,283 @@ +"""用量计量（providers/usage.py；细案 01 做法 3）：助手建图（含挑表、接着写、自查修正）和复核的用量也记。 + +守的是： +- 按模型累计 tokens 和调用次数，金额按价格目录算；目录外的模型、回复没带用量（而目录价又不是 0）的调用记 unpriced_calls， + 不当成 0 元； +- metered 只计生成过程自己的调用（含它在一步里起的任务）；调用方在两次 yield 之间做的事（发起运行）不计——运行的用量 + 另有 llm.end，包进来就重复了；嵌套时内层并进外层； +- answer_ops 末尾的 usage 记下这一轮建图的每一次调用（起草、自查修正各一次）；复核的结果带 usage。 + +模型都是剧本，不调真接口；价格用目录里登记过的模型 id，名字之外都是虚构的。 +""" +from __future__ import annotations + +import asyncio +import json +from typing import Any + +import pytest +from httpx import ASGITransport, AsyncClient +from langchain_core.language_models.chat_models import BaseChatModel +from langchain_core.messages import AIMessage, AIMessageChunk +from langchain_core.outputs import ChatGeneration, ChatGenerationChunk, ChatResult + +from app.main import app +from app.providers import usage +from app.providers.catalog import estimate_cost + +PRICED = "claude-opus-5" # 价格目录里有 +UNPRICED = "made-up-model-x" # 价格目录里没有 +FREE = "mock-fast" # 目录价是 0 + + +class Scripted(BaseChatModel): + """真的 LangChain 模型：回调照常触发。每次调用吐 replies 里的下一份（用完重复最后一份），用量按 tokens 报； + report=False 时回复不带用量（有的网关流式不给）。""" + + model: str = PRICED + replies: list[list[str]] = [["好的"]] + tokens: tuple[int, int] = (1000, 200) + report: bool = True + calls: int = 0 + + @property + def _llm_type(self) -> str: + return "scripted" + + @property + def _identifying_params(self) -> dict[str, Any]: + return {"model": self.model} + + def _next(self) -> list[str]: + lines = self.replies[min(self.calls, len(self.replies) - 1)] + self.calls += 1 + return lines + + def _usage(self) -> dict[str, int] | None: + if not self.report: + return None + inp, out = self.tokens + return {"input_tokens": inp, "output_tokens": out, "total_tokens": inp + out} + + def _generate(self, messages, stop=None, run_manager=None, **kwargs) -> ChatResult: + message = AIMessage(content="\n".join(self._next())) + if (meta := self._usage()) is not None: + message.usage_metadata = meta + return ChatResult(generations=[ChatGeneration(message=message)]) + + async def _astream(self, messages, stop=None, run_manager=None, **kwargs): + """像 Anthropic 那样报用量：第一段带输入 token，流末尾单独一段带输出 token。""" + lines = self._next() + inp, out = self.tokens + for i, line in enumerate(lines): + meta = {"input_tokens": inp, "output_tokens": 0, "total_tokens": inp} if self.report and i == 0 else None + yield ChatGenerationChunk(message=AIMessageChunk(content=line + "\n", **({"usage_metadata": meta} if meta + else {}))) + if self.report: + yield ChatGenerationChunk(message=AIMessageChunk(content="", usage_metadata={ + "input_tokens": 0, "output_tokens": out, "total_tokens": out})) + + +# -------------------------------------------------------------------------- +# 计量本身 +# -------------------------------------------------------------------------- + + +async def test_metering_counts_by_model_and_prices_from_the_catalog(): + m = usage.meter() + with usage.metering(m): + await Scripted(model=PRICED).ainvoke("你好") + await Scripted(model=PRICED).ainvoke("再来") + await Scripted(model=UNPRICED).ainvoke("你好") + await Scripted(model=PRICED + "-silent", report=False).ainvoke("你好") + await Scripted(model=FREE, report=False).ainvoke("你好") + got = m.summary() + assert got["by_model"][PRICED] == {"input_tokens": 2000, "output_tokens": 400, "calls": 2, + "cost_usd": round(estimate_cost(PRICED, 2000, 400), 6), "unpriced_calls": 0} + assert got["by_model"][UNPRICED]["unpriced_calls"] == 1 and got["by_model"][UNPRICED]["cost_usd"] == 0 + # 目录外的模型名（多了后缀）照样算估不出；目录价是 0 的本地模型没报用量也不算估不出 + assert got["by_model"][PRICED + "-silent"]["unpriced_calls"] == 1 + assert got["by_model"][FREE]["unpriced_calls"] == 0 + assert got["calls"] == 5 and got["unpriced_calls"] == 2 + assert got["cost_usd"] == pytest.approx(estimate_cost(PRICED, 2000, 400)) + + +async def test_outside_the_meter_nothing_is_counted(): + m = usage.meter() + with usage.metering(m): + pass + await Scripted().ainvoke("你好") + assert m.summary()["calls"] == 0 + assert usage.current() is None + + +async def test_metered_counts_the_generator_and_its_tasks_but_not_the_caller(): + """生成过程里的调用（含它起的任务）计进来；调用方在两次 yield 之间发起的「运行」不计。""" + m = usage.meter() + leaked: list[asyncio.Task] = [] + + async def generation(): + await Scripted(model=PRICED).ainvoke("起草") + # 心跳那一层每一步起一个任务：任务在步里创建，拷走的上下文带着计量 + await asyncio.ensure_future(Scripted(model=PRICED).ainvoke("挑表")) + yield "first" + await Scripted(model=PRICED).ainvoke("自查修正") + yield "second" + + seen = [] + async for item in usage.metered(generation(), m): + seen.append(item) + # 调用方拿到一条操作后自己调模型（相当于发起运行）：它的上下文里没有计量 + await Scripted(model=UNPRICED).ainvoke("运行里的调用") + leaked.append(asyncio.ensure_future(Scripted(model=UNPRICED).ainvoke("运行任务"))) + await asyncio.gather(*leaked) + got = m.summary() + assert seen == ["first", "second"] + assert got["by_model"] == {PRICED: got["by_model"][PRICED]} and got["calls"] == 3 + assert usage.current() is None + + +async def test_nested_meters_fold_into_the_outer_one(): + outer, inner = usage.meter(), usage.meter() + + async def generation(): + await Scripted(model=PRICED).ainvoke("起草") + yield 1 + + with usage.metering(outer): + await Scripted(model=PRICED).ainvoke("外层自己的") + async for _ in usage.metered(generation(), inner): + pass + with usage.metering(usage.meter()): + await Scripted(model=PRICED).ainvoke("复核") + assert inner.summary()["calls"] == 1 + assert outer.summary()["calls"] == 3 + + +# -------------------------------------------------------------------------- +# answer_ops 末尾的 usage、复核的 usage +# -------------------------------------------------------------------------- + + +def _ops(*ops: dict) -> list[str]: + return [json.dumps(o, ensure_ascii=False) for o in ops] + + +GRAPH = _ops( + {"op": "plan", "summary": "数一数"}, + {"op": "add_node", "node": {"id": "start", "type": "input", "label": "输入", + "config": {"fields": [{"name": "question"}]}}}, + {"op": "add_node", "node": {"id": "lp", "type": "loop", "label": "循环", + "config": {"mode": "while", "condition": "foo(vars.x)", "max_iterations": 2}}}, + {"op": "add_edge", "edge": {"source": "start", "target": "lp"}}, + {"op": "add_node", "node": {"id": "body", "type": "transform", "label": "体", + "config": {"mode": "expression", "expression": "1"}}}, + {"op": "add_edge", "edge": {"source": "lp", "target": "body", "sourceHandle": "body"}}, + {"op": "add_edge", "edge": {"source": "body", "target": "lp"}}, + {"op": "add_node", "node": {"id": "out", "type": "output", "label": "成果", + "config": {"fields": [{"name": "r", "value": "1"}]}}}, + {"op": "add_edge", "edge": {"source": "lp", "target": "out", "sourceHandle": "done"}}, + {"op": "done", "explanation": "数一数", "run": True}, +) +FIX = _ops({"op": "update_node", "id": "lp", "config": {"mode": "while", "condition": "len(vars.x) > 1", + "max_iterations": 2}}, + {"op": "done", "explanation": "改好了"}) + + +@pytest.fixture +async def client(): + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + yield c + + +async def test_answer_ops_ends_with_the_usage_of_every_build_call(client, monkeypatch): + """起草、自查修正各一次。生成过程读到 done 就关流（后面的闲话不要），流末尾才报的输出 token 拿不到：这两次照样计次数、 + 计已报的输入 token，记成估不出（unpriced_calls），金额只是估得出的那部分——不悄悄少算。""" + import app.api.copilot as copilot + + model = Scripted(model=PRICED, replies=[GRAPH, FIX], tokens=(3000, 500)) + + async def _model(*_a, **_k): + return model, PRICED + + monkeypatch.setattr(copilot, "get_chat_model", _model) + events = [] + async with client.stream("POST", "/api/copilot/generate-stream", json={"instruction": "数一数", "intent": "answer"}) as r: + async for line in r.aiter_lines(): + if line.startswith("data: "): + events.append(json.loads(line[6:])) + assert [e["op"] for e in events if e["op"] == "check"] and events[-2]["op"] == "final" + spent = events[-1] + assert spent["op"] == "usage", spent + assert model.calls == 2, spent # 起草一次、自查修正一次 + assert spent["by_model"][PRICED]["calls"] == 2, spent + assert spent["by_model"][PRICED]["input_tokens"] == 6000 and spent["by_model"][PRICED]["output_tokens"] == 0 + assert spent["cost_usd"] == pytest.approx(estimate_cost(PRICED, 6000, 0), abs=1e-6) + assert spent["unpriced_calls"] == 2, "关流前没报全的两次记成估不出" + + +async def test_a_build_on_an_unpriced_model_says_it_cannot_be_priced(client, monkeypatch): + import app.api.copilot as copilot + + model = Scripted(model=UNPRICED, replies=[_ops({"op": "reply", "text": "已经查过了"})]) + + async def _model(*_a, **_k): + return model, UNPRICED + + monkeypatch.setattr(copilot, "get_chat_model", _model) + events = [] + async with client.stream("POST", "/api/copilot/generate-stream", json={"instruction": "刚才那个数是多少"}) as r: + async for line in r.aiter_lines(): + if line.startswith("data: "): + events.append(json.loads(line[6:])) + assert [e["op"] for e in events] == ["model", "reply", "usage"] + assert events[-1]["calls"] == 1 and events[-1]["unpriced_calls"] == 1 and events[-1]["cost_usd"] == 0 + + +async def test_a_stream_read_to_the_end_is_fully_priced(): + """读完的流（直接回答那一行之后模型自己停了）：输入、输出 token 都报全了，金额估得出。""" + m = usage.meter() + + async def generation(): + async for _ in Scripted(model=PRICED, tokens=(400, 60)).astream("问"): + pass + yield 1 + + async for _ in usage.metered(generation(), m): + pass + row = m.summary()["by_model"][PRICED] + assert (row["input_tokens"], row["output_tokens"], row["unpriced_calls"]) == (400, 60, 0) + assert row["cost_usd"] == round(estimate_cost(PRICED, 400, 60), 6) + + +async def test_the_review_carries_its_own_usage(client, monkeypatch): + import app.api.copilot as copilot + from app.core.events import EventType + from app.db.base import SessionLocal + from app.db.models import Run, RunEvent + + reply = json.dumps({"note": "agent 用满了步数，结论不完整。", "answer": "", "retry": True}, ensure_ascii=False) + model = Scripted(model=PRICED, replies=[[reply]], tokens=(800, 100)) + + async def _model(*_a, **_k): + return model, PRICED + + monkeypatch.setattr(copilot, "get_chat_model", _model) + async with SessionLocal() as session: + run = Run(status="succeeded", output={"answer": "大概是 3 家"}, graph={}) + session.add(run) + await session.flush() + session.add(RunEvent(run_id=run.id, seq=1, type=EventType.LOG, node_id="a", ts=0.0, + data={"level": "warn", "code": "step_limit", "message": "用满了步数"})) + await session.commit() + run_id = run.id + body = (await client.post("/api/copilot/review", json={"run_id": run_id, "question": "有几家门店"})).json() + assert body["usage"]["calls"] == 1 and body["usage"]["by_model"][PRICED]["input_tokens"] == 800 + # 干净的运行不调模型：usage 全 0 + async with SessionLocal() as session: + clean = Run(status="succeeded", output={"answer": "3 家"}, graph={}) + session.add(clean) + await session.commit() + clean_id = clean.id + body = (await client.post("/api/copilot/review", json={"run_id": clean_id, "question": "有几家门店"})).json() + assert body["verdict"] == "ok" and body["usage"]["calls"] == 0 and body["usage"]["by_model"] == {}