状态图里要调模型、写数据库，怎么让它能单独测试？

简化版

核心是让图只管「什么时候调哪个动作」，不亲自调模型、不亲自写库。调模型、写库这些有副作用的动作做成函数，从外面注入进去（放进初始 State，或作为回调参数传入）；测试时换成假函数，按脚本依次返回「通过 / 不通过」「相关 / 不相关」，一条脚本就编排出一条执行路径。这样图的测试不连数据库、不调模型，毫秒级跑完，能把首轮通过、改完复审、到上限收尾、中途中断、从断点续跑这些路径逐条覆盖；断言的是调用了哪些动作、按什么顺序、最终 State 是什么。落库和模型解析另外用服务层测试覆盖，真实模型的端到端测试保留少量，专门抓偶发的格式问题。

详细版

生产：execute_run() 准备材料 → 把 review / refine / finalize / should_stop 四个函数放进初始 State → graph.ainvoke()
测试：直接构造初始 State，四个位置放假函数 → graph.ainvoke() → 检查假函数被调用的顺序和参数

测试层	替换什么	验证什么	速度
图测试	所有动作换成假函数	分支、循环、上限、中断、续跑的走向	毫秒级
服务层测试	只把模型调用换成 mock	运行记录、步骤记录、状态字段写得对不对	秒级
端到端测试	什么都不换	真实模型的输出能否通过结构校验	分钟级，有费用

两种注入方式：

动作注入：图里的节点调用 state["review"](...)，真实实现由服务层用闭包包好传进来；
回调注入：节点照常做业务，但每做完一步调用 state["on_step"](...) 记录，落库由外面注入的记录器完成。

完整版教学

一、为什么状态图天生难测

状态图的价值在于分支和回路，而这恰恰是最需要测试的部分：资料不相关时会不会回到改写？到了上限会不会收尾？中断之后会不会停？但如果节点里直接调模型、直接写库，测一条路径就需要：

一个能连的数据库 + 初始化数据
一个有余额的模型 Key
让模型「恰好」在第 1 轮说不通过、第 2 轮说通过——做不到稳定复现

用真实模型测分支，结果取决于模型当次的输出，这次过、下次不过，测试本身就不可信。所以要先把「控制流」和「副作用」拆开：控制流是确定的，可以精确测；副作用是不确定或有成本的，要么替换，要么放到少量专门的测试里。

二、动作注入：图只决定调用时机

最彻底的做法是让图完全不认识数据库和模型。节点只从 State 里拿出一个函数来调用：

async def review_node(state):
    if await state["should_stop"]():          # 注入：是否已被中断
        return {"interrupted": True}
    round_no = state.get("round_no", 0) + 1
    result = await state["review"](state["working"], round_no)   # 注入：审查动作
    issues = result.get("issues") or []
    return {"round_no": round_no, "issues": issues,
            "passed": bool(result.get("passed")) or not issues}

生产环境里，服务层用闭包把 review 包好：它内部调用模型、写步骤记录、更新运行记录；图对这些一无所知。测试时换成一个按脚本返回结果的假函数，同时记下自己被调用的参数。这种写法的代价是 State 里混进了函数，不能直接序列化；如果需要把 State 快照落库，要把函数字段排除掉。

记忆钩子：图是导演，动作是演员。测试导演，只要换一批「按剧本念台词」的替身演员。

三、回调注入：业务留在节点，记录交给外面

另一种常见需求是「每一步都要落库」，但节点里的业务（检索、判断、生成）又希望保留。这时注入一个记录回调：

class StepRecorder:
    def __init__(self, run_id):
        self.run_id, self.index = run_id, 0
    async def __call__(self, node, name, request, response):
        self.index += 1                      # 跨节点累计步号
        await save_step(self.run_id, self.index, node, name, request, response)

实现了 __call__ 的对象可以像函数一样被调用，直接作为 on_step 放进 State。做成类而不是普通函数，是因为要跨多次调用累计步号。测试时注入一个「往列表里追加」的函数，跑完检查列表里记下了哪些节点、各几次，图的测试就和数据库彻底脱钩了。

四、假模型按脚本编排路径

节点里仍然调模型的情况，可以把模型调用换成一个按顺序吐出预设回复的假模型。一条脚本就是一条路径：

脚本：["不相关", "换个说法的问题", "相关", "答案", "有支撑"]
路径：检索 → 评分(不相关) → 改写 → 检索 → 评分(相关) → 生成 → 校验(有支撑) → 结束
断言：检索发生 2 次、改写发生 2 次（第一轮 + 重试）、结束原因为「正常回答」

脚本的每一项对应一次模型调用的返回，顺序必须和图的执行顺序一致。脚本写错时测试会在「假模型没有更多回复」处失败，这本身也是一种检查：说明图多调了一次模型，或者少调了一次。

五、按路径设计用例，数一数有多少条

有回路的图，路径条数是有限的，可以数出来逐条覆盖。以「审查 → 不通过就重写 → 复审」为例，最多 2 轮时：

