问数据整轮放到服务端跑(后端):一轮是一个后台任务,建图、运行、复核、按复核重来、落库都在服务端,页面断了也会跑完

为什么:一轮问数据以前是浏览器在推(store/chat.ts 的 plan / launch / settle):网络一抖建图那条流就断、建了一半的图丢掉;
发起运行、复核、重来、落库都要页面在,页面一关这一轮就停在 running,15 分钟后才被判成断了(总方案 §0 的调查结论)。

做了什么:
- 流程和评测共用一份:eval/answer.Driver 里的流程抽到 chat/drive.TurnDriver,差异写成子类给的几项(生成的请求、目录视图、
  改图、入口、发起参数、停在审批上怎么办、复核等多久)。评测的 Driver 成了它的子类,行为和测试一字不变(test_eval_* 照跑)。
- chat/job.py:问数据的一轮是 core/job_runtime 上的后台任务(不占数据源的队,并发归「交互」类,第 2 波再给上限和优先级)。
  停在审批上照旧等人(评测是取消);复核最多等 25 秒,晚到的那次用量照记(review.late);按复核自动重来一次;用户点的
  重试本轮、重新运行、放宽步数、跑一下、继续运行、接着收尾都是同一轮里的一次尝试,上一次的结局归档进 meta.attempts。
- 事件日志(新表 chat_turn_events,models_chat.py,不进运行的封存范围):每一步按 seq 落库再叫醒订阅方,订阅方按 seq
  回库里读——回放和实时是同一段代码;心跳一拍也回库里看一次,第 2 波多进程时别的进程写的最迟一拍送到。
- 接口 api/chat_turns.py:POST /api/chat/turns(202 带 turn_id)、GET /{id}、GET /{id}/stream?after=(SSE:先回放、再推
  实时、带心跳、结束补 stream.end)、POST /{id}/cancel。挂 chat_scope,每个接口按这一轮所属的对话核对归属;访问密钥调不了。
- conversation_turns 那一行由服务端写(chat/record.py:状态、运行 id、答案、复核、meta 各项和页面以前写的一一对应);失败
  只记原因(meta.cause),说法留在界面层按 lib/explain 讲。TurnOut 带这一轮的任务(job),列表判「还在跑」先看任务。
- 使用者的问数据运行规则(白名单、自己的记忆范围、遮罩、只读)抽到 chat/launch.user_chat_run,POST /api/runs 和服务端的
  一轮共用,说法一字不变。
- 服务重启:还在建图的标中断(可重试本轮);已发起运行的接着等那次运行、照常复核收尾;和调度循环同一个开关,拷来的数据
  目录启动时不替人收尾,页面打开对话时再请服务端接着收尾(mode=resume)。老轮次停在建图中的同样标中断。
- 删对话时先停掉它正在跑的几轮(取消各自的运行),任务和事件日志一并删。

测试:tests/test_chat_turns.py——一轮跑完不依赖页面、直接回答与反问、开流前被拒、实时流回放后接实时与按 seq 重连(补齐、
不重复)、思考合并、建图中与运行中停止、重启时建图中标中断与已发起运行的接着收尾、开关关着时等人来接、等审批后收尾、
按复核重来、用户重试归档、一段对话一次一轮、继续运行被拒照 /runs 的说法、删对话停掉这一轮、使用者的归属与遮罩只读、
访问密钥调不了、驳回与失败与发起被拒的原因、晚到的复核、跑一下与重新运行不重建。去掉几处关键实现回跑,相应的测试都会失败。
This commit is contained in:
ThearchyHelios
2026-10-18 16:05:53 +08:00
parent 3f3e39e4d5
commit d18a496ef9
15 changed files with 3020 additions and 332 deletions
+116
View File
@@ -0,0 +1,116 @@
"""问数据的一轮:发起、看、停(docs/design/orchestration-round/00-总方案.md §1;做法在 app/chat/job.py)。
- POST /api/chat/turns:发起一轮,202 {turn_id, turn}。新问带 conversation_id、question(可选 datasource_ids 限定数据源);对已有
的一轮重试、重新运行、放宽步数、跑一下、继续运行、接着收尾带 retry_of 和 mode(MODES),重试时 datasource_ids 给了就换范围
(空数组是不限),不给沿用这一轮的。同一段对话里有一轮正在跑(停在审批上的不算)时 409 chat_turn_busy;继续运行被拒照
POST /api/runs/{id}/continue 的说法(409,带那边的机器码)。
- GET /api/chat/turns/{id}:这一轮的任务现在怎样(status、stage、run_id、last_seq、attempt_seq……),连同那一行(turn)。老轮次
没有任务时 job 为 null。
- GET /api/chat/turns/{id}/stream?after=<seq>:SSE。先回放 seq > after 的,再推实时;隔一拍没有新事件发一行注释心跳;这一轮不在
进行中、也没有新的了,补一条 stream.end 再关(chat/log.py)。断了按收到的最后一个 seq 重连,补齐、不重复。
- POST /api/chat/turns/{id}/cancel:停止(取消生成、取消运行,照现有的取消规则)。已经结束的 409 chat_turn_finished。
权限照问数据的规矩:路由挂 chat_scope(使用者要管理员开了问数据),每个接口再按这一轮所属的对话核对归属(对话按人隔离,
conversations.turn_access / ensure_conversation_access);访问密钥调不了(不在 auth/keys.KEY_ROUTES 里)。
"""
from __future__ import annotations
from typing import Any
from fastapi import APIRouter, Depends, Header, HTTPException, Query
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
from sqlalchemy.ext.asyncio import AsyncSession
from app.api.coded import CodedHTTPException
from app.api.deps import actor_of
from app.db.base import get_session
router = APIRouter(prefix="/api/chat/turns", tags=["chat"])
class TurnStartIn(BaseModel):
conversation_id: str = Field(min_length=1)
question: str = ""
#: 这一问只查这几个数据源(id 或名字)。不给:新问不限,重试沿用这一轮的;空数组:不限
datasource_ids: list[str] | None = None
#: 对已有的哪一轮(重试、重新运行、继续运行……)
retry_of: str | None = None
#: ask | retry | rerun | max_steps | run | continue | resume(app/chat/job.MODES)。不给:新问是 ask,带 retry_of 是 retry
mode: str | None = None
#: 放宽到几步(mode=max_steps)
max_steps: int | None = Field(default=None, ge=1, le=10_000)
def _refused(e: Any) -> HTTPException:
if e.code:
return CodedHTTPException(e.status_code, e.message, e.code)
return HTTPException(e.status_code, e.message)
@router.post("", status_code=202)
async def start_turn(payload: TurnStartIn, session: AsyncSession = Depends(get_session),
x_actor: str | None = Header(default=None)) -> dict[str, Any]:
from app.api.conversations import _get, turn_access, turn_out
from app.chat import job as chat_job
retry_of = None
if payload.retry_of:
conversation, retry_of = await turn_access(session, payload.retry_of)
if conversation.id != payload.conversation_id:
raise HTTPException(404, "该轮对话不属于这段对话")
else:
conversation = await _get(session, payload.conversation_id)
try:
turn_id = await chat_job.start(session, conversation=conversation, question=payload.question,
datasource_ids=payload.datasource_ids, actor=actor_of(x_actor), mode=payload.mode,
retry_of=retry_of, max_steps=payload.max_steps)
except chat_job.TurnRefused as e:
raise _refused(e) from e
from app.db.base import SessionLocal
from app.db.models import ConversationTurn
async with SessionLocal() as fresh:
row = await fresh.get(ConversationTurn, turn_id)
turn = (await turn_out(fresh, row)).model_dump(mode="json") if row is not None else None
return {"turn_id": turn_id, "turn": turn}
@router.get("/{turn_id}")
async def get_turn(turn_id: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
from app.api.conversations import turn_access, turn_out
from app.chat import job as chat_job
_, row = await turn_access(session, turn_id)
return {"id": row.id, "conversation_id": row.conversation_id, "job": await chat_job.state(session, row.id),
"turn": (await turn_out(session, row)).model_dump(mode="json")}
@router.get("/{turn_id}/stream")
async def stream_turn(turn_id: str, after: int = Query(default=0, ge=0),
session: AsyncSession = Depends(get_session)) -> StreamingResponse:
from app.api.conversations import turn_access
from app.chat import job as chat_job
from app.chat import log as turn_log
await turn_access(session, turn_id)
await session.close() # 流可能开很久:核对完就把这次请求的会话还回去
async def active() -> str | None:
return await chat_job.stream_status(turn_id)
return StreamingResponse(turn_log.stream(turn_id, after, active=active), media_type="text/event-stream",
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})
@router.post("/{turn_id}/cancel")
async def cancel_turn(turn_id: str, session: AsyncSession = Depends(get_session),
x_actor: str | None = Header(default=None)) -> dict[str, Any]:
from app.api.conversations import turn_access
from app.chat import job as chat_job
await turn_access(session, turn_id)
await session.close()
if not await chat_job.cancel(turn_id, actor_of(x_actor)):
raise CodedHTTPException(409, "这一轮已经结束,无需停止", chat_job.TURN_FINISHED)
return {"ok": True}
+69 -18
View File
@@ -5,10 +5,11 @@
同时服务两件事:给用户看的历史列表,和给模型看的上下文(`_history_text`
被 copilot 拿去拼进 prompt)。
轮次的写入由前端驱动,分两步:开轮(只有问题)→ 回填(图、run、答案)。
让 copilot 或 runner 自己去写会更"自动",但建图失败、运行失败、用户中途
取消这三种半途而废的情况它们各自只知道一半,最后还是要在三个地方各接一次。
前端本来就完整地知道这一轮走到哪了。
问数据的轮次由服务端写(docs/design/orchestration-round §1):一轮是一个后台任务(app/chat/job.py),建图、运行、复核、
按复核重来、落库都在服务端,这一行的状态、运行 id、答案、复核、meta 由它写(app/chat/record.py 说明了每一项);页面只发起
(POST /api/chat/turns)、只看。以前这一行是页面写的(开轮 → 回填),页面一关就没人收尾。开轮、回填这两个接口留着:画布助手的
对话和老页面还在用,回填也用来改标题这类事,但问数据的一轮不再依赖它们。每一轮带着它的任务现在怎样(TurnOut.job),列表判
「还在跑」先看任务,不再只靠 15 分钟没动静来猜。
对话按人隔离(D-d):新建时记下建立人(created_by = actor_of)。开登录后 user 角色只列得出、读得到、改得动、删得掉
自己建的对话;created_by 为空的老对话(关登录时建的)只有 builder 以上看得到。应用编辑器的助手对话(kind=app)另要
@@ -99,6 +100,8 @@ class TurnOut(BaseModel):
created_at: Any = None
#: 最后一次落库。列表判「建图断了」按它算(BUILD_STALE),前端用同一只钟
updated_at: Any = None
#: 这一轮在服务端的任务现在怎样(app/chat/job.job_view:status、stage、last_seq、attempt_seq……)。老轮次没有任务,为 None
job: dict[str, Any] | None = None
model_config = {"from_attributes": True}
@@ -153,17 +156,35 @@ def turn_run_id(turn: Any) -> str | None:
return turn.run_id or None
def _turn_out(turn: ConversationTurn) -> TurnOut:
def _turn_out(turn: ConversationTurn, job: Any = None) -> TurnOut:
from app.chat.job import job_view
out = TurnOut.model_validate(turn)
out.meta = turn_meta(turn)
out.job = job_view(job) if job is not None else None
return out
def _detail(row: Conversation, turns: list[ConversationTurn]) -> ConversationDetail:
async def _jobs(session: AsyncSession, turn_ids: list[str]) -> dict[str, Any]:
"""这几轮在服务端的任务(一次查完)。"""
from app.db.models_chat import ChatTurnJob
if not turn_ids:
return {}
rows = (await session.execute(select(ChatTurnJob).where(ChatTurnJob.id.in_(turn_ids)))).scalars()
return {j.id: j for j in rows}
async def turn_out(session: AsyncSession, turn: ConversationTurn) -> TurnOut:
"""一轮(带着它的任务),同 GET /api/conversations/{id} 里的一轮:服务端跑的一轮收尾时把它放进 end 事件。"""
return _turn_out(turn, (await _jobs(session, [turn.id])).get(turn.id))
def _detail(row: Conversation, turns: list[ConversationTurn], jobs: dict[str, Any] | None = None) -> ConversationDetail:
base = ConversationOut.model_validate(row)
return ConversationDetail(
**base.model_dump(),
turns=[_turn_out(t) for t in turns],
turns=[_turn_out(t, (jobs or {}).get(t.id)) for t in turns],
)
@@ -174,10 +195,14 @@ BUILD_STALE = timedelta(minutes=15)
def last_status(
turn: Any, run_status: str | None, pending: bool, now: datetime | None = None, end_reason: str | None = None,
job_stage: str | None = None,
) -> str:
"""一轮停在哪。轮次状态只有 running / done / error 三种,细分要看运行和元数据。
轮次状态是前端写的:页面关掉、服务重启时它停在 running,这时要以运行的状态为准;
job_stage:这一轮在服务端的任务还在跑时它在哪一步(app/chat/job.py;没有任务、任务不在跑时为 None)。任务在跑就是在跑
(停在审批上的是 waiting),不按 15 分钟没动静去猜——建图慢的一轮,那一行要到建完才改。
没有任务的老轮次是页面写的:页面关掉、服务重启时它停在 running,这时要以运行的状态为准;
运行也没有的,是断在了建图阶段。已经收尾的轮次以轮次自己为准——它写下的就是
用户看到的结局(取消、服务重启挂起在 error 里分开记)。
@@ -186,6 +211,8 @@ def last_status(
"""
meta = turn_meta(turn) or {}
outcome = meta.get("outcome")
if job_stage is not None and turn.status == "running":
return "waiting" if job_stage == "waiting" else "running"
if turn.status == "done":
return "done"
if turn.status == "error":
@@ -258,12 +285,17 @@ async def _statuses(
select(Approval.run_id).where(
Approval.run_id.in_(stopped), Approval.status == "pending")
)).scalars())
from app.core.job_runtime import ACTIVE
jobs = await _jobs(session, [t.id for t in turns.values()])
now = utcnow()
out: dict[str, tuple[str, str | None]] = {}
for cid, turn in turns.items():
run_id = run_of[cid] if run_of[cid] in runs else None
job = jobs.get(turn.id)
stage = (job.stage or "build") if job is not None and job.status in ACTIVE else None
out[cid] = (last_status(turn, runs.get(run_id or ""), run_id in waiting, now,
end_reason=ended.get(run_id or "")),
end_reason=ended.get(run_id or ""), job_stage=stage),
run_id)
return out
@@ -437,7 +469,7 @@ async def conversation_detail(session: AsyncSession, row: Conversation) -> Conve
.where(ConversationTurn.conversation_id == conversation_id)
.order_by(ConversationTurn.seq)
)).scalars())
out = _detail(row, turns)
out = _detail(row, turns, await _jobs(session, [t.id for t in turns]))
out.turn_count = len(turns)
out.last_question = turns[-1].question if turns else ""
if turns:
@@ -465,9 +497,14 @@ async def patch_conversation(
async def delete_conversation(
conversation_id: str, session: AsyncSession = Depends(get_session)
) -> None:
from app.chat import job as chat_job
row = await _get(session, conversation_id)
# 正在服务端跑的几轮先停下(取消各自的运行):不然删掉之后它还在跑、还在计费,收尾时又找不到那一行
await chat_job.stop_conversation(conversation_id)
# 显式删 turns 而不是只靠 ondelete=CASCADE:SQLite 的外键级联要 PRAGMA
# 打开才生效(base.py 里是打开的),但这条路径不该依赖一个连接级的开关
# 打开才生效(base.py 里是打开的),但这条路径不该依赖一个连接级的开关。任务和事件日志同理
await chat_job.forget(session, conversation_id)
await session.execute(
delete(ConversationTurn).where(ConversationTurn.conversation_id == conversation_id)
)
@@ -487,25 +524,39 @@ async def create_turn(
session: AsyncSession = Depends(get_session),
) -> TurnOut:
conversation = await _get(session, conversation_id)
turn = await new_turn(session, conversation, payload.question)
await session.commit()
await session.refresh(turn)
return _turn_out(turn)
async def new_turn(session: AsyncSession, conversation: Conversation, question: str) -> ConversationTurn:
"""开一轮(status=running,只有问题),不提交:开轮接口和服务端发起的一轮(app/chat/job.start)共用。"""
next_seq = int((await session.execute(
select(func.coalesce(func.max(ConversationTurn.seq), -1))
.where(ConversationTurn.conversation_id == conversation_id)
.where(ConversationTurn.conversation_id == conversation.id)
)).scalar_one()) + 1
turn = ConversationTurn(
conversation_id=conversation_id, seq=next_seq,
question=payload.question, status="running",
conversation_id=conversation.id, seq=next_seq,
question=question, status="running",
)
session.add(turn)
conversation.last_active_at = utcnow()
# 第一轮的问题就是这次对话的标题。用户改过名就不要再覆盖回去
if next_seq == 0 and conversation.title in ("", "新对话"):
conversation.title = title_from(payload.question)
conversation.title = title_from(question)
await session.flush()
return turn
await session.commit()
await session.refresh(turn)
return _turn_out(turn)
async def turn_access(session: AsyncSession, turn_id: str) -> tuple[Conversation, ConversationTurn]:
"""这一轮这个人能不能用(按它所属的对话判,规则同 _get:对话按人隔离)。不存在 404,不能用 403。"""
turn = await session.get(ConversationTurn, turn_id)
if turn is None:
raise HTTPException(404, "该轮对话不存在,可能已被删除")
return await _get(session, turn.conversation_id), turn
@router.patch("/{conversation_id}/turns/{turn_id}", response_model=TurnOut)
+3 -1
View File
@@ -349,12 +349,14 @@ async def _chat_scope(conn: HTTPConnection) -> None:
ensure_role("user")
path = conn.url.path.rstrip("/")
method = conn.scope.get("method", "GET")
if user_can_chat() and (path.startswith("/api/conversations") or (method, path) in _CHAT_ROUTES):
if user_can_chat() and (path.startswith(_CHAT_PREFIXES) or (method, path) in _CHAT_ROUTES):
return
raise CodedHTTPException(403, "没有权限:问数据和助手需要搭建者及以上的角色,或由管理员为使用者开启问数据",
CHAT_FORBIDDEN)
#: user 角色开了问数据时能用的整组接口:对话,和服务端跑的问数据一轮(发起、看、停,app/chat/job.py)。归属在接口里按对话判
_CHAT_PREFIXES = ("/api/conversations", "/api/chat/turns")
#: user 角色开了问数据时能用的其余接口:问数据的生成流、读当前助手模型、问数据页选数据源用的数据源名单
#: (GET /api/datasources/names,只有名字、说明和种类,不含连接信息)
_CHAT_ROUTES = frozenset({("POST", "/api/copilot/generate-stream"), ("GET", "/api/copilot/model"),
+7 -7
View File
@@ -112,7 +112,6 @@ async def start_run(
"""发起规则全在 engine/launch.py(定时任务到点发起也用它),这里只把拒绝换回 HTTP 响应。"""
from app.api.deps import restricted
from app.engine.origin import CLIENT_ORIGINS
from app.tools.datasource import MASK_RESULTS_OPTION, READ_ONLY_OPTION
origin = payload.origin or "api"
if origin not in CLIENT_ORIGINS:
@@ -133,15 +132,16 @@ async def start_run(
# 数据源查询一律只读,写语句不交给审批(Agent 的 approval 是它自己写在图里的)
if origin != "chat" or payload.run_class != "exploratory" or payload.workflow_id or payload.version is not None:
raise HTTPException(403, "没有权限:使用者只能发起问数据的运行,工作流的运行需要搭建者及以上的角色")
from app.auth.chat_policy import SCOPE_REFUSED, chat_graph_problem, chat_memory_scope
from app.auth.chat_policy import SCOPE_REFUSED
from app.chat.launch import ChatRunRefused, user_chat_run
if problem := chat_graph_problem(payload.graph):
raise HTTPException(403, problem)
# 白名单、记忆范围、遮罩与只读这几条和服务端跑的一轮问数据(chat/job.py)共用 chat/launch.user_chat_run
try:
memory_scope, options = user_chat_run(payload.graph, actor)
except ChatRunRefused as e:
raise HTTPException(403, str(e)) from e
if payload.memory_scope or payload.collection:
raise HTTPException(403, SCOPE_REFUSED)
# 记忆固定为这位使用者自己的范围、知识库取设置里的默认值:记忆、知识库的内容不经遮罩(见 chat_policy 开头)
memory_scope = chat_memory_scope(actor)
options = {MASK_RESULTS_OPTION: True, READ_ONLY_OPTION: True}
try:
return await launch_run(
session,
+414
View File
@@ -0,0 +1,414 @@
"""一轮问数据的流程:问数据(服务端跑整轮,app/chat/job.py)和评测(app/eval/answer.py)共用这一份。
**为什么只有一份。** 评测量的必须是用户拿到的那一轮(细案 01 做法 1),以前问数据的一轮是浏览器在推(frontend/src/store/chat.ts
的 plan / launch / settle),评测照着在服务端另串了一遍(eval/answer.Driver)。现在问数据也整轮放到服务端(总方案 §1),两边各写
一份流程迟早走样,所以把流程抽到这里,差异写成子类给的几项(总方案上一轮 X「一种说法」):
answer_ops → final.next(chat/turn.next_step)→ launch_run → 等终态 → review_of_run → 要的话带原图重来一次
- 生成:generate_in 给这一次生成的请求(问数据带对话、限定的数据源;评测带变体的助手模型),catalog_view 给读哪一份数据目录
(评测按变体,问数据是当前目录)。每一条操作都先交给 on_op(问数据写进这一轮的事件日志,评测不管),思考和心跳不进结局判断。
- 结局:reply 直接回答、asked 反问、ready 只建不跑、build_failed 建图失败、retry 带着原因重来(次数照 MAX_ATTEMPTS)、run 运行。
每次重来前调 on_retry(问数据记一条「重来」,把换下来的运行的用量并进这一轮)。
- 运行:prepare_graph 改图(评测按变体写进写作模型)、run_input 填入口(问数据带前几轮的问答,评测单轮)、launch_args 给来源、
选项、发起人(遮罩、只读、目录视图照各自现有的规则),发起之后调 on_launched。
- 终态:停在审批上交给 on_interrupted——评测取消(不代批),问数据照旧等人(这是两边唯一的流程差别);被驳回的记 rejected,
失败、被停止记 run_failed。
- 复核:_review(评测一直等、复核自己出错照实记一条失败的复核;问数据最多等 25 秒,见 chat/job.py)。给了
retry_instruction 且 retry 打开时带着这次运行的图重来一次。
drive 的几个参数只有问数据用:graph(不建图、直接运行这张图:重新运行本轮、放宽步数、跑一下)、run_id(不发起、接着等这次
运行:继续运行、服务重启后接着收尾)、retry(复核之后要不要自动重来:用户点的「重新运行」照以前不自动重来)、instruction /
base_graph(用户点「重试本轮」时带着原因和上一次的图重建)。不给就是评测那一轮,和抽出来之前逐条一样。
时限、停止由调用方管:这里随时把正在跑的那次运行记在 turn.current_run,调用方超时、停止时据此取消它。
"""
from __future__ import annotations
import asyncio
import copy
import logging
import time
from collections.abc import Awaitable, Callable, Mapping
from dataclasses import dataclass, field
from typing import Any
from sqlalchemy import func, select
logger = logging.getLogger(__name__)
#: 取消一次运行之后最多再等它收尾多久
CANCEL_GRACE_S = 15.0
@dataclass
class Turn:
"""一轮作答的结果(也是进行中的状态:current_run、phase 随时更新)。"""
path: str | None = None
run_id: str | None = None
#: 这一轮发起过的全部运行(重来的话两次),失败的也在
runs: list[str] = field(default_factory=list)
attempts: int = 0
#: 每次自动重来的原因(和 attempts 一一对应):没过检查、复核说不可信、生成过程一个操作都没给
retries: list[str] = field(default_factory=list)
reason: str | None = None
#: 助手调用失败(生成过程交出 error、开流前就被拒):评测据此止损
copilot_failed: bool = False
timed_out: bool = False
#: 直接回答的原文、反问的问题
reply: str | None = None
asked: dict[str, Any] | None = None
review: dict[str, Any] | None = None
looks: list[dict[str, Any]] = field(default_factory=list)
#: 助手这一轮用的模型 id(model 操作)
assistant_model: str | None = None
#: {assistant: usage 汇总(生成,含重来), review: usage 汇总}
usage: dict[str, Any] = field(default_factory=dict)
timings: dict[str, int] = field(default_factory=dict)
#: 写作模型按变体换了,但节点自己点了模型、没被覆盖的节点 id(评测)
own_models: list[str] = field(default_factory=list)
#: 正在跑的那次运行(调用方超时、停止时取消它)
current_run: str | None = None
phase: str | None = None
def to_dict(self) -> dict[str, Any]:
return {"path": self.path, "run_id": self.run_id, "runs": list(self.runs), "attempts": self.attempts,
"retries": list(self.retries), "reason": self.reason, "copilot_failed": self.copilot_failed, "timed_out": self.timed_out,
"reply": self.reply, "asked": self.asked, "review": self.review, "looks": list(self.looks),
"assistant_model": self.assistant_model, "usage": copy.deepcopy(self.usage),
"timings": dict(self.timings), "own_models": list(self.own_models)}
#: 用量汇总里相加的计数(providers/usage 的 summary):unfinished_calls 是没做完的调用(被取消、超时,token 只计了已报的)
_COUNTED = ("input_tokens", "output_tokens", "calls", "unpriced_calls", "unfinished_calls")
def add_usage(into: dict[str, Any] | None, more: Mapping[str, Any] | None) -> dict[str, Any]:
"""两份 providers/usage 汇总相加(by_model 逐个模型加)。"""
out = copy.deepcopy(into) if into else {"by_model": {}, **dict.fromkeys(_COUNTED, 0), "cost_usd": 0.0}
if not more:
return out
for key in _COUNTED:
out[key] = int(out.get(key) or 0) + int(more.get(key) or 0)
out["cost_usd"] = round(float(out.get("cost_usd") or 0) + float(more.get("cost_usd") or 0), 6)
for model, row in (more.get("by_model") or {}).items():
mine = out["by_model"].setdefault(model, {**dict.fromkeys(_COUNTED, 0), "cost_usd": 0.0})
for key in _COUNTED:
mine[key] = int(mine.get(key) or 0) + int(row.get(key) or 0)
mine["cost_usd"] = round(float(mine.get("cost_usd") or 0) + float(row.get("cost_usd") or 0), 6)
return out
class TurnDriver:
"""一轮问数据。drive() 跑完一轮,结果在 turn 上;on_phase(阶段) 在每一步开始时调。子类给差异(见模块说明)。
跑完之后这几项给调用方收尾用(问数据据此写 conversation_turns):terminal 最后一次生成的结局操作;refusal 开流前的拒绝
(HTTPException,带 detail / code);launch_refused 发起被拒(LaunchRefused);graph 最后一次建出的图;self_checked 服务端
这一次自查过;last_run 最后一次运行的库里记录;last_review 最后一次复核的完整结果。"""
def __init__(self, question: str, *, datasource_ids: list[str],
on_phase: Callable[[str], Awaitable[None]] | None = None,
provider: str | None = None, model: str | None = None) -> None:
self.question = question
self.datasource_ids = list(datasource_ids)
self.on_phase = on_phase
#: 助手模型(生成、复核都用它):评测按变体覆盖,问数据用设置里的
self.provider = provider
self.model = model
self.turn = Turn()
self.terminal: dict[str, Any] | None = None
self.refusal: Exception | None = None
self.launch_refused: Exception | None = None
self.graph: dict[str, Any] | None = None
self.self_checked = False
self.last_run: Any = None
self.last_review: dict[str, Any] | None = None
async def _phase(self, phase: str) -> None:
self.turn.phase = phase
if self.on_phase is not None:
try:
await self.on_phase(phase)
except Exception: # noqa: BLE001 - 写进度失败不挡作答
logger.exception("一轮作答:记进度出错")
# ------------------------------------------------------------------ 子类给的差异
def generate_in(self, instruction: str, base_graph: dict[str, Any] | None) -> Any:
"""这一次生成的请求(api/copilot.GenerateIn)。"""
from app.api.copilot import GenerateIn
return GenerateIn(instruction=instruction, base_graph=base_graph, intent="answer", provider=self.provider,
model=self.model, datasource_ids=self.datasource_ids or None)
def catalog_view(self) -> Any:
"""这一轮读哪一份数据目录(data/catalog_view.py),None 是当前目录。"""
return None
async def on_op(self, op: dict[str, Any]) -> None:
"""生成过程的每一条操作(含思考、心跳),收下之前先交给这里。"""
async def on_retry(self, reason: str, *, kind: str, base_graph: dict[str, Any] | None) -> None:
"""要自动重来一次了(kind:empty 生成过程一个操作都没给、blocked 没过检查、review 复核判不可用)。"""
def prepare_graph(self, graph: dict[str, Any]) -> dict[str, Any]:
"""发起之前改一下图(评测按变体写进写作模型)。"""
return graph
async def run_input(self, session: Any, graph: dict[str, Any]) -> dict[str, Any]:
"""入口字段怎么填(chat/turn.fill_input)。缺省单轮:前几轮的问答为空。"""
from app.chat import turn as policy
return policy.fill_input(graph, self.question, "")
async def launch_args(self, session: Any, graph: dict[str, Any]) -> dict[str, Any]:
"""launch_run 的其余参数:来源、选项、发起人……可以抛 LaunchRefused(照发起被拒记)。"""
raise NotImplementedError
async def on_launched(self, run: Any) -> None:
"""运行发起了。"""
async def on_interrupted(self, run_id: str, row: Any) -> Any | None:
"""运行停在断点上(审批、服务重启挂起):交回接着等到的运行记录,或者记好路径、交回 None。"""
raise NotImplementedError
# ------------------------------------------------------------------ 生成
async def _generate(self, instruction: str, base_graph: dict[str, Any] | None) -> tuple[list[dict[str, Any]], str | None]:
"""跑一次生成过程,收下全部操作。开流前就被拒(HTTPException)交回原因。"""
from fastapi import HTTPException
from app.api.copilot import answer_ops
payload = self.generate_in(instruction, base_graph)
ops: list[dict[str, Any]] = []
try:
async for op in answer_ops(payload, catalog_view=self.catalog_view()):
await self.on_op(op)
if op.get("op") in ("thinking", "heartbeat"):
continue
ops.append(op)
except HTTPException as e:
self.refusal = e
return ops, str(e.detail)
return ops, None
def _take(self, ops: list[dict[str, Any]]) -> dict[str, Any] | None:
"""收下这一次生成的用量、查看记录、模型;交回结局(final / reply / error),只有反问的为 None。"""
terminal = None
for op in ops:
kind = op.get("op")
if kind == "usage":
self.turn.usage["assistant"] = add_usage(self.turn.usage.get("assistant"),
{k: v for k, v in op.items() if k != "op"})
elif kind == "look":
self.turn.looks.append({k: op.get(k) for k in ("id", "by", "need", "status") if k in op})
elif kind == "model":
self.turn.assistant_model = str(op.get("model") or "") or None
elif kind in ("final", "reply", "error"):
terminal = op
return terminal
# ------------------------------------------------------------------ 运行
async def _launch(self, graph: dict[str, Any], payload: dict[str, Any] | None = None) -> str | None:
from app.db.base import SessionLocal
from app.engine.launch import LaunchRefused, launch_run
graph = self.prepare_graph(graph)
try:
async with SessionLocal() as session:
inputs = payload if payload is not None else await self.run_input(session, graph)
run = await launch_run(session, graph=graph, input_payload=inputs,
**(await self.launch_args(session, graph)))
except LaunchRefused as e:
self.launch_refused = e
self.turn.path, self.turn.reason = "run_failed", f"运行没有发起:{e.message}"
return None
self.turn.runs.append(run.id)
self.turn.run_id = self.turn.current_run = run.id
await self.on_launched(run)
return run.id
async def _wait(self, run_id: str) -> Any:
"""等运行到终态(或停在审批上)。交回库里的运行。"""
from app.db.base import SessionLocal
from app.db.models import Run
from app.engine.runner import run_manager
while True:
await run_manager.wait_idle(run_id, timeout=5.0)
async with SessionLocal() as session:
row = await session.get(Run, run_id)
if row is None or row.status in ("succeeded", "failed", "cancelled", "interrupted"):
# 封存在终态之后:等执行任务收完尾(含封存),评分只读封存范围
await run_manager.wait_idle(run_id, timeout=CANCEL_GRACE_S)
async with SessionLocal() as session:
return await session.get(Run, run_id)
await asyncio.sleep(0.2)
async def _pending_approvals(self, run_id: str) -> int:
from app.db.base import SessionLocal
from app.db.models import Approval
async with SessionLocal() as session:
return int(await session.scalar(select(func.count()).select_from(Approval).where(
Approval.run_id == run_id, Approval.status.in_(("pending", "answered")))) or 0)
async def _follow(self, run_id: str) -> Any | None:
"""等这次运行到终态。成功结束的交回运行;别的结局记好路径和原因,交回 None。"""
from app.engine.runner import END_REJECTED
t = self.turn
row = await self._wait(run_id)
t.current_run = None
self.last_run = row
usage = dict((row.usage or {}) if row is not None else {})
t.timings["run_wall_ms"] = t.timings.get("run_wall_ms", 0) + int(usage.get("wall_ms") or 0)
t.timings["run_active_ms"] = t.timings.get("run_active_ms", 0) + int(
usage.get("active_ms") or usage.get("duration_ms") or 0)
if row is None:
t.path, t.reason = "run_failed", "运行记录不见了"
return None
if row.status == "interrupted":
return await self.on_interrupted(run_id, row)
if row.status == "cancelled" and row.end_reason == END_REJECTED:
t.path, t.reason = "rejected", "运行被驳回(人的决定,不算运行失败)"
return None
if row.status != "succeeded":
t.path = "run_failed"
t.reason = f"运行{'失败' if row.status == 'failed' else '被停止'}:{row.error or '没有记下原因'}"
return None
return row
async def _run(self, graph: dict[str, Any], payload: dict[str, Any] | None = None) -> Any | None:
"""发起、等到终态。成功结束的交回运行;别的结局记好路径和原因,交回 None。"""
run_id = await self._launch(graph, payload)
if run_id is None:
return None
return await self._follow(run_id)
# ------------------------------------------------------------------ 一轮
async def drive(self, *, graph: dict[str, Any] | None = None, run_id: str | None = None, retry: bool = True,
instruction: str | None = None, base_graph: dict[str, Any] | None = None) -> Turn:
from app.chat import turn as policy
t = self.turn
started = time.monotonic()
instruction = instruction or self.question
# 不建图、直接运行这张图(graph),或者不发起、接着等这次运行(run_id):只有第一圈这样,重来还是照常建图
direct, follow = graph, run_id
try:
while True:
if direct is None and follow is None:
await self._phase("build")
building = time.monotonic()
ops, refused = await self._generate(instruction, base_graph)
t.timings["build_ms"] = t.timings.get("build_ms", 0) + int((time.monotonic() - building) * 1000)
terminal = self._take(ops)
self.terminal = terminal
if refused is not None:
t.path, t.reason, t.copilot_failed = "build_failed", f"助手没有开始:{refused}", True
return t
if terminal is None:
asked = next((o for o in reversed(ops) if o.get("op") == "ask"), None)
t.path, t.asked = "asked", {k: asked.get(k) for k in ("question", "options")} if asked else None
return t
kind = terminal.get("op")
if kind == "error":
# 生成过程一个操作都没给(只写了思考):按重来一次的规矩带着原因重来(chat/turn.next_step)
step = terminal.get("next") or policy.next_step(terminal)
if step == policy.NEXT_RETRY and t.attempts + 1 < policy.MAX_ATTEMPTS:
reason = policy.error_reason(terminal)
t.attempts += 1
t.retries.append(reason)
instruction = policy.retry_instruction(self.question, reason, policy.has_nodes(base_graph))
await self.on_retry(reason, kind="empty", base_graph=base_graph)
continue
t.path, t.reason, t.copilot_failed = "build_failed", str(terminal.get("message") or "助手本轮未完成"), True
return t
if kind == "reply":
t.path, t.reply = "reply", str(terminal.get("text") or "")
return t
if terminal.get("asked"):
self.graph = terminal.get("graph") or None
t.path, t.asked = "asked", dict(terminal["asked"])
return t
self.self_checked = any(o.get("op") == "check" for o in ops)
step = terminal.get("next") or policy.next_step(terminal, self_checked=self.self_checked)
built = terminal.get("graph") or {}
self.graph = built
if step == policy.NEXT_RETRY:
if t.attempts + 1 < policy.MAX_ATTEMPTS:
t.attempts += 1
t.retries.append(policy.blocked_reason(terminal))
instruction = policy.retry_instruction(self.question, policy.blocked_reason(terminal),
policy.has_nodes(built))
base_graph = built if policy.has_nodes(built) else None
await self.on_retry(t.retries[-1], kind="blocked", base_graph=base_graph)
continue
t.path, t.reason = "build_failed", policy.blocked_reason(terminal)
return t
if step == policy.NEXT_FAIL:
errors = [str(i.get("message") or "") for i in terminal.get("issues") or []
if isinstance(i, Mapping) and i.get("level") == "error"]
t.path, t.reason = "build_failed", "生成的工作流没有通过校验:" + ";".join(errors)
return t
if step == policy.NEXT_READY:
t.path, t.reason = "ready", "助手只建了图、没有运行(它判断用户要的是流程本身)"
return t
direct = built
await self._phase("run")
if follow is not None:
t.runs.append(follow)
t.run_id = t.current_run = follow
row = await self._follow(follow)
else:
row = await self._run(direct)
direct = follow = None
if row is None:
return t
await self._phase("review")
reviewing = time.monotonic()
self.last_review = None
review = await self._review(row)
spent = int((time.monotonic() - reviewing) * 1000)
t.timings["review_ms"] = t.timings.get("review_ms", 0) + spent
if review is not None:
self.last_review = review
t.review = {k: v for k, v in review.items() if k != "usage"}
t.usage["review"] = add_usage(t.usage.get("review"), review.get("usage"))
if retry and review.get("retry_instruction"):
t.attempts += 1
t.retries.append(policy.review_reason(review) or "复核判答案不可用")
instruction = str(review["retry_instruction"])
base_graph = row.graph if policy.has_nodes(row.graph) else None
await self.on_retry(t.retries[-1], kind="review", base_graph=base_graph)
continue
t.path = "ran"
return t
finally:
t.timings["total_ms"] = int((time.monotonic() - started) * 1000)
async def _review(self, run: Any) -> dict[str, Any] | None:
"""复核(review_of_run,问数据和评测同一个函数)。复核自己出错不挡收尾:照实记一条失败的复核。交回 None 表示没有
复核结论(问数据等不及的那种,见 chat/job.py)。"""
from app.api.copilot import review_of_run
from app.core.errors import describe_exception
from app.db.base import SessionLocal
try:
async with SessionLocal() as session:
return await review_of_run(session, run, question=self.question, provider=self.provider,
model=self.model, attempt=self.turn.attempts)
except Exception as e: # noqa: BLE001
logger.exception("一轮作答:复核运行 %s 出错", run.id)
return {"verdict": "failed", "note": f"复核出错:{describe_exception(e)}", "signals": [],
"retry_instruction": None}
__all__ = ["CANCEL_GRACE_S", "Turn", "TurnDriver", "add_usage"]
File diff suppressed because it is too large Load Diff
+30
View File
@@ -0,0 +1,30 @@
"""问数据的运行怎么发起:开登录后 user 角色的那几条规则,HTTP 发起(api/runs.start_run)和服务端跑的一轮(chat/job.py)共用。
以前只有 start_run 一处(浏览器发 POST /api/runs)。一轮问数据搬到服务端之后,发起运行的是这一轮的后台任务,规则不能另抄一份——
使用者的问数据运行只许白名单里的节点和工具、记忆固定为自己的范围、查询结果在工具边界遮、数据源一律只读,这几条一处说清。
请求体能不能带记忆范围、知识库,能不能发正式运行,这些是 HTTP 请求本身的事,仍在 start_run 里判。
"""
from __future__ import annotations
from typing import Any
class ChatRunRefused(Exception):
"""这张图使用者在问数据里不能运行:str(e) 就是给人看的原因(HTTP 发起时是 403 的 detail)。"""
def user_chat_run(graph: Any, actor: str | None) -> tuple[str, dict[str, Any]]:
"""开登录后 user 角色发起问数据运行:交回 (记忆范围, 运行选项)。图里有白名单之外的节点、工具抛 ChatRunRefused。
记忆固定为这位使用者自己的范围、知识库取设置里的默认值(记忆、知识库的内容不经遮罩,见 auth/chat_policy 开头);查询结果
在工具边界就遮(事件、回答里只有遮过的值——它读得到自己这次运行的事件,api/deps.py 的 _run_scope);数据源查询一律只读,
写语句不交给审批。"""
from app.auth.chat_policy import chat_graph_problem, chat_memory_scope
from app.tools.datasource import MASK_RESULTS_OPTION, READ_ONLY_OPTION
if problem := chat_graph_problem(graph):
raise ChatRunRefused(problem)
return chat_memory_scope(actor), {MASK_RESULTS_OPTION: True, READ_ONLY_OPTION: True}
__all__ = ["ChatRunRefused", "user_chat_run"]
+138
View File
@@ -0,0 +1,138 @@
"""问数据一轮的事件日志(chat_turn_events,见 db/models_chat.py):追加、回放、实时推送。
**为什么「先落库、再通知、订阅方回库里读」。** 页面要的是:先回放、再接实时,断了按 seq 补齐,刷新、关掉再打开都能接着看。
事件一律先写库,再叫醒等着的订阅方,订阅方按自己收到的最后一个 seq 回库里读——实时和回放是同一段代码,补齐不会漏、不会重复,
也不用像运行的事件总线那样处理慢消费者的积压。订阅方在读库**之前**先登记等待(watch),读完没东西才睡:读和写之间新写进来的
那一条会把它叫醒,不会漏掉。
叫醒只在本进程里(进程内的事件);隔一拍心跳(HEARTBEAT_S)订阅方也会回库里看一次,所以换成多进程(第 2 波)时,别的进程写的
事件最迟一拍心跳就送到——那时再换成 PostgreSQL 的 LISTEN/NOTIFY,这里的接口不用变。
线上格式(SSE,每条一行 data: <JSON>):{seq, type, data, ts};回放完、这一轮也不在进行中了,补一条不带 seq 的
{type: stream.end, status}(任务的状态)再关;隔一拍没有新事件发一行注释心跳「: ping」(页面据此知道连接还活着)。
事件种类见 app/chat/job.py 的模块说明。
"""
from __future__ import annotations
import asyncio
import contextlib
import json
import time
from collections import defaultdict
from collections.abc import AsyncIterator, Awaitable, Callable, Iterator
from typing import Any
from sqlalchemy import func, select, update
from sqlalchemy.ext.asyncio import AsyncSession
from app.db.base import SessionLocal
#: 隔多久没有新事件就发一拍心跳(秒),也是订阅方回库里看一次的间隔
HEARTBEAT_S = 15.0
#: 一次回放最多读多少条
READ_BATCH = 500
#: 回放、实时推送完、这一轮不在进行中时的结束标记(不落库、不带 seq)
STREAM_END = "stream.end"
class _Waiters:
"""本进程里等某一轮新事件的订阅方:一个订阅方一个 asyncio.Event。"""
def __init__(self) -> None:
self._by_turn: dict[str, set[asyncio.Event]] = defaultdict(set)
@contextlib.contextmanager
def watch(self, turn_id: str) -> Iterator[asyncio.Event]:
event = asyncio.Event()
self._by_turn[turn_id].add(event)
try:
yield event
finally:
waiters = self._by_turn.get(turn_id)
if waiters is not None:
waiters.discard(event)
if not waiters:
self._by_turn.pop(turn_id, None)
def notify(self, turn_id: str) -> None:
for event in list(self._by_turn.get(turn_id, ())):
event.set()
waiters = _Waiters()
def notify(turn_id: str) -> None:
"""这一轮有了变化(新事件、任务状态变了):叫醒本进程里等着的订阅方。"""
waiters.notify(turn_id)
async def append(session: AsyncSession, turn_id: str, type_: str, data: dict[str, Any] | None = None) -> int:
"""追加一条,提交,叫醒订阅方。交回它的 seq。
seq 按这一轮库里已有的最大值往下接:同一轮同一时刻只有一个写的人(它的后台任务;任务不在跑时才由停止、启动整理来写),
唯一约束 (turn_id, seq) 兜底。任务行的 last_seq 顺手跟上(GET /api/chat/turns/{id} 照它说现在到哪了)。"""
from app.db.models_chat import ChatTurnEvent, ChatTurnJob
last = int(await session.scalar(select(func.coalesce(func.max(ChatTurnEvent.seq), 0))
.where(ChatTurnEvent.turn_id == turn_id)) or 0)
seq = last + 1
session.add(ChatTurnEvent(turn_id=turn_id, seq=seq, type=type_, ts=time.time(), data=dict(data or {})))
await session.execute(update(ChatTurnJob).where(ChatTurnJob.id == turn_id).values(last_seq=seq)
.execution_options(synchronize_session=False))
await session.commit()
notify(turn_id)
return seq
def wire(row: Any) -> dict[str, Any]:
"""库里的一条 → 线上的一条。"""
return {"seq": row.seq, "type": row.type, "data": row.data or {}, "ts": row.ts}
async def read(turn_id: str, after: int = 0, limit: int = READ_BATCH) -> list[dict[str, Any]]:
"""seq > after 的事件(按 seq),最多 limit 条。"""
from app.db.models_chat import ChatTurnEvent
async with SessionLocal() as session:
rows = (await session.execute(
select(ChatTurnEvent).where(ChatTurnEvent.turn_id == turn_id, ChatTurnEvent.seq > after)
.order_by(ChatTurnEvent.seq).limit(limit))).scalars().all()
return [wire(r) for r in rows]
def frame(event: dict[str, Any]) -> str:
return f"data: {json.dumps(event, ensure_ascii=False, default=str)}\n\n"
#: 心跳(SSE 注释行):页面的解析器只认 data: 开头的帧,这一行只是告诉它连接还活着
PING = ": ping\n\n"
async def stream(turn_id: str, after: int, *, active: Callable[[], Awaitable[str | None]],
heartbeat: float | None = None) -> AsyncIterator[str]:
"""SSE 的每一帧:先回放 seq > after 的,再推实时的;这一轮不在进行中、也没有新的了,补一条 stream.end 再结束。
active() 交回这一轮还在进行中时为 None,不在进行中时交回任务的状态(stream.end 里带的就是它)。"""
beat = HEARTBEAT_S if heartbeat is None else heartbeat
last = after
while True:
with waiters.watch(turn_id) as woke:
rows = await read(turn_id, last)
for row in rows:
last = max(last, int(row["seq"]))
yield frame(row)
if rows:
continue
status = await active()
if status is not None:
yield frame({"type": STREAM_END, "status": status, "data": {"status": status}})
return
try:
await asyncio.wait_for(woke.wait(), timeout=beat)
except asyncio.TimeoutError:
yield PING
__all__ = ["HEARTBEAT_S", "PING", "READ_BATCH", "STREAM_END", "append", "frame", "notify", "read", "stream", "waiters",
"wire"]
+257
View File
@@ -0,0 +1,257 @@
"""问数据一轮落进 conversation_turns 的那一行:服务端写(状态、运行 id、答案、复核、meta),页面只读。
以前这一行是页面写的(frontend/src/store/chat.ts 的 persist / buildMeta),一轮的编排搬到服务端之后由这里写,字段和页面以前
写的一一对应,读的一方(问数据页的 restoreTurn、助手拼历史的 copilot._history_section、列表的 conversations.last_status)不用变:
- status:running | done | error;error 列是一句原话(运行的报错、助手的说法、「已取消」「服务重启,本轮已中断」),助手拼
历史时照它写「这一轮失败了:…」;
- answer:答案正文(成果各字段拼起来,同页面的 answerText,上限 ANSWER_CAP);复核改写了的存改写后的;
- review:复核原样(含 verdict=ok、usage、retry_instruction);没复核过是 {};
- meta(v=1):runId、runClass、runStatus、issuance(瘦过的出具信息)、noQuery、outcome(cancelled | suspended)、rejected
(驳回的人、哪一步、备注:run.cancelled 的原样)、cause(失败的原因,见下)、attempts(用户点重试之前的各次)、ms、queries、
scope、asked、looks、evidence、usage({copilot, review, earlier, run})、at。
**失败只记原因(cause),不记说法。** 页面以前记的是翻成人话之后的 failure(标题、原因、怎么办:lib/explain 那一套),说法在
界面层;服务端只记发生了什么,页面恢复时照同一套 lib/explain 讲(frontend/src/store/chat.ts 的 failureFromCause):
{kind: refused, message, code} 开流之前就被拒(模型没配好、限定的数据源都不在)
{kind: copilot, message, hint, detail} 生成过程交出 error(助手本轮未完成)
{kind: blocked, issues, self_checked} 生成的图没过检查
{kind: start, message, code, status} 发起运行被拒(绑定的工具不在、使用者用不了的节点)
{kind: run, error, detail} 运行失败
{kind: gone} 运行记录不见了
{kind: build_interrupted} 建图时服务重启,没有可恢复的断点
{kind: crash, message} 这一轮的后台任务自己出错
老轮次的 meta.failure 照旧认(页面优先读它)。
"""
from __future__ import annotations
import json
import time
from collections.abc import Mapping
from typing import Any
#: 落库答案的上限(同页面的 ANSWER_CAP):只是防跑飞,不是显示上限
ANSWER_CAP = 20_000
ANSWER_CLIPPED = "\n\n…(内容过长,已截断,完整内容请查看运行记录)"
#: 归档的旧答案存多少(同页面的 ATTEMPT_ANSWER_CAP)
ATTEMPT_ANSWER_CAP = 4000
#: 停止、服务重启挂起时 error 列里的那一句(同页面落库时写的)
CANCELLED = "已取消"
SUSPENDED = "服务重启,本轮已中断"
#: 建图时服务重启
BUILD_INTERRUPTED = "服务重启,本轮在生成工作流时中断"
GONE = "运行记录不见了"
#: 用量里记的几项(同页面的 USAGE_KEYS)
USAGE_KEYS = ("cost_usd", "unpriced_calls", "calls", "unfinished_calls", "input_tokens", "output_tokens")
#: 出具信息里无法回指的数字最多存几个(同页面的 slimIssuance)
ISSUANCE_UNMATCHED_CAP = 20
def answer_text(output: Mapping[str, Any] | None) -> str:
"""成果压成一句话(同页面的 answerText):下划线开头的、空的不要;字符串原样,别的按 JSON;超过上限截断并说出来。"""
if not output:
return ""
parts = [value if isinstance(value, str) else json.dumps(value, ensure_ascii=False, separators=(",", ":"))
for key, value in output.items() if not str(key).startswith("_") and value not in (None, "")]
text = "\n".join(parts)
return text[:ANSWER_CAP] + ANSWER_CLIPPED if len(text) > ANSWER_CAP else text
def slim_issuance(issuance: Any) -> dict[str, Any] | None:
"""出具信息存进 meta 前瘦一下身(同页面的 slimIssuance):无法回指的数字只留前 20 个。"""
if not isinstance(issuance, Mapping) or not issuance.get("tier"):
return None
out = dict(issuance)
unmatched = issuance.get("unmatched_numbers")
if isinstance(unmatched, list):
out["unmatched_numbers"] = unmatched[:ISSUANCE_UNMATCHED_CAP]
return out
def has_evidence(output: Mapping[str, Any] | None) -> bool:
"""成果带不带逐段证据(同页面 lib/evidence 的 hasEvidence):_evidence 或它的 others 里有文档、有字段。"""
ev = (output or {}).get("_evidence") if isinstance(output, Mapping) else None
if not isinstance(ev, Mapping):
return False
refs = [ev, *[r for r in ev.get("others") or [] if isinstance(r, Mapping)]]
return any(isinstance(r.get("doc_artifact"), str) and r.get("doc_artifact") and isinstance(r.get("fields"), list)
and r.get("fields") for r in refs)
def _names(value: Any) -> list[str]:
return [str(v) for v in value if v not in (None, "")] if isinstance(value, list) else []
def _num(value: Any) -> float | None:
return value if isinstance(value, (int, float)) and not isinstance(value, bool) else None
def look_event(op: Mapping[str, Any]) -> dict[str, Any] | None:
"""助手的 look 操作 → 存进 meta.looks 的那一份(同页面 run/decode 的 lookEventOf):下一轮服务端据此写「上一轮查看过」。"""
if op.get("op") != "look":
return None
listed = op.get("objects") if isinstance(op.get("objects"), list) else op.get("sources")
if not isinstance(listed, list):
return None
objects = []
for x in listed:
if not isinstance(x, Mapping):
continue
kind = x.get("kind") if x.get("kind") in ("capability", "workflow_version", "caliber") else "table"
total, explored = _num(x.get("total")), _num(x.get("explored"))
obj: dict[str, Any] = {"kind": kind}
if isinstance(x.get("source"), str) and x.get("source"):
obj["source"] = x["source"]
if isinstance(x.get("source_id"), str) and x.get("source_id"):
obj["source_id"] = x["source_id"]
obj.update(named=_names(x.get("named")), picked=_names(x.get("picked")), joined=_names(x.get("joined")),
missing=_names(x.get("missing")))
if total is not None:
obj["total"] = total
if explored is not None and explored < total:
obj["explored"] = explored
if _num(x.get("catalog_tables")) is not None:
obj["catalog_tables"] = x["catalog_tables"]
if _names(x.get("added_to_scope")):
obj["added_to_scope"] = _names(x.get("added_to_scope"))
if _num(x.get("prefiltered")) is not None:
obj["prefiltered"] = x["prefiltered"]
objects.append(obj)
out: dict[str, Any] = {"op": "look", "id": str(op.get("id") or ""), "by": "system" if op.get("by") == "system" else "model",
"need": str(op.get("need") or ""),
"status": op.get("status") if op.get("status") in ("index", "failed") else "ok"}
if isinstance(op.get("reason"), str) and op.get("reason"):
out["reason"] = op["reason"]
out["elapsed_ms"] = _num(op.get("elapsed_ms")) or 0
out["objects"] = objects
return out
# --------------------------------------------------------------------------
# 用量:{copilot, review, earlier, run},每份是 UsageTotal(同页面的 TurnUsage)
# --------------------------------------------------------------------------
def usage_total(usage: Mapping[str, Any] | None) -> dict[str, Any] | None:
"""一份用量(providers/usage 的汇总、运行的 usage)只留要记的几项。不是对象交回 None。"""
return add_total({}, usage) if isinstance(usage, Mapping) else None
def add_total(a: Mapping[str, Any] | None, b: Mapping[str, Any] | None) -> dict[str, Any]:
"""两份用量相加(缺的当 0,同页面的 addUsage):只记不为 0 的项,金额保留 6 位。"""
out = {k: v for k, v in (a or {}).items() if k in USAGE_KEYS}
for key in USAGE_KEYS:
try:
value = float((b or {}).get(key) or 0)
except (TypeError, ValueError):
continue
if value:
out[key] = (out.get(key) or 0) + value
for key in USAGE_KEYS:
if key in out and key != "cost_usd":
out[key] = int(out[key])
if "cost_usd" in out:
out["cost_usd"] = round(float(out["cost_usd"]), 6)
return out
def retire_run(usage: Mapping[str, Any] | None) -> dict[str, Any]:
"""当前这次运行换下来了(自动重来、用户点重试):它的用量并进 earlier,这一轮的花费不丢这一笔(同页面的 retireRun)。"""
out = dict(usage or {})
run = out.pop("run", None)
if run:
out["earlier"] = add_total(out.get("earlier"), run)
return out
# --------------------------------------------------------------------------
# 用户点重试:上一次的结局归档
# --------------------------------------------------------------------------
#: 归档那一次的一句话(同页面以前 attemptOf 的写法)。也是重试时交给助手的原因
ATTEMPT_TEXT = {
"cancelled": "已手动停止",
"suspended": "服务重启,运行中断",
"rejected": "被驳回",
"unusable": "结论不可用:{note}",
"partial": "存在缺口:{note}",
"done": "已给出答案",
}
def _rejected_line(rejected: Mapping[str, Any]) -> str:
"""被驳回的那一句(同页面 REJECTED_TEXT.sub):谁在哪一步驳回、备注;页面记过 sub 的照用。"""
if isinstance(rejected.get("sub"), str) and rejected["sub"]:
return rejected["sub"]
who = str(rejected.get("actor") or "").strip()
label = str(rejected.get("label") or "").strip()
where = f"「{label}」" if label else "这一步"
note = str(rejected.get("note") or "").strip()
if rejected.get("decision") == "expired_reject":
return f"{where}超过时限,按设置驳回"
return (f"{who} 驳回了{where}" if who else f"{where}被驳回") + (f":{note}" if note else "")
def attempt_record(row: Any, meta: Mapping[str, Any], *, n: int, at_ms: int | None = None) -> dict[str, Any]:
"""这一轮现在的样子折成一条留档(同页面以前的 attemptOf):{n, outcome, summary, detail?, runId, answer?, at, cause?}。
用户主动重试不能把上一次抹掉:那一次的答案可能只是有缺口、并非全错,新的一次也未必更好,两份要能对照。"""
from app.api.conversations import turn_run_id
review = row.review if isinstance(row.review, Mapping) else {}
answer = row.answer or ""
if len(answer) > ATTEMPT_ANSWER_CAP:
answer = answer[:ATTEMPT_ANSWER_CAP] + "…"
base: dict[str, Any] = {"n": n, "at": at_ms if at_ms is not None else int(time.time() * 1000),
"runId": turn_run_id(row)}
if answer:
base["answer"] = answer
signals = "\n".join(str(s.get("detail") or "") for s in review.get("signals") or [] if isinstance(s, Mapping)) or None
outcome = meta.get("outcome")
if outcome == "cancelled" and isinstance(meta.get("rejected"), Mapping):
sub = _rejected_line(meta["rejected"])
return {**base, "outcome": "rejected", "summary": ATTEMPT_TEXT["rejected"] + (f":{sub}" if sub else "")}
if outcome == "cancelled" or (row.status == "error" and (row.error or "").strip() == CANCELLED):
return {**base, "outcome": "cancelled", "summary": ATTEMPT_TEXT["cancelled"]}
if outcome == "suspended":
return {**base, "outcome": "suspended", "summary": ATTEMPT_TEXT["suspended"]}
if row.status == "error":
failure = meta.get("failure") if isinstance(meta.get("failure"), Mapping) else None
line = (f"{failure.get('title')}:{failure.get('reason')}" if failure and failure.get("reason")
else str(failure.get("title")) if failure and failure.get("title") else (row.error or "发生错误"))
out = {**base, "outcome": "failed", "summary": line}
if failure and failure.get("detail"):
out["detail"] = str(failure["detail"])
if isinstance(meta.get("cause"), Mapping):
out["cause"] = dict(meta["cause"])
return out
note = str(review.get("note") or "")
if review.get("severity") == "broken":
return {**base, "outcome": "unusable", "summary": ATTEMPT_TEXT["unusable"].format(note=note), "detail": signals}
if review.get("severity") == "degraded":
return {**base, "outcome": "partial", "summary": ATTEMPT_TEXT["partial"].format(note=note), "detail": signals}
return {**base, "outcome": "done", "summary": ATTEMPT_TEXT["done"]}
def attempt_reason(record: Mapping[str, Any]) -> str:
"""重试时交给助手的原因(同页面以前的写法):那一次的一句话,有细节的括号里带前 300 字。"""
summary = str(record.get("summary") or "")
detail = record.get("detail")
return f"{summary}({str(detail)[:300]})" if detail else summary
def with_max_steps(graph: Mapping[str, Any], steps: int) -> dict[str, Any]:
"""图里每个 agent 节点的步数上限调到 steps(「放宽步数重跑」,同页面的 withMaxSteps)。"""
nodes = []
for node in graph.get("nodes") or []:
if isinstance(node, Mapping) and node.get("type") == "agent":
data = dict(node.get("data") or {})
data["config"] = {**dict(data.get("config") or {}), "max_steps": steps}
node = {**node, "data": data}
nodes.append(node)
return {**dict(graph), "nodes": nodes}
__all__ = ["ANSWER_CAP", "ATTEMPT_ANSWER_CAP", "ATTEMPT_TEXT", "BUILD_INTERRUPTED", "CANCELLED", "GONE", "SUSPENDED",
"USAGE_KEYS", "add_total", "answer_text", "attempt_reason", "attempt_record", "has_evidence", "look_event",
"retire_run", "slim_issuance", "usage_total", "with_max_steps"]
+17 -4
View File
@@ -14,7 +14,8 @@
/copilot/review 的结果带 retry_instruction(要重来时是那句话,不重来是 null)。问数据页照后端给的走;老后端没有这两个字段时退回它自己那份同样的
规则(tests/test_chat_turn_policy.py 逐条对照两边的常量和措辞)。评测(第 3 波)用同一组函数把一轮串起来。
只搬「决定」:重连、断点续看、取消、复核等多久这些仍在问数据页(把整轮编排搬到后端要重做这些,单独立项)。
整轮编排后来也搬到了服务端(docs/design/orchestration-round §1):流程在 chat/drive.py(问数据和评测共用),问数据的后台任务在
chat/job.py;页面只发起、只看,不再自己判这些。老服务端那条退路(页面自己那份规则)随之去掉。
"""
from __future__ import annotations
@@ -50,6 +51,12 @@ RETRY_WITH_BASE = ("这个工作流上一次运行没有成功:{reason}。\n\n
"改完继续回答原来的问题:{question}")
#: 没有图可交回(上一次连图都没建出来)时:原问题后面缀上原因,从零再建
RETRY_FROM_SCRATCH = "{question}\n\n(上一次没有成功:{reason}。请针对这个原因调整,重新给出完整的工作流。)"
#: 用户点「重试本轮」时交给助手的话(以前在问数据页 retryTurn 里拼,一轮搬到服务端之后在这里):有图时说的是「没有给出可用的
#: 结果」——上一次未必是运行失败(停止了、被驳回、结论不可用都算),原因是那一次的归档说法(chat/record.attempt_reason);
#: 没有图的同 RETRY_FROM_SCRATCH
RETRY_USER_WITH_BASE = ("这个工作流上一次没有给出可用的结果:{reason}。\n\n"
"请针对这个原因改它(比如调大最大步数、改正工具参数、换个查询方式),"
"改完继续回答原来的问题:{question}")
def _blockers(final: Mapping[str, Any]) -> list[str]:
@@ -142,12 +149,18 @@ def retry_instruction(question: str, reason: str, has_base: bool) -> str:
return template.format(question=question, reason=reason)
def user_retry_instruction(question: str, reason: str, has_base: bool) -> str:
"""用户点「重试本轮」时交给助手的话。has_base:有上一次建出的图可以交回去改。"""
template = RETRY_USER_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__ = ["EMPTY_GENERATION", "MAX_ATTEMPTS", "NEXT_ASKED", "NEXT_FAIL", "NEXT_READY", "NEXT_REPLY", "NEXT_RETRY",
"NEXT_RUN", "NEXT_STEPS", "QUESTION_FIELD", "RETRY_ERRORS", "RETRY_FROM_SCRATCH", "RETRY_WITH_BASE",
"blocked_reason", "error_reason", "fill_input", "has_nodes", "next_step", "retry_after_review",
"retry_instruction", "review_reason"]
"NEXT_RUN", "NEXT_STEPS", "QUESTION_FIELD", "RETRY_ERRORS", "RETRY_FROM_SCRATCH", "RETRY_USER_WITH_BASE",
"RETRY_WITH_BASE", "blocked_reason", "error_reason", "fill_input", "has_nodes", "next_step", "retry_after_review",
"retry_instruction", "review_reason", "user_retry_instruction"]
+1 -1
View File
@@ -1093,4 +1093,4 @@ from app.db.models_blueprints import ( # noqa: E402,F401
)
# 下一轮第 2 波(docs/design/next-round):数据源对象索引与后台探查(models_sources)、评测(models_eval)。整模块导入,
# 各自的模型类由那两个文件自己登记,这里不逐个列名
from app.db import models_eval, models_sources # noqa: E402,F401
from app.db import models_chat, models_eval, models_sources # noqa: E402,F401
+73
View File
@@ -0,0 +1,73 @@
"""问数据的一轮在服务端跑(docs/design/orchestration-round/00-总方案.md §1):这一轮的后台任务和它的事件日志。
由 app/db/models.py 末尾整模块导入,create_all 建表(都是新表,不改已有表)。这里不导入 app.db.models:models.py 导入本模块时
自己还没加载完,外键一律按表名写。
**为什么一轮一行任务、id 就是 conversation_turns 的 id。** 一轮问数据(建图 → 运行 → 复核 → 按复核重来、落库)以前是浏览器在推,
页面一断这一轮就没人管了。现在它是服务端的一个后台任务:发起、重试本轮、重新运行本轮、继续运行都还是「这一轮」,用户看到的
是同一轮里的几次尝试,所以任务按轮记、一轮一行,重试时把这一行重新排队(上一次的结局归档进 conversation_turns.meta.attempts)。
- ChatTurnJob:状态用后台任务的状态词(core/job_runtime:queued → running → succeeded / failed / cancelled / interrupted),
stage 是这一轮现在在哪一步(build 建图、run 运行、waiting 停在审批上等人、review 复核),run_id 是当前这次运行。request 是
这次要做什么(问题、限定的数据源、方式、重试的原因、放宽到几步……),principal 是发起的人(身份快照:服务重启后接着做时
按这个人的身份——遮罩、只读、对话归属都看它,见 app/chat/job.py)。attempt_seq 是这一次尝试在事件日志里从哪一条开始:
刷新、重开页面时从这里回放,不必把之前几次尝试的过程整条重放。
- ChatTurnEvent:这一轮的事件日志。每一步(助手的操作、心跳、查看、final、用量、运行 id、复核结论、重来、终局)按 seq
一条,页面先回放、再接实时,断了按 seq 补齐(GET /api/chat/turns/{id}/stream?after=)。不进运行的封存范围:运行自己的
事件仍在 run_events,这里只记「这一轮」的事。
"""
from __future__ import annotations
from datetime import datetime
from typing import Any
from sqlalchemy import Float, ForeignKey, Integer, String, Text, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base, TimestampMixin, UTCDateTime, new_id
#: 这一轮现在在哪一步
STAGES = ("build", "run", "waiting", "review")
class ChatTurnJob(Base, TimestampMixin):
"""问数据一轮的后台任务(见模块说明)。"""
__tablename__ = "chat_turn_jobs"
#: 就是 conversation_turns.id
id: Mapped[str] = mapped_column(ForeignKey("conversation_turns.id", ondelete="CASCADE"), primary_key=True)
conversation_id: Mapped[str] = mapped_column(ForeignKey("conversations.id", ondelete="CASCADE"), index=True)
#: queued | running | succeeded | failed | cancelled | interrupted(core/job_runtime 的状态词)
status: Mapped[str] = mapped_column(String(20), default="queued", index=True)
stage: Mapped[str | None] = mapped_column(String(20), default=None)
#: ask | retry | rerun | max_steps | run | continue | resume(app/chat/job.py 的 MODES)
mode: Mapped[str] = mapped_column(String(20), default="ask")
run_id: Mapped[str | None] = mapped_column(String(32), default=None)
#: 这一次尝试里自动重来了几次(从 0 数,同 chat/turn.MAX_ATTEMPTS 的口径)
attempt: Mapped[int] = mapped_column(Integer, default=0)
request: Mapped[dict[str, Any]] = mapped_column(default=dict)
principal: Mapped[dict[str, Any]] = mapped_column(default=dict)
actor: Mapped[str | None] = mapped_column(String(100), default=None)
last_seq: Mapped[int] = mapped_column(Integer, default=0)
attempt_seq: Mapped[int] = mapped_column(Integer, default=0)
error: Mapped[str | None] = mapped_column(Text, default=None)
started_at: Mapped[datetime | None] = mapped_column(UTCDateTime, default=None)
finished_at: Mapped[datetime | None] = mapped_column(UTCDateTime, default=None)
class ChatTurnEvent(Base):
"""问数据一轮的事件日志里的一条(见模块说明)。"""
__tablename__ = "chat_turn_events"
__table_args__ = (UniqueConstraint("turn_id", "seq", name="uq_chat_turn_events_turn_seq"),)
id: Mapped[str] = mapped_column(String(32), primary_key=True, default=new_id)
turn_id: Mapped[str] = mapped_column(ForeignKey("conversation_turns.id", ondelete="CASCADE"), index=True)
seq: Mapped[int] = mapped_column(Integer)
type: Mapped[str] = mapped_column(String(40))
ts: Mapped[float] = mapped_column(Float)
data: Mapped[dict[str, Any]] = mapped_column(default=dict)
__all__ = ["STAGES", "ChatTurnEvent", "ChatTurnJob"]
+32 -301
View File
@@ -1,8 +1,8 @@
"""评测里的一轮作答:无头地走一遍问数据页走的那一轮(细案 01 做法 1、9;总方案 2.1)。
**量的就是用户拿到的那一轮。** 不另写一份 Python 版编排:生成过程是 api/copilot.answer_ops(问数据页经 SSE 收的同一串
操作),几个「决定」是 chat/turn 的纯函数(问数据页照服务端给的 next、retry_instruction 走),复核是 api/copilot.review_of_run
(问数据页经 /copilot/review 调的同一个函数)。这里只把它们串起来:
**量的就是用户拿到的那一轮。** 不另写一份编排:一轮的流程是 app/chat/drive.TurnDriver(问数据的一轮在服务端跑也是它,
app/chat/job.py),生成过程是 api/copilot.answer_ops,几个「决定」是 chat/turn 的纯函数,复核是 api/copilot.review_of_run。这里
只写评测和问数据不一样的那几项(Driver 是 TurnDriver 的子类):
answer_ops → final.next(chat/turn.next_step)→ launch_run(origin=eval) → 等终态 → review_of_run → 要的话带原图重来一次
@@ -29,22 +29,21 @@ llm.end(按模型分)。重来的话两次都算——那都是这一题花
"""
from __future__ import annotations
import asyncio
import copy
import logging
import time
from collections.abc import Awaitable, Callable, Mapping
from dataclasses import dataclass, field
from typing import Any
from sqlalchemy import func, select
from sqlalchemy import select
# 一轮的流程和结果抽到了 chat/drive.py(问数据、评测共用),这里留着老名字:runner、stats 和测试照旧从这里取
from app.chat.drive import CANCEL_GRACE_S, Turn, TurnDriver, add_usage # noqa: F401
logger = logging.getLogger(__name__)
#: 问数据页最多等复核这么久(chat.ts 的 REVIEW_TIMEOUT_MS):评测照等,超过的单列
#: 问数据页最多等复核这么久(chat/job.REVIEW_WAIT_S,和以前页面上的 REVIEW_TIMEOUT_MS 同一个数):评测照等,超过的单列
REVIEW_PAGE_WAIT_S = 25.0
#: 取消一次运行之后最多再等它收尾多久
CANCEL_GRACE_S = 15.0
#: 运行停在审批上:评测不代批
APPROVAL_REASON = "运行停在审批上(危险工具、人工节点):评测不代批,已取消这次运行"
#: 每一步在做什么(EvalResult.current)
@@ -71,69 +70,6 @@ class TurnContext:
view: Mapping[str, Any] | None = None
@dataclass
class Turn:
"""一轮作答的结果(也是进行中的状态:current_run、phase 随时更新)。"""
path: str | None = None
run_id: str | None = None
#: 这一题发起过的全部作答运行(重来的话两次),失败的也在
runs: list[str] = field(default_factory=list)
attempts: int = 0
#: 每次自动重来的原因(和 attempts 一一对应):没过检查、复核说不可信、生成过程一个操作都没给
retries: list[str] = field(default_factory=list)
reason: str | None = None
#: 助手调用失败(生成过程交出 error、开流前就被拒):runner 据此止损
copilot_failed: bool = False
timed_out: bool = False
#: 直接回答的原文、反问的问题
reply: str | None = None
asked: dict[str, Any] | None = None
review: dict[str, Any] | None = None
looks: list[dict[str, Any]] = field(default_factory=list)
#: 助手这一轮用的模型 id(model 操作)
assistant_model: str | None = None
#: {assistant: usage 汇总(生成,含重来), review: usage 汇总}
usage: dict[str, Any] = field(default_factory=dict)
timings: dict[str, int] = field(default_factory=dict)
#: 写作模型按变体换了,但节点自己点了模型、没被覆盖的节点 id
own_models: list[str] = field(default_factory=list)
#: 正在跑的那次运行(调用方超时、停止时取消它)
current_run: str | None = None
phase: str | None = None
def to_dict(self) -> dict[str, Any]:
return {"path": self.path, "run_id": self.run_id, "runs": list(self.runs), "attempts": self.attempts,
"retries": list(self.retries), "reason": self.reason, "copilot_failed": self.copilot_failed, "timed_out": self.timed_out,
"reply": self.reply, "asked": self.asked, "review": self.review, "looks": list(self.looks),
"assistant_model": self.assistant_model, "usage": copy.deepcopy(self.usage),
"timings": dict(self.timings), "own_models": list(self.own_models)}
#: 用量汇总里相加的计数(providers/usage 的 summary):unfinished_calls 是没做完的调用(被取消、超时,token 只计了已报的)
_COUNTED = ("input_tokens", "output_tokens", "calls", "unpriced_calls", "unfinished_calls")
def _add_usage(into: dict[str, Any] | None, more: Mapping[str, Any] | None) -> dict[str, Any]:
"""两份 providers/usage 汇总相加(by_model 逐个模型加)。"""
out = copy.deepcopy(into) if into else {"by_model": {}, **dict.fromkeys(_COUNTED, 0), "cost_usd": 0.0}
if not more:
return out
for key in _COUNTED:
out[key] = int(out.get(key) or 0) + int(more.get(key) or 0)
out["cost_usd"] = round(float(out.get("cost_usd") or 0) + float(more.get("cost_usd") or 0), 6)
for model, row in (more.get("by_model") or {}).items():
mine = out["by_model"].setdefault(model, {**dict.fromkeys(_COUNTED, 0), "cost_usd": 0.0})
for key in _COUNTED:
mine[key] = int(mine.get(key) or 0) + int(row.get(key) or 0)
mine["cost_usd"] = round(float(mine.get("cost_usd") or 0) + float(row.get("cost_usd") or 0), 6)
return out
def add_usage(into: dict[str, Any] | None, more: Mapping[str, Any] | None) -> dict[str, Any]:
return _add_usage(into, more)
async def writer_usage(run_ids: list[str]) -> dict[str, Any]:
"""答案运行里的模型调用(llm.end,按模型分):写作这一角色的用量,形状同 providers/usage 的汇总。封存之后追加的
(按需裁判、质询)不算:那是事后另记的账(裁判记在评测的账上另算,见 judging.py)。"""
@@ -142,7 +78,7 @@ async def writer_usage(run_ids: list[str]) -> dict[str, Any]:
from app.db.models import RunEvent
from app.providers.catalog import unpriced_call
out = _add_usage(None, None)
out = add_usage(None, None)
if not run_ids:
return out
async with SessionLocal() as session:
@@ -156,7 +92,7 @@ async def writer_usage(run_ids: list[str]) -> dict[str, Any]:
piece = {"input_tokens": int(data.get("input_tokens") or 0), "output_tokens": int(data.get("output_tokens") or 0),
"calls": int(data.get("calls") or 1), "cost_usd": float(data.get("cost_usd") or 0.0),
"unpriced_calls": 1 if unpriced_call(data) else 0}
out = _add_usage(out, {**piece, "by_model": {model: piece}})
out = add_usage(out, {**piece, "by_model": {model: piece}})
return out
@@ -208,150 +144,46 @@ async def stop_run(run_id: str | None, actor: str | None = None) -> None:
logger.exception("评测:取消运行 %s 时出错", run_id)
class Driver:
"""一轮作答。drive() 跑完一轮,结果在 turn 上;on_phase(阶段) 在每一步开始时调(runner 据此写「正在做什么」)。"""
class Driver(TurnDriver):
"""一轮作答(流程在 TurnDriver)。drive() 跑完一轮,结果在 turn 上;on_phase(阶段) 在每一步开始时调(runner 据此写
「正在做什么」)。和问数据不一样的:助手模型、目录视图按变体;写作模型按变体写进图;来源 eval、只读和遮罩;停在审批上
直接取消(评测不代批);复核一直等。"""
def __init__(self, question: str, *, datasource_ids: list[str], ctx: TurnContext,
on_phase: Callable[[str], Awaitable[None]] | None = None) -> None:
self.question = question
self.datasource_ids = list(datasource_ids)
super().__init__(question, datasource_ids=datasource_ids, on_phase=on_phase,
provider=ctx.assistant.get("provider"), model=ctx.assistant.get("model"))
self.ctx = ctx
self.on_phase = on_phase
self.turn = Turn()
async def _phase(self, phase: str) -> None:
self.turn.phase = phase
if self.on_phase is not None:
try:
await self.on_phase(phase)
except Exception: # noqa: BLE001 - 写进度失败不挡作答
logger.exception("评测:记进度出错")
# ------------------------------------------------------------------ 生成
async def _generate(self, instruction: str, base_graph: dict[str, Any] | None) -> tuple[list[dict[str, Any]], str | None]:
"""跑一次生成过程,收下全部操作。开流前就被拒(HTTPException)交回原因。"""
from fastapi import HTTPException
from app.api.copilot import GenerateIn, answer_ops
payload = GenerateIn(instruction=instruction, base_graph=base_graph, intent="answer",
provider=self.ctx.assistant.get("provider"), model=self.ctx.assistant.get("model"),
datasource_ids=self.datasource_ids or None)
ops: list[dict[str, Any]] = []
try:
async for op in answer_ops(payload, catalog_view=self.ctx.view):
if op.get("op") in ("thinking", "heartbeat"):
continue
ops.append(op)
except HTTPException as e:
return ops, str(e.detail)
return ops, None
def _take(self, ops: list[dict[str, Any]]) -> dict[str, Any] | None:
"""收下这一次生成的用量、查看记录、模型;交回结局(final / reply / error),只有反问的为 None。"""
terminal = None
for op in ops:
kind = op.get("op")
if kind == "usage":
self.turn.usage["assistant"] = _add_usage(self.turn.usage.get("assistant"),
{k: v for k, v in op.items() if k != "op"})
elif kind == "look":
self.turn.looks.append({k: op.get(k) for k in ("id", "by", "need", "status") if k in op})
elif kind == "model":
self.turn.assistant_model = str(op.get("model") or "") or None
elif kind in ("final", "reply", "error"):
terminal = op
return terminal
# ------------------------------------------------------------------ 运行
async def _launch(self, graph: dict[str, Any], payload: dict[str, Any] | None = None) -> str | None:
from app.chat import turn as policy
from app.db.base import SessionLocal
from app.engine.launch import LaunchRefused, launch_run
def catalog_view(self) -> Any:
return self.ctx.view
def prepare_graph(self, graph: dict[str, Any]) -> dict[str, Any]:
graph, own = with_writer(graph, self.ctx.writer)
self.turn.own_models = sorted(set(self.turn.own_models) | set(own))
return graph
async def launch_args(self, session: Any, graph: dict[str, Any]) -> dict[str, Any]:
origin = {k: v for k, v in {
"suite_id": self.ctx.suite_id, "eval_run_id": self.ctx.eval_run_id, "case_id": self.ctx.case_id,
"role": "answer", "arm": self.ctx.arm, "repeat": self.ctx.repeat, "attempt": self.turn.attempts}.items()
if v is not None}
try:
async with SessionLocal() as session:
inputs = payload if payload is not None else policy.fill_input(graph, self.question, "")
run = await launch_run(session, graph=graph, input_payload=inputs,
actor=self.ctx.actor, options=_options(self.ctx.view), origin_kind="eval",
origin=origin, workflow_name=run_name(self.ctx.suite_name, self.ctx.case_seq))
except LaunchRefused as e:
self.turn.path, self.turn.reason = "run_failed", f"运行没有发起:{e.message}"
return None
self.turn.runs.append(run.id)
self.turn.run_id = self.turn.current_run = run.id
return run.id
async def _wait(self, run_id: str) -> Any:
"""等运行到终态(或停在审批上)。交回库里的运行。"""
from app.db.base import SessionLocal
from app.db.models import Run
from app.engine.runner import run_manager
while True:
await run_manager.wait_idle(run_id, timeout=5.0)
async with SessionLocal() as session:
row = await session.get(Run, run_id)
if row is None or row.status in ("succeeded", "failed", "cancelled", "interrupted"):
# 封存在终态之后:等执行任务收完尾(含封存),评分只读封存范围
await run_manager.wait_idle(run_id, timeout=CANCEL_GRACE_S)
async with SessionLocal() as session:
return await session.get(Run, run_id)
await asyncio.sleep(0.2)
async def _pending_approvals(self, run_id: str) -> int:
from app.db.base import SessionLocal
from app.db.models import Approval
async with SessionLocal() as session:
return int(await session.scalar(select(func.count()).select_from(Approval).where(
Approval.run_id == run_id, Approval.status.in_(("pending", "answered")))) or 0)
# ------------------------------------------------------------------ 一轮
async def _run(self, graph: dict[str, Any], payload: dict[str, Any] | None = None) -> Any | None:
"""发起、等到终态。成功结束的交回运行;别的结局记好路径和原因,交回 None。"""
from app.engine.runner import END_REJECTED
return {"actor": self.ctx.actor, "options": _options(self.ctx.view), "origin_kind": "eval", "origin": origin,
"workflow_name": run_name(self.ctx.suite_name, self.ctx.case_seq)}
async def on_interrupted(self, run_id: str, row: Any) -> Any | None:
t = self.turn
run_id = await self._launch(graph, payload)
if run_id is None:
return None
row = await self._wait(run_id)
t.current_run = None
usage = dict((row.usage or {}) if row is not None else {})
t.timings["run_wall_ms"] = t.timings.get("run_wall_ms", 0) + int(usage.get("wall_ms") or 0)
t.timings["run_active_ms"] = t.timings.get("run_active_ms", 0) + int(
usage.get("active_ms") or usage.get("duration_ms") or 0)
if row is None:
t.path, t.reason = "run_failed", "运行记录不见了"
return None
if row.status == "interrupted":
waiting = await self._pending_approvals(run_id)
await stop_run(run_id, self.ctx.actor)
t.path = "run_failed"
t.reason = APPROVAL_REASON if waiting else "运行停在断点上(服务可能重启过):评测不继续这次运行,已取消"
return None
if row.status == "cancelled" and row.end_reason == END_REJECTED:
t.path, t.reason = "rejected", "运行被驳回(人的决定,不算运行失败)"
return None
if row.status != "succeeded":
t.path = "run_failed"
t.reason = f"运行{'失败' if row.status == 'failed' else '被停止'}:{row.error or '没有记下原因'}"
return None
return row
waiting = await self._pending_approvals(run_id)
await stop_run(run_id, self.ctx.actor)
t.path = "run_failed"
t.reason = APPROVAL_REASON if waiting else "运行停在断点上(服务可能重启过):评测不继续这次运行,已取消"
return None
async def drive_workflow(self, graph: dict[str, Any], inputs: dict[str, Any]) -> Turn:
"""工作流题:钉住那一版的图按输入跑一次探索运行(来源 eval)。不建图、不复核:考的是这一版本身。不发正式运行——
正式运行会被 pick_base 和应用视图的「最近一次封存」取到(同期望探针不发正式运行的理由)。"""
import time
t = self.turn
started = time.monotonic()
try:
@@ -362,107 +194,6 @@ class Driver:
finally:
t.timings["total_ms"] = int((time.monotonic() - started) * 1000)
async def drive(self) -> Turn:
from app.chat import turn as policy
t = self.turn
started = time.monotonic()
instruction, base_graph = self.question, None
try:
while True:
await self._phase("build")
building = time.monotonic()
ops, refused = await self._generate(instruction, base_graph)
t.timings["build_ms"] = t.timings.get("build_ms", 0) + int((time.monotonic() - building) * 1000)
terminal = self._take(ops)
if refused is not None:
t.path, t.reason, t.copilot_failed = "build_failed", f"助手没有开始:{refused}", True
return t
if terminal is None:
asked = next((o for o in reversed(ops) if o.get("op") == "ask"), None)
t.path, t.asked = "asked", {k: asked.get(k) for k in ("question", "options")} if asked else None
return t
kind = terminal.get("op")
if kind == "error":
# 生成过程一个操作都没给(只写了思考):按重来一次的规矩带着原因重来(同问数据页,chat/turn.next_step)
step = terminal.get("next") or policy.next_step(terminal)
if step == policy.NEXT_RETRY and t.attempts + 1 < policy.MAX_ATTEMPTS:
reason = policy.error_reason(terminal)
t.attempts += 1
t.retries.append(reason)
instruction = policy.retry_instruction(self.question, reason, policy.has_nodes(base_graph))
continue
t.path, t.reason, t.copilot_failed = "build_failed", str(terminal.get("message") or "助手本轮未完成"), True
return t
if kind == "reply":
t.path, t.reply = "reply", str(terminal.get("text") or "")
return t
if terminal.get("asked"):
t.path, t.asked = "asked", dict(terminal["asked"])
return t
self_checked = any(o.get("op") == "check" for o in ops)
step = terminal.get("next") or policy.next_step(terminal, self_checked=self_checked)
graph = terminal.get("graph") or {}
if step == policy.NEXT_RETRY:
if t.attempts + 1 < policy.MAX_ATTEMPTS:
t.attempts += 1
t.retries.append(policy.blocked_reason(terminal))
instruction = policy.retry_instruction(self.question, policy.blocked_reason(terminal),
policy.has_nodes(graph))
base_graph = graph if policy.has_nodes(graph) else None
continue
t.path, t.reason = "build_failed", policy.blocked_reason(terminal)
return t
if step == policy.NEXT_FAIL:
errors = [str(i.get("message") or "") for i in terminal.get("issues") or []
if isinstance(i, Mapping) and i.get("level") == "error"]
t.path, t.reason = "build_failed", "生成的工作流没有通过校验:" + ";".join(errors)
return t
if step == policy.NEXT_READY:
t.path, t.reason = "ready", "助手只建了图、没有运行(它判断用户要的是流程本身)"
return t
await self._phase("run")
row = await self._run(graph)
if row is None:
return t
await self._phase("review")
reviewing = time.monotonic()
review = await self._review(row)
spent = int((time.monotonic() - reviewing) * 1000)
t.timings["review_ms"] = t.timings.get("review_ms", 0) + spent
t.review = {k: v for k, v in review.items() if k != "usage"}
t.usage["review"] = _add_usage(t.usage.get("review"), review.get("usage"))
if review.get("retry_instruction"):
from app.chat.turn import review_reason
t.attempts += 1
t.retries.append(review_reason(review) or "复核判答案不可用")
instruction = str(review["retry_instruction"])
base_graph = row.graph if policy.has_nodes(row.graph) else None
continue
t.path = "ran"
return t
finally:
t.timings["total_ms"] = int((time.monotonic() - started) * 1000)
async def _review(self, run: Any) -> dict[str, Any]:
"""复核(review_of_run,问数据页同一个函数)。复核自己出错不挡评分:照实记一条失败的复核。"""
from app.api.copilot import review_of_run
from app.core.errors import describe_exception
from app.db.base import SessionLocal
try:
async with SessionLocal() as session:
return await review_of_run(session, run, question=self.question,
provider=self.ctx.assistant.get("provider"),
model=self.ctx.assistant.get("model"), attempt=self.turn.attempts)
except Exception as e: # noqa: BLE001
logger.exception("评测:复核运行 %s 出错", run.id)
return {"verdict": "failed", "note": f"复核出错:{describe_exception(e)}", "signals": [],
"retry_instruction": None}
__all__ = ["APPROVAL_REASON", "Driver", "PHASE_TEXT", "REVIEW_PAGE_WAIT_S", "Turn", "TurnContext", "add_usage",
"run_name", "stop_run", "with_writer", "writer_usage"]
+14
View File
@@ -27,6 +27,8 @@ from app.api import (
blueprints as blueprints_api, copilot_blueprint, files as files_api, personal as personal_api,
templates as templates_api,
)
# 编排完善第 1 波(docs/design/orchestration-round §1):问数据的一轮在服务端跑
from app.api import chat_turns as chat_turns_api
from app.api.coded import CodedHTTPException, coded_handler
from app.api.deps import chat_scope, evidence_scope, private_run_scope, require_role, role_by_method, run_scope
from app.auth.keys import key_gate
@@ -132,6 +134,14 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
logger.exception("评测任务的启动整理失败,服务照常启动")
await run_manager.setup()
# 问数据的一轮(app/chat/job.py):在 run_manager 之后整理——服务重启时在跑的运行这时已经标成挂起。还在建图的标中断;
# 已经发起了运行的接着等它、照常复核收尾(和调度循环同一个开关:拷来的数据目录不替人调模型,等人打开对话时再接着收尾)
from app.chat import job as chat_job
try:
await chat_job.startup()
except Exception: # noqa: BLE001
logger.exception("问数据的启动整理失败,服务照常启动")
# 流程:存量的待审批补建待办(派给任何人,幂等);上次进程没了时已发出、没拿到结果的写入收成结果未知、安排核实
from app.process import actions as _actions, tasks as _tasks
@@ -163,6 +173,8 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
await catalog_jobs.manager.shutdown()
await catalog_profile_jobs.manager.shutdown()
await introspect_jobs.manager.shutdown()
# 问数据的一轮停下不改那几行(下次启动照重启的规矩整理),也要在运行之前停
await chat_job.shutdown()
# 评测要在运行之前停:它停下时会取消自己发起的作答、算期望的运行,再标中断
await eval_runner.shutdown()
await eval_snapshots.manager.shutdown()
@@ -262,6 +274,8 @@ for router, guard in (
(sandbox.router, _BUILDER),
(copilot.router, [chat_scope]),
(conversations.router, [chat_scope]),
# 问数据的一轮(发起、看、停):同问数据的门,接口里再按这一轮所属的对话核对归属
(chat_turns_api.router, [chat_scope]),
# 工件:user 角色只能取能读的运行引用的、遮罩后的(api/artifacts.py)
(artifacts.router, _USER),
(governance.router, _BUILDER),
+836
View File
@@ -0,0 +1,836 @@
"""问数据的一轮在服务端跑(docs/design/orchestration-round/00-总方案.md §1;app/chat/job.py、app/api/chat_turns.py)。
守的是:
- 一轮跑完(建图 → 运行 → 复核 → 落库)不依赖任何前端请求:发起之后只等后台任务,conversation_turns 那一行由服务端写好;
- 事件日志:每一步按 seq 记下,实时流先回放、再推实时,中途断开、按 seq 重连,补齐、不重复;思考增量合并成一条;
- 停止:建图中、运行中都能停,运行跟着取消,那一行收成「已取消」;
- 服务重启:建图中的标中断(照实说、可重试本轮);已发起运行的接着等那次运行、照常复核收尾;开关关着(拷来的数据目录)时停着,
页面请服务端接着收尾(mode=resume)才做;
- 按复核重来:复核给了 retry_instruction,带着这次运行的图重建一次,换下来的运行的用量并进这一轮;
- 用户点的重试本轮:上一次的结局归档进 meta.attempts,交给助手的话照以前页面上的措辞;
- 同一段对话一次只有一轮在跑(停在审批上等人的不算);
- 使用者(user 角色):只碰得到自己的对话的轮次;发起的运行照旧遮罩、只读、记忆是自己的范围;访问密钥调不了这些接口。
库、问题、名字都是虚构的;助手和报告撰写都是剧本,不调真接口。
"""
from __future__ import annotations
import asyncio
import json
from typing import Any
import pytest
from httpx import ASGITransport, AsyncClient
from sqlalchemy import delete, select
from app.chat import job as chat_job
from app.chat import log as turn_log
from app.db.base import SessionLocal
from app.db.models import Approval, Conversation, ConversationTurn, Run
from app.db.models_chat import ChatTurnEvent, ChatTurnJob
from app.engine.runner import run_manager
from app.main import app
from tests import eval_samples as es
from tests.fixtures.sources import drop_source
QUESTION = es.QUESTION
@pytest.fixture(autouse=True)
async def engine_up():
await run_manager.setup()
yield
await chat_job.shutdown()
await run_manager.shutdown()
@pytest.fixture
async def client():
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
yield c
@pytest.fixture
async def source(tmp_path):
name, sid = await es.make_source(tmp_path)
yield {"name": name, "id": sid}
await drop_source(sid)
async def _conversation(client, title: str = "虚构的问数据会话") -> str:
r = await client.post("/api/conversations", json={"kind": "chat", "title": title})
assert r.status_code == 201, r.text
return r.json()["id"]
async def _ask(client, conversation_id: str, question: str = QUESTION, **extra: Any) -> str:
r = await client.post("/api/chat/turns", json={"conversation_id": conversation_id, "question": question, **extra})
assert r.status_code == 202, r.text
body = r.json()
assert body["turn"]["status"] == "running" and body["turn"]["job"]["status"] in ("queued", "running")
return body["turn_id"]
async def _done(turn_id: str, timeout: float = 60.0) -> ChatTurnJob:
await chat_job.manager.wait(turn_id, timeout=timeout)
async with SessionLocal() as session:
job = await session.get(ChatTurnJob, turn_id)
assert job is not None and job.status not in ("queued", "running"), job and (job.status, job.stage)
return job
async def _row(turn_id: str) -> ConversationTurn:
async with SessionLocal() as session:
return await session.get(ConversationTurn, turn_id)
async def _events(turn_id: str) -> list[dict[str, Any]]:
return await turn_log.read(turn_id, 0, limit=10_000)
async def _until(cond, timeout: float = 20.0) -> None:
for _ in range(int(timeout / 0.05)):
if await cond():
return
await asyncio.sleep(0.05)
raise AssertionError("等不到条件成立")
async def _stage(turn_id: str, stage: str) -> None:
async def hit() -> bool:
async with SessionLocal() as session:
job = await session.get(ChatTurnJob, turn_id)
return job is not None and job.stage == stage
await _until(hit)
def _sse(text: str) -> list[dict[str, Any]]:
return [json.loads(line[6:]) for line in text.splitlines() if line.startswith("data: ")]
# --------------------------------------------------------------------------
# 一轮跑完不依赖页面
# --------------------------------------------------------------------------
async def test_a_turn_runs_to_the_end_without_the_page(client, monkeypatch, source):
"""发起之后页面一个请求都不再发:建图、运行、复核、落库都在服务端做完。"""
es.use_copilot(monkeypatch, es.Copilot(source["name"]))
es.use_report(monkeypatch)
conversation_id = await _conversation(client)
turn_id = await _ask(client, conversation_id, datasource_ids=[source["id"]])
job = await _done(turn_id)
assert job.status == "succeeded" and job.stage is None and job.run_id, (job.status, job.error)
row = await _row(turn_id)
assert row.status == "done", (row.status, row.error)
assert "330" in row.answer, row.answer
assert row.run_id == job.run_id and row.graph and row.graph.get("nodes")
meta = row.meta
assert meta["runId"] == job.run_id and meta["runClass"] == "exploratory" and meta["runStatus"] == "succeeded"
assert meta["queries"] == 1 and isinstance(meta["ms"], int)
assert meta["scope"] == [{"id": source["id"], "name": source["name"]}]
assert meta["looks"] and meta["looks"][0]["op"] == "look", "查看记录照页面以前的形状存进 meta.looks"
assert set(meta["usage"]) >= {"copilot", "run"}, "这一轮的花费 = 助手建图 + 运行(+ 复核)"
assert isinstance(row.review, dict) and row.review.get("verdict") == "ok", "复核原样落库(含 verdict=ok)"
async with SessionLocal() as session:
run = await session.get(Run, job.run_id)
assert run.origin_kind == "chat" and run.origin.get("conversation_id") == conversation_id
assert run.origin.get("turn_id") == turn_id and run.input.get("question") == QUESTION
events = await _events(turn_id)
seqs = [e["seq"] for e in events]
assert seqs == list(range(1, len(seqs) + 1)), "seq 从 1 起连续"
kinds = [e["type"] for e in events]
assert kinds[0] == "turn.attempt" and kinds[-1] == "end"
assert [e["data"]["stage"] for e in events if e["type"] == "stage"] == ["build", "run", "review"]
ops = [e["data"]["op"] for e in events if e["type"] == "op"]
assert {"model", "look", "add_node", "final", "usage"} <= set(ops), ops
[launched] = [e for e in events if e["type"] == "run"]
assert launched["data"]["run_id"] == job.run_id
end = events[-1]["data"]
assert end["outcome"] == "done" and end["turn"]["status"] == "done" and end["turn"]["answer"] == row.answer
assert end["run"]["id"] == job.run_id and end["output"], "end 带完整成果,页面不必再回源"
# 对话详情里这一轮带着它的任务;列表状态是 done
detail = (await client.get(f"/api/conversations/{conversation_id}")).json()
assert detail["turns"][-1]["job"]["status"] == "succeeded" and detail["last_status"] == "done"
state = (await client.get(f"/api/chat/turns/{turn_id}")).json()
assert state["job"]["status"] == "succeeded" and state["job"]["last_seq"] == seqs[-1]
assert state["job"]["attempt_seq"] == 1 and state["turn"]["id"] == turn_id
async def test_reply_and_ask_settle_like_the_page_did(client, monkeypatch, source):
"""直接回答:done、没查库(noQuery);助手反问:done,答案列是那句问题,meta.asked 记着问题和候选。"""
es.use_copilot(monkeypatch, es.Copilot(source["name"], mode="reply"))
conversation_id = await _conversation(client)
turn_id = await _ask(client, conversation_id, "刚才那个数是多少")
await _done(turn_id)
row = await _row(turn_id)
assert row.status == "done" and row.answer == "销售额是 330。" and row.meta["noQuery"] is True
assert (await _events(turn_id))[-1]["data"]["outcome"] == "reply"
es.use_copilot(monkeypatch, es.Copilot(source["name"], mode="ask"))
turn_id = await _ask(client, conversation_id, "上周销售额")
await _done(turn_id)
row = await _row(turn_id)
assert row.status == "done" and row.answer == "要按下单日还是支付日?"
assert row.meta["asked"]["options"] == ["下单日", "支付日"] and not row.meta.get("noQuery")
async def test_a_refusal_before_the_stream_is_recorded_as_its_cause(client, monkeypatch, source):
"""限定的数据源都不在了:开流之前就被拒,这一轮记 error,原因带机器码(页面据此给「不限数据源重试」)。"""
es.use_copilot(monkeypatch, es.Copilot(source["name"]))
conversation_id = await _conversation(client)
turn_id = await _ask(client, conversation_id, datasource_ids=["not-a-source"])
await _done(turn_id)
row = await _row(turn_id)
assert row.status == "error" and row.meta["cause"]["kind"] == "refused"
assert row.meta["cause"]["code"] == "datasource_scope_empty" and row.error.startswith("限定的数据源")
# --------------------------------------------------------------------------
# 事件日志、实时流
# --------------------------------------------------------------------------
class Gated(es.Copilot):
"""第一次起草前停在闸门上,放行之后照剧本走(测实时流中途接入、断开重连)。思考先吐几段增量。"""
def __init__(self, source: str, gate: asyncio.Event, **kw: Any) -> None:
super().__init__(source, **kw)
self.gate = gate
async def astream(self, messages):
from langchain_core.messages import AIMessageChunk
if len(self.calls) == 0:
for piece in ("先想一想", "要查哪张表", "……"):
yield AIMessageChunk(content="", additional_kwargs={"reasoning_content": piece})
await self.gate.wait()
async for chunk in super().astream(messages):
yield chunk
async def test_the_stream_replays_then_follows_and_reconnects_by_seq(client, monkeypatch, source):
"""订阅方中途接入:先回放已有的、再推实时的;断开之后按收到的最后一个 seq 重连,补齐、不重复,最后是 stream.end。"""
gate = asyncio.Event()
es.use_copilot(monkeypatch, Gated(source["name"], gate))
es.use_report(monkeypatch)
conversation_id = await _conversation(client)
turn_id = await _ask(client, conversation_id, datasource_ids=[source["id"]])
await _stage(turn_id, "build")
async def active() -> str | None:
return await chat_job.stream_status(turn_id)
first: list[dict[str, Any]] = []
stream = turn_log.stream(turn_id, 0, active=active, heartbeat=0.2)
async for frame in stream:
if frame.startswith("data: "):
first.append(json.loads(frame[6:]))
if any(e.get("type") == "stage" for e in first):
break
await stream.aclose() # 断开
assert first[0]["type"] == "turn.attempt" and first[0]["seq"] == 1
gate.set()
rest: list[dict[str, Any]] = []
pings = 0
async for frame in turn_log.stream(turn_id, first[-1]["seq"], active=active, heartbeat=0.2):
if frame == turn_log.PING:
pings += 1
continue
rest.append(json.loads(frame[6:]))
assert rest[-1]["type"] == "stream.end" and rest[-1]["status"] == "succeeded"
got = [e["seq"] for e in first + rest[:-1]]
assert got == list(range(1, len(got) + 1)), "补齐、不重复"
assert got[-1] == (await _events(turn_id))[-1]["seq"]
# 跑完之后按 seq 重连(HTTP):只给 seq 之后的,再给 stream.end
after = got[-3]
text = (await client.get(f"/api/chat/turns/{turn_id}/stream", params={"after": after})).text
frames = _sse(text)
assert [f["seq"] for f in frames[:-1]] == got[-2:] and frames[-1]["type"] == "stream.end"
assert [f["seq"] for f in _sse((await client.get(f"/api/chat/turns/{turn_id}/stream")).text)[:-1]] == got
# 思考增量合并成一条,不逐条落库
thinking = [e for e in await _events(turn_id) if e["type"] == "op" and e["data"]["op"] == "thinking"]
assert len(thinking) == 1 and thinking[0]["data"]["delta"] == "先想一想要查哪张表……", thinking
# --------------------------------------------------------------------------
# 停止
# --------------------------------------------------------------------------
async def test_stop_while_building(client, monkeypatch, source):
gate = asyncio.Event()
es.use_copilot(monkeypatch, Gated(source["name"], gate))
conversation_id = await _conversation(client)
turn_id = await _ask(client, conversation_id)
await _stage(turn_id, "build")
r = await client.post(f"/api/chat/turns/{turn_id}/cancel")
assert r.status_code == 200, r.text
job = await _done(turn_id)
assert job.status == "cancelled"
row = await _row(turn_id)
assert row.status == "error" and row.error == "已取消" and row.meta["outcome"] == "cancelled"
assert (await _events(turn_id))[-1]["data"]["outcome"] == "cancelled"
r = await client.post(f"/api/chat/turns/{turn_id}/cancel")
assert r.status_code == 409 and r.json()["code"] == "chat_turn_finished"
async def test_stop_while_running_cancels_the_run(client, monkeypatch, source):
"""运行中停止:运行跟着取消(不只是断开订阅——以前点了停止,模型和 SQL 其实还在跑、还在计费)。"""
from app.providers import mock_model
es.use_copilot(monkeypatch, es.Copilot(source["name"]))
hold = asyncio.Event()
async def slow(self, messages, stop=None, run_manager=None, **kwargs):
await hold.wait()
raise AssertionError("不该放行")
async def slow_stream(self, messages, stop=None, run_manager=None, **kwargs):
await hold.wait()
raise AssertionError("不该放行")
yield # noqa: unreachable - 让它成为异步生成器
# 报告撰写卡在模型调用上(流式、非流式都卡住):运行一直在跑,等着被停止
monkeypatch.setattr(mock_model.MockChatModel, "_agenerate", slow)
monkeypatch.setattr(mock_model.MockChatModel, "_astream", slow_stream)
conversation_id = await _conversation(client)
turn_id = await _ask(client, conversation_id, datasource_ids=[source["id"]])
await _stage(turn_id, "run")
async with SessionLocal() as session:
run_id = (await session.get(ChatTurnJob, turn_id)).run_id
assert (await client.post(f"/api/chat/turns/{turn_id}/cancel")).status_code == 200
job = await _done(turn_id)
assert job.status == "cancelled"
await run_manager.wait_idle(run_id)
async with SessionLocal() as session:
assert (await session.get(Run, run_id)).status == "cancelled"
row = await _row(turn_id)
assert row.status == "error" and row.error == "已取消" and row.meta["runStatus"] == "cancelled"
# --------------------------------------------------------------------------
# 服务重启
# --------------------------------------------------------------------------
async def _stranded(conversation_id: str, *, run: Run | None, stage: str, mode: str = "ask") -> str:
"""造一轮「上次进程没了时还在跑」的:那一行 running、任务 running(不在这个进程里)。"""
async with SessionLocal() as session:
if run is not None:
session.add(run)
await session.flush()
turn = ConversationTurn(conversation_id=conversation_id, seq=0, question="上周各区域销售额", status="running",
run_id=run.id if run else None, meta={"v": 1, "runId": run.id if run else None})
session.add(turn)
await session.flush()
session.add(ChatTurnJob(id=turn.id, conversation_id=conversation_id, status="running", stage=stage, mode=mode,
run_id=run.id if run else None, request={"started_ms": 0}, principal={}))
await session.commit()
return turn.id
async def test_restart_interrupts_builds_and_finishes_launched_runs(client, monkeypatch):
conversation_id = await _conversation(client)
building = await _stranded(conversation_id, run=None, stage="build")
finished = await _stranded(conversation_id, run=Run(status="succeeded", output={"答案": "华东 300。"},
graph={"nodes": [], "edges": []}, run_class="exploratory",
origin_kind="chat"), stage="review")
suspended = await _stranded(conversation_id, run=Run(status="interrupted", output={}, graph={"nodes": [], "edges": []},
run_class="exploratory", origin_kind="chat"), stage="run")
counts = await chat_job.startup(resume=True)
assert counts["interrupted"] >= 1 and counts["resumed"] >= 2
for turn_id in (finished, suspended):
await chat_job.manager.wait(turn_id, timeout=30)
row = await _row(building)
assert row.status == "error" and row.meta["cause"] == {"kind": "build_interrupted"}, "建图中的照实标中断"
async with SessionLocal() as session:
assert (await session.get(ChatTurnJob, building)).status == "interrupted"
assert (await _events(building))[-1]["data"]["outcome"] == "error"
row = await _row(finished)
assert row.status == "done" and row.answer == "华东 300。", "已发起运行的:接着等那次运行、照常复核收尾"
assert row.review.get("verdict") == "ok"
row = await _row(suspended)
assert row.status == "error" and row.meta["outcome"] == "suspended" and row.error == "服务重启,本轮已中断"
async def test_with_the_switch_off_launched_runs_wait_for_someone_to_open_the_conversation(client):
"""拷来的数据目录(AGENTLAB_SCHEDULER=0):启动时不替人收尾,那一行不动;页面请服务端接着收尾(resume)才做。"""
conversation_id = await _conversation(client)
turn_id = await _stranded(conversation_id, run=Run(status="succeeded", output={"答案": "华北 30。"},
graph={"nodes": [], "edges": []}, run_class="exploratory",
origin_kind="chat"), stage="review")
counts = await chat_job.startup(resume=False)
assert counts["parked"] >= 1
assert (await _row(turn_id)).status == "running"
async with SessionLocal() as session:
assert (await session.get(ChatTurnJob, turn_id)).status == "interrupted"
r = await client.post("/api/chat/turns", json={"conversation_id": conversation_id, "retry_of": turn_id,
"mode": "resume"})
assert r.status_code == 202, r.text
await _done(turn_id)
row = await _row(turn_id)
assert row.status == "done" and row.answer == "华北 30。"
assert not row.meta.get("attempts"), "接着收尾不是重试:不归档"
async def test_a_legacy_turn_stuck_building_is_marked_interrupted_at_startup(client):
"""这一波之前页面推的老轮次(没有任务)停在 running、又没有运行的:它的建图那条流随上次的进程一起没了。"""
conversation_id = await _conversation(client)
async with SessionLocal() as session:
turn = ConversationTurn(conversation_id=conversation_id, seq=0, question="老轮次", status="running")
session.add(turn)
await session.commit()
legacy = turn.id
await chat_job.startup(resume=False)
row = await _row(legacy)
assert row.status == "error" and row.meta["cause"] == {"kind": "build_interrupted"}
async def test_a_turn_waits_for_the_approver_and_finishes_after(client, monkeypatch):
"""停在审批上:照旧等人(评测是取消)。审批过了运行接着跑,这一轮照常复核收尾。"""
conversation_id = await _conversation(client)
run = Run(status="interrupted", output={}, graph={"nodes": [], "edges": []}, run_class="exploratory", origin_kind="chat")
turn_id = await _stranded(conversation_id, run=run, stage="run")
async with SessionLocal() as session:
session.add(Approval(run_id=run.id, node_id="ask", mode="approve", title="确认一下", payload={}, status="pending"))
await session.commit()
monkeypatch.setattr(chat_job, "APPROVAL_POLL_S", 0.1)
await chat_job.startup(resume=True)
await _stage(turn_id, "waiting")
detail = (await client.get(f"/api/conversations/{conversation_id}")).json()
assert detail["last_status"] == "waiting"
# 别的轮次照样能问(停在审批上的不算占着这段对话)
async with SessionLocal() as session:
busy = await chat_job._busy(session, conversation_id)
assert busy is False
# 审批人处理了、运行跑完了
async with SessionLocal() as session:
row = await session.get(Run, run.id)
row.status, row.output = "succeeded", {"答案": "审批之后的答案"}
approval = (await session.execute(select(Approval).where(Approval.run_id == run.id))).scalar_one()
approval.status = "approved"
await session.commit()
await _done(turn_id)
row = await _row(turn_id)
assert row.status == "done" and row.answer == "审批之后的答案"
stages = [e["data"]["stage"] for e in await _events(turn_id) if e["type"] == "stage"]
assert stages[-3:] == ["waiting", "run", "review"], stages
# --------------------------------------------------------------------------
# 重来:按复核自动重来一次、用户点的重试本轮
# --------------------------------------------------------------------------
async def test_the_review_retries_once_with_the_run_graph(client, monkeypatch, source):
import app.api.copilot as copilot
from app.chat import turn as policy
model = es.Copilot(source["name"])
es.use_copilot(monkeypatch, model)
es.use_report(monkeypatch)
real = copilot.review_of_run
attempts: list[int] = []
async def review(session, run, *, question, attempt=0, **kw):
attempts.append(attempt)
out = await real(session, run, question=question, attempt=attempt, **kw)
if attempt == 0:
out = {**out, "verdict": "annotated", "severity": "broken", "note": "查到一半就停了", "retry": True,
"signals": [{"kind": "step_limit", "detail": "用满了步数", "severity": "broken"}],
"retry_instruction": policy.retry_instruction(question, "用满了步数", True),
"usage": {"calls": 1, "cost_usd": 0.003}}
return out
monkeypatch.setattr(copilot, "review_of_run", review)
conversation_id = await _conversation(client)
turn_id = await _ask(client, conversation_id, datasource_ids=[source["id"]])
await _done(turn_id)
assert attempts == [0, 1], "第二次复核按第二次算,不再给重来"
events = await _events(turn_id)
[retry] = [e["data"] for e in events if e["type"] == "turn.retry"]
assert retry["kind"] == "review" and retry["reason"] == "用满了步数" and retry["base"] is True
runs = [e["data"]["run_id"] for e in events if e["type"] == "run"]
assert len(runs) == 2 and runs[0] != runs[1]
row = await _row(turn_id)
assert row.status == "done" and row.meta["runId"] == runs[1]
usage = row.meta["usage"]
assert usage["review"]["calls"] == 1 and usage["review"]["cost_usd"] == 0.003, "复核那一次的用量记进这一轮"
assert "earlier" in usage, "换下来的那次运行的用量并进 earlier"
# 重来那一次交给助手的就是复核给的那句话,带着上一次运行的图
last = model.calls[-2] if len(model.calls) >= 2 else model.calls[-1]
assert any("这个工作流上一次运行没有成功:用满了步数" in str(m[1]) for m in last)
async def test_retrying_a_turn_archives_the_previous_attempt(client, monkeypatch, source):
"""用户点「重试本轮」:上一次的结局归档进 meta.attempts(照以前页面上的说法),交给助手的话照以前的措辞,那一行回到 running。"""
es.use_copilot(monkeypatch, es.Copilot(source["name"], mode="reply"))
conversation_id = await _conversation(client)
turn_id = await _ask(client, conversation_id, "上周各区域销售额")
await _done(turn_id)
seen: list[str] = []
real = chat_job.ChatDriver.generate_in
def spy(self, instruction, base_graph):
seen.append(instruction)
return real(self, instruction, base_graph)
monkeypatch.setattr(chat_job.ChatDriver, "generate_in", spy)
r = await client.post("/api/chat/turns", json={"conversation_id": conversation_id, "retry_of": turn_id,
"mode": "retry"})
assert r.status_code == 202, r.text
assert r.json()["turn"]["status"] == "running"
await _done(turn_id)
row = await _row(turn_id)
[archived] = row.meta["attempts"]
assert archived["n"] == 1 and archived["outcome"] == "done" and archived["summary"] == "已给出答案"
assert archived["answer"] == "销售额是 330。"
assert seen == ["上周各区域销售额\n\n(上一次没有成功:已给出答案。请针对这个原因调整,重新给出完整的工作流。)"]
attempt_events = [e for e in await _events(turn_id) if e["type"] == "turn.attempt"]
assert [e["data"]["mode"] for e in attempt_events] == ["ask", "retry"]
assert attempt_events[-1]["data"]["attempts"] == row.meta["attempts"]
async with SessionLocal() as session:
job = await session.get(ChatTurnJob, turn_id)
assert job.attempt_seq == attempt_events[-1]["seq"], "刷新时从这一次尝试的第一条回放"
async def test_one_turn_at_a_time_per_conversation(client, monkeypatch, source):
gate = asyncio.Event()
es.use_copilot(monkeypatch, Gated(source["name"], gate))
conversation_id = await _conversation(client)
turn_id = await _ask(client, conversation_id)
r = await client.post("/api/chat/turns", json={"conversation_id": conversation_id, "question": "再问一句"})
assert r.status_code == 409 and r.json()["code"] == "chat_turn_busy"
r = await client.post("/api/chat/turns", json={"conversation_id": conversation_id, "retry_of": turn_id})
assert r.status_code == 409
other = await _conversation(client, "另一段对话")
other_turn = await _ask(client, other, "别的对话照常问")
for t in (turn_id, other_turn):
await client.post(f"/api/chat/turns/{t}/cancel")
await _done(t)
async def test_continuing_a_finished_run_is_refused_like_the_runs_api(client):
conversation_id = await _conversation(client)
run = Run(status="succeeded", output={"答案": "x"}, graph={"nodes": [], "edges": []}, run_class="exploratory")
async with SessionLocal() as session:
session.add(run)
await session.flush()
turn = ConversationTurn(conversation_id=conversation_id, seq=0, question="q", status="done", answer="x",
run_id=run.id, meta={"v": 1, "runId": run.id})
session.add(turn)
await session.commit()
turn_id = turn.id
r = await client.post("/api/chat/turns", json={"conversation_id": conversation_id, "retry_of": turn_id,
"mode": "continue"})
assert r.status_code == 409 and "无需继续运行" in r.json()["detail"], r.text
assert (await _row(turn_id)).status == "done", "被拒时那一行不动"
async def test_deleting_a_conversation_stops_its_turn(client, monkeypatch, source):
gate = asyncio.Event()
es.use_copilot(monkeypatch, Gated(source["name"], gate))
conversation_id = await _conversation(client)
turn_id = await _ask(client, conversation_id)
await _stage(turn_id, "build")
assert (await client.delete(f"/api/conversations/{conversation_id}")).status_code == 204
assert not chat_job.manager.running(turn_id)
async with SessionLocal() as session:
assert await session.get(ChatTurnJob, turn_id) is None
assert not (await session.execute(select(ChatTurnEvent).where(ChatTurnEvent.turn_id == turn_id))).first()
# --------------------------------------------------------------------------
# 使用者(user 角色):归属、遮罩、只读、访问密钥
# --------------------------------------------------------------------------
@pytest.fixture
async def people():
from tests.fakes import auth_env
await auth_env.reset()
admin = await auth_env.turn_on_local()
assert (await admin.put("/api/auth/settings", json={"user_can_chat": True})).status_code == 200
for name, role in (("uma", "user"), ("ula", "user"), ("bob", "builder")):
await auth_env.make_user(name, role)
clients = {name: await auth_env.login(name) for name in ("uma", "ula", "bob")}
try:
yield {"admin": admin, **clients}
finally:
async with SessionLocal() as s:
ids = list((await s.execute(select(Conversation.id).where(
Conversation.created_by.in_(["uma", "ula", "bob"])))).scalars())
await s.execute(delete(ConversationTurn).where(ConversationTurn.conversation_id.in_(ids)))
await s.execute(delete(Conversation).where(Conversation.id.in_(ids)))
await s.execute(delete(Run).where(Run.started_by.in_(["uma", "ula", "bob"])))
await s.commit()
await auth_env.reset()
async def test_a_users_turns_are_theirs_and_their_runs_stay_masked_and_read_only(people, monkeypatch, source):
from app.tools.datasource import MASK_RESULTS_OPTION, READ_ONLY_OPTION
es.use_copilot(monkeypatch, es.Copilot(source["name"]))
es.use_report(monkeypatch)
uma, ula, bob = people["uma"], people["ula"], people["bob"]
mine = await _conversation(uma, "uma 的对话")
theirs = await _conversation(ula, "ula 的对话")
turn_id = await _ask(uma, mine, datasource_ids=[source["id"]])
job = await _done(turn_id)
assert job.status == "succeeded", job.error
async with SessionLocal() as session:
run = await session.get(Run, job.run_id)
assert run.started_by == "uma" and run.origin_kind == "chat"
assert run.options.get(MASK_RESULTS_OPTION) is True and run.options.get(READ_ONLY_OPTION) is True
assert run.memory_scope == "chat:uma", "记忆固定为这位使用者自己的范围"
assert (await _row(turn_id)).status == "done"
# 只碰得到自己的对话的轮次
for path in (f"/api/chat/turns/{turn_id}", f"/api/chat/turns/{turn_id}/stream"):
assert (await ula.get(path)).status_code == 403, path
assert (await uma.get(path)).status_code == 200, path
assert (await ula.post(f"/api/chat/turns/{turn_id}/cancel")).status_code == 403
r = await uma.post("/api/chat/turns", json={"conversation_id": theirs, "question": "偷看别人的对话"})
assert r.status_code == 403
r = await ula.post("/api/chat/turns", json={"conversation_id": mine, "retry_of": turn_id})
assert r.status_code == 403
# 搭建者照旧看得到(对话按人隔离只收窄使用者)
assert (await bob.get(f"/api/chat/turns/{turn_id}")).status_code == 200
# 管理员关了问数据:使用者整组接口都用不了
assert (await people["admin"].put("/api/auth/settings", json={"user_can_chat": False})).status_code == 200
r = await uma.post("/api/chat/turns", json={"conversation_id": mine, "question": "再问一句"})
assert r.status_code == 403 and r.json()["code"] == "chat_forbidden"
async def test_a_user_cannot_launch_a_graph_outside_the_chat_whitelist(people, monkeypatch, source):
"""使用者的图里有白名单之外的节点:自查挡下(blocked)或发起被拒(start),这一轮照实记 error,不会跑起来。"""
class CodeNode(es.Copilot):
async def astream(self, messages):
from tests.eval_samples import _line
self.calls.append(list(messages))
if "上线前自查" in messages[1][1]:
yield _line({"op": "done"})
return
for op in ({"op": "add_node", "node": {"id": "start", "type": "input", "label": "输入", "config": {}}},
{"op": "add_node", "node": {"id": "c", "type": "code", "label": "算一下",
"config": {"code": "result = 1"}}},
{"op": "add_edge", "edge": {"source": "start", "target": "c"}},
{"op": "add_node", "node": {"id": "out", "type": "output", "label": "答案", "config": {}}},
{"op": "add_edge", "edge": {"source": "c", "target": "out"}},
{"op": "done", "explanation": "", "run": True}):
yield _line(op)
es.use_copilot(monkeypatch, CodeNode(source["name"]))
conversation_id = await _conversation(people["uma"], "uma 的对话")
turn_id = await _ask(people["uma"], conversation_id)
await _done(turn_id)
row = await _row(turn_id)
assert row.status == "error", (row.status, row.error)
cause = row.meta.get("cause") or {}
assert cause.get("kind") in ("blocked", "start"), cause
async def test_access_keys_cannot_reach_the_chat_turn_api(people):
"""访问密钥照旧调不了问数据(不在 auth/keys.KEY_ROUTES 里):发起、看、停一律 403 key_route_forbidden。"""
from app.db.models_identity import User
from tests.fakes import auth_env
await auth_env.personal_keys(True)
async with SessionLocal() as s:
uma = (await s.execute(select(User).where(User.username == "uma"))).scalar_one()
_, secret = await auth_env.make_key(uma)
keyed = auth_env.keyed(secret)
for method, path in (("post", "/api/chat/turns"), ("get", "/api/chat/turns/x"),
("get", "/api/chat/turns/x/stream"), ("post", "/api/chat/turns/x/cancel")):
r = await getattr(keyed, method)(path, **({"json": {"conversation_id": "c"}} if method == "post" else {}))
assert r.status_code == 403 and r.json().get("code") == "key_route_forbidden", (path, r.text)
# --------------------------------------------------------------------------
# 各种结局照以前页面落库的写法
# --------------------------------------------------------------------------
async def _seed_events(run_id: str, *events: tuple[str, dict[str, Any]]) -> None:
import time as _time
from app.db.models import RunEvent
async with SessionLocal() as session:
for seq, (kind, data) in enumerate(events, start=1):
session.add(RunEvent(run_id=run_id, seq=seq, type=kind, ts=_time.time(), data=data))
await session.commit()
async def test_rejected_failed_and_refused_runs_settle_with_their_cause(client, monkeypatch, source):
conversation_id = await _conversation(client)
rejected = Run(status="cancelled", end_reason="rejected", output={}, graph={"nodes": [], "edges": []},
run_class="exploratory", error="人工驳回:数字对不上")
failed = Run(status="failed", output={}, graph={"nodes": [], "edges": []}, run_class="exploratory", error="查询超时")
r_id = await _stranded(conversation_id, run=rejected, stage="run")
f_id = await _stranded(conversation_id, run=failed, stage="run")
await _seed_events(rejected.id, ("run.cancelled", {"reason": "rejected", "actor": "张工", "label": "查数",
"note": "数字对不上", "decision": "reject"}))
await _seed_events(failed.id, ("run.failed", {"error": "查询超时", "detail": "TimeoutError: 30s"}))
await chat_job.startup(resume=True)
for t in (r_id, f_id):
await chat_job.manager.wait(t, timeout=30)
row = await _row(r_id)
assert row.status == "error" and row.error == "已取消" and row.meta["outcome"] == "cancelled"
assert row.meta["rejected"] == {"actor": "张工", "label": "查数", "note": "数字对不上", "decision": "reject"}
detail = (await client.get(f"/api/conversations/{conversation_id}")).json()
assert [t for t in detail["turns"] if t["id"] == r_id][0]["meta"]["rejected"]["actor"] == "张工"
row = await _row(f_id)
assert row.status == "error" and row.error == "查询超时"
assert row.meta["cause"] == {"kind": "run", "error": "查询超时", "detail": "TimeoutError: 30s"}
# 发起被拒:照 LaunchRefused 记下说法、机器码、状态码
import app.engine.launch as launch
from app.engine.launch import LaunchRefused
async def refuse(*_a, **_k):
raise LaunchRefused(422, "绑定的工具不存在:「查订单」(调用工具)绑定的 db_query__nope。请到「数据」页接入", "run_tool_missing")
monkeypatch.setattr(launch, "launch_run", refuse)
es.use_copilot(monkeypatch, es.Copilot(source["name"]))
other = await _conversation(client, "发起被拒")
turn_id = await _ask(client, other, datasource_ids=[source["id"]])
await _done(turn_id)
row = await _row(turn_id)
assert row.status == "error" and row.meta["cause"]["kind"] == "start"
assert row.meta["cause"]["code"] == "run_tool_missing" and row.meta["cause"]["status"] == 422
assert row.meta.get("runId") is None
async def test_a_late_review_still_counts_toward_the_turn(client, monkeypatch, source):
"""复核慢过 REVIEW_WAIT_S:先交原答案(不等它),那一次复核晚到了照样把用量记进这一轮(review.late)。"""
import app.api.copilot as copilot
es.use_copilot(monkeypatch, es.Copilot(source["name"]))
es.use_report(monkeypatch)
monkeypatch.setattr(chat_job, "REVIEW_WAIT_S", 0.05)
release = asyncio.Event()
async def slow(session, run, **kw):
await release.wait()
return {"verdict": "ok", "signals": [], "retry_instruction": None, "usage": {"calls": 1, "cost_usd": 0.004}}
monkeypatch.setattr(copilot, "review_of_run", slow)
conversation_id = await _conversation(client)
turn_id = await _ask(client, conversation_id, datasource_ids=[source["id"]])
await _done(turn_id)
row = await _row(turn_id)
assert row.status == "done" and "330" in row.answer and row.review == {}, "等不及:原答案照交、没有复核结论"
assert "review" not in row.meta["usage"]
assert chat_job.late_pending(turn_id) and await chat_job.stream_status(turn_id) is None, "实时流等复核回来再关"
release.set()
await _until(lambda: _not_late(turn_id))
row = await _row(turn_id)
assert row.meta["usage"]["review"] == {"calls": 1, "cost_usd": 0.004}
assert (await _events(turn_id))[-1]["type"] == "review.late"
assert await chat_job.stream_status(turn_id) == "succeeded"
async def _not_late(turn_id: str) -> bool:
return not chat_job.late_pending(turn_id)
async def test_run_and_rerun_reuse_the_built_graph_without_rebuilding(client, monkeypatch, source):
"""只建不跑的那一轮点「运行」:直接跑那张图,不归档、不重建;重新运行本轮:上一次归档,原图再跑一次。两种都不按复核
自动重来(同以前页面上 launch 不带 retry)。"""
import app.api.copilot as copilot
model = es.Copilot(source["name"])
es.use_copilot(monkeypatch, model)
es.use_report(monkeypatch)
conversation_id = await _conversation(client)
turn_id = await _ask(client, conversation_id, datasource_ids=[source["id"]])
await _done(turn_id)
built = (await _row(turn_id)).graph
async with SessionLocal() as session:
row = await session.get(ConversationTurn, turn_id)
row.meta = {k: v for k, v in (row.meta or {}).items() if k not in ("runId",)} | {"runId": None}
row.run_id, row.answer = None, ""
await session.commit()
calls = len(model.calls)
real = copilot.review_of_run
async def review(session, run, *, question, attempt=0, **kw):
out = await real(session, run, question=question, attempt=attempt, **kw)
return {**out, "retry_instruction": "(复核说要重来)"}
monkeypatch.setattr(copilot, "review_of_run", review)
r = await client.post("/api/chat/turns", json={"conversation_id": conversation_id, "retry_of": turn_id, "mode": "run"})
assert r.status_code == 202, r.text
await _done(turn_id)
row = await _row(turn_id)
assert row.status == "done" and "330" in row.answer and not row.meta.get("attempts")
assert len(model.calls) == calls, "不重建图"
assert not [e for e in await _events(turn_id) if e["type"] == "turn.retry"], "不按复核自动重来"
r = await client.post("/api/chat/turns", json={"conversation_id": conversation_id, "retry_of": turn_id,
"mode": "rerun"})
assert r.status_code == 202, r.text
await _done(turn_id)
row = await _row(turn_id)
assert row.status == "done" and len(row.meta["attempts"]) == 1 and row.graph == built
assert len(model.calls) == calls, "重新运行本轮不重建图"
r = await client.post("/api/chat/turns", json={"conversation_id": conversation_id, "retry_of": turn_id,
"mode": "max_steps"})
assert r.status_code == 422 and r.json()["code"] == "chat_turn_invalid", "放宽到几步要写清楚"
def test_max_steps_only_touches_agent_nodes():
from app.chat import record
graph = {"nodes": [{"id": "a", "type": "agent", "data": {"config": {"max_steps": 8, "tools": ["x"]}}},
{"id": "o", "type": "output", "data": {"config": {}}}], "edges": []}
out = record.with_max_steps(graph, 40)
assert out["nodes"][0]["data"]["config"] == {"max_steps": 40, "tools": ["x"]}
assert out["nodes"][1] == graph["nodes"][1] and graph["nodes"][0]["data"]["config"]["max_steps"] == 8
def test_attempt_records_say_what_the_page_used_to_say():
"""归档那一次的说法同以前页面上的 attemptOf:已手动停止、服务重启、被驳回(谁在哪一步)、结论不可用、存在缺口、失败原文。"""
from types import SimpleNamespace
from app.chat import record
def row(**kw):
base = {"status": "done", "error": "", "answer": "答案", "review": {}, "run_id": "r1", "meta": None,
"updated_at": None}
return SimpleNamespace(**{**base, **kw})
assert record.attempt_record(row(status="error", error="已取消"), {"outcome": "cancelled"}, n=1)["summary"] == "已手动停止"
rej = record.attempt_record(row(status="error", error="已取消"),
{"outcome": "cancelled", "rejected": {"actor": "张工", "label": "查数", "note": "不对"}}, n=1)
assert rej["outcome"] == "rejected" and rej["summary"] == "被驳回:张工 驳回了「查数」:不对"
old = record.attempt_record(row(status="error"), {"outcome": "cancelled", "rejected": {"sub": "页面记过的那句"}}, n=1)
assert old["summary"] == "被驳回:页面记过的那句"
assert record.attempt_record(row(status="error"), {"outcome": "suspended"}, n=2)["summary"] == "服务重启,运行中断"
broken = record.attempt_record(row(review={"severity": "broken", "note": "查到一半",
"signals": [{"detail": "用满了步数"}]}), {}, n=1)
assert (broken["outcome"], broken["summary"], broken["detail"]) == ("unusable", "结论不可用:查到一半", "用满了步数")
assert record.attempt_reason(broken) == "结论不可用:查到一半(用满了步数)"
assert record.attempt_record(row(review={"severity": "degraded", "note": "检索降级"}), {}, n=1)["summary"] == "存在缺口:检索降级"
failed = record.attempt_record(row(status="error", error="查询超时"), {"cause": {"kind": "run", "error": "查询超时"}}, n=1)
assert failed["outcome"] == "failed" and failed["summary"] == "查询超时" and failed["cause"]["kind"] == "run"
legacy = record.attempt_record(row(status="error", error="x"), {"failure": {"title": "操作超时", "reason": "查询超时。"}}, n=1)
assert legacy["summary"] == "操作超时:查询超时。"
assert record.attempt_record(row(answer="字" * 5000), {}, n=1)["answer"].endswith("…")