编号	路径	期望结果
1	审查(过) → 汇总	1 次模型调用，没有修订稿
2	审查(不过) → 重写 → 审查(过) → 汇总	3 次调用
3	审查(不过) → 重写 → 审查(不过) → 重写 → 汇总	4 次调用，到上限收尾
4	开始前已被中断	0 次调用，直接结束
5	第 2 轮审查前被中断	2 次调用后结束，不写汇总
6	从「已完成 1 轮」续跑	第一次审查的轮次号是 2

再加一条「最多 5 轮」检查递归上限设得够不够，一共七八条用例就把控制流覆盖完了。断言时比较完整的调用序列最可靠，例如 [("review", 1), ("refine", 1), ("review", 2), ("finalize",)]，顺序、次数、参数一次比对清楚。

六、分三层，各管一段

图测试只证明「走向对」，不能证明「落库对」和「模型输出能用」，所以还要两层：

层	替换	用例数（示意）	单次耗时（示意）
图测试	全部动作	6–10 条	几毫秒
服务层测试	只替换模型调用（如 `AsyncMock`）	5–10 条	几百毫秒
端到端	不替换	1–2 条	按一轮 4 次调用、每次约 3 秒估算，约 12 秒

服务层测试从真实入口进去，检查运行记录的状态、轮次、步骤条数有没有写对；端到端测试用真实 Key 跑完整流程，专门抓单测抓不到的问题，比如模型偶发漏掉一个字段、结构校验不过。数量上是金字塔：底层多、顶层少。

七、常见误区与追问

误区：状态图只能用真实模型测。 把动作注入进来，图的走向可以用假函数精确测试，真实模型只留给少量端到端。
误区：用真实模型测分支是可靠的。 模型输出每次可能不同，同一条用例这次过下次不过，测试本身不可信。
误区：图测试通过就说明功能没问题。 图测试只证明走向，落库和模型输出要靠服务层和端到端测试。
误区：注入的函数可以随 State 一起序列化。 State 里有函数就不能直接转 JSON，快照落库时要排除这些字段。
误区：假模型脚本随便写几条就行。 脚本顺序必须和执行顺序一致，路径要按分支和上限逐条列出来，才谈得上覆盖。
追问：断言最终结果就够了，为什么还要断言调用序列？ 同样的最终结果可能来自不同的路径，比如多重写了一轮也能通过，只有调用序列能看出流程是否按预期走。
追问：注入函数的参数约定改了，图测试能发现吗？ 发现不了，图测试里的假函数是按旧约定写的；要靠服务层测试从真实入口跑一遍，才能发现注入的函数和图的调用方式对不上。

八、加强记忆

让状态图可测记「拆、换、数、分层」：拆，是把控制流和副作用拆开，图只决定什么时候调哪个动作；换，是动作注入（审查、重写、中断检查都从 State 里取函数）或回调注入（on_step 记录器，实现 __call__ 累计步号），测试时换成假函数或按脚本吐回复的假模型；数，是把有回路的图的路径逐条列出来（首轮过、改后过、到上限、开始前中断、中途中断、续跑），断言完整调用序列。分层，是图测试毫秒级覆盖走向，服务层用 mock 模型验证落库，端到端少量真跑专抓偶发格式问题。

项目实战落地

项目里怎么做的

《AI Agent智能会议纪要辅助系统》的纪要自检状态图不碰数据库，也不直接调模型：

四个动作由服务层注入：ReviewState 里除了纪要、问题清单、轮次这些数据字段，还有 review、refine、finalize、should_stop 四个函数字段；execute_agent_run 用闭包把运行记录、模型配置、审查材料包进去，图只按约定的参数调用；
图测试不连库不调模型：test_agent_graph.py 用假函数驱动首轮通过、改完复审、到上限收尾、最大轮数、中途中断、从已完成轮次续跑 6 条路径，比如两次审查依次返回不通过、通过，断言调用序列为 [("review", 1, 0), ("refine", 1, 0), ("review", 2, 1), ("finalize",)]，第三项是纪要的版本号，证明第二轮审查拿到的是重写后的那一版；
服务层单独测：test_agent.py 把 call_model 换成 AsyncMock，从 execute_agent_run 入口检查运行记录和步骤记录的落库结果。

《AI Agentic RAG高级企业知识库平台》用的是回调注入：State 里有一个 on_step，由服务层注入 StepRecorder（实现 __call__，跨节点累计步号）；图测试用假模型按脚本依次返回，一条脚本编排一条路径；「模型没检索就回答」这条用 monkeypatch 把检索节点换成什么都不做的函数，断言运行被判为失败。

为什么这样取舍

图不碰数据库、不调模型：纪要自检的 6 条路径测试因此不需要数据库，也不调模型；知识库平台的记录回调在测试里换成一个往列表里追加的函数就行。
服务层测试仍然要有：图只决定什么时候调哪个动作，轮次、判定、修订稿写没写进运行记录，要从服务入口检查。

面试官还会追问

纪要生成最早用 json_object 模式，为什么后来换成了严格的 JSON Schema？这个问题为什么单测很难发现？
结构约束为什么写在代码常量里，而不是放进管理员能改的 Prompt 模板？
审查节点每轮开头调用的 should_stop，为什么要重新读一次数据库里的运行状态？

学完《AI Agent智能会议纪要辅助系统》，上面这些追问你都会迎刃而解。

简化版

详细版