给 LangGraph 企业审批 Agent 接 Qwen3 时最容易被卡住的不是StateGraph的节点编排而是 Step 1 的推理基座本地 vLLM 起 Qwen3-72B-Instruct 要 4×A100 或 8×L40S显存、驱动、推理框架、tool call 解析器都得自己拼。现在可以把这一步换成 TaoToken打开 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_content 创建 API Key再把 LangGraph 里 Qwen3 的 Base URL 填为 https://taotoken.net/api模型名先写 Qwen3。这样审批流跑到rule_check、execute_tool、human_review时Qwen3 的请求由统一 API 通道承担本地只留 MCP 工具层和状态检查点。原文那套结构仍然成立Qwen3 负责中文审批规则理解MCP 负责工具声明与调用LangGraph 负责把init→rule_check→execute_tool→human_review→audit串成可中断、可恢复、可审计的流程。区别只在推理基座的接入方式。自建 vLLM 是重资产路线模型权重、GPU 机器、并发压测、版本升级、安全加固都得自己管TaoToken 路线更像把 Qwen3 作为外部 API 通道接进 Agent先让审批 POC 跑通再决定哪些节点需要本地化。对于长会话、多工具、任务编排型 Agent第一步不是把机器堆满而是把状态机和工具边界先定清楚。下面按原文的目录节奏改写先处理 Step 1 的 Qwen3 接入再搭 LangGraph 图接着把 MCP 工具层和human_review中断接好最后补审计、排障和上线前检查。代码可以直接复制到 POC 工程但模型 ID、数据库连接、企业 OA 地址需要换成你自己的配置。1. 别急着买 4×A100LangGraph 审批 Agent 的 Qwen3 从 vLLM 切到 API1.1 原文 Step 1 的 vLLM 部署为什么卡住 POC原文 Step 1 做了两件事用 vLLM 暴露 Qwen3 的 OpenAI 兼容接口再用 MCP Proxy 把多个 MCP Server 聚合成单一端点。问题通常出在第一件事。Qwen3-72B-Instruct 的权重要显存4×A100 或 8×L40S 只是推荐量级实际还要看上下文长度、并发数、量化方案和 KV Cache 策略。 POC 阶段如果先把时间花在买卡、装驱动、配 CUDA、调--tool-call-parser审批业务本身反而没验证。更麻烦的是 Agent 场景不是单轮问答。LangGraph 会在一个审批实例里多次调用 Qwen3init之后可能先让模型抽取申请信息rule_check让模型判断自动审批还是升级人工execute_tool失败后可能让模型解释错误human_review恢复后还要做结果归纳。长会话意味着上下文持续增长多工具意味着每轮都可能带 MCP 工具 schema。自建 vLLM 时这些都会变成显存和调参问题而不是审批规则问题。MCP 适配层本身可以保留。原文用npx modelcontextprotocol/proxy把 OA、ERP、HR、文档解析几个 MCP Server 聚到一起这个思路在 API 路线下仍然有效。变化的是 Qwen3 不再由本地 vLLM 提供。你仍然需要 MCP Server 做工具声明、权限约束、参数校验和沙箱隔离但不需要为了模型推理先维持一套 GPU 集群。1.2 在 LangGraph 里把 Qwen3 的 base_url 指向 https://taotoken.net/api先处理密钥。打开 TaoToken 注册并创建 API KeyKey 用占位符YOUR_API_KEY表示。注意两个地址不要混注册、创建 Key、看模型广场、看用量走 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_content 填进 LangGraph 或 OpenAI 兼容 SDK 的 Base URL 写https://taotoken.net/api末尾不要带/v1。模型名可以先写Qwen3实际模型 ID 以模型广场当时列表为准。LangGraph 里接 Qwen3最省事的方式是用langchain-openai的ChatOpenAI因为它兼容 OpenAI 风格的base_url和api_key。不要自己拼 HTTP 请求也不要在代码里写死 Key。环境变量方式适合本地 POC显式参数方式适合把配置集中到settings.py。如果团队用.env保持下面三个字段一致即可。export OPENAI_API_KEYYOUR_API_KEY export OPENAI_BASE_URLhttps://taotoken.net/api export QWEN3_MODELQwen3# graph_qwen3.py from langchain_openai import ChatOpenAI llm ChatOpenAI( modelQwen3, # 实际模型 ID 以 TaoToken 模型广场为准 api_keyYOUR_API_KEY, # 从 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_content 创建 base_urlhttps://taotoken.net/api, temperature0, )这段代码解决的是原文 Step 1 中“部署 Qwen3”那一环。MCP Proxy 继续跑mcp-servers.yaml继续描述工具只是模型请求不再落到本地 8000 端口而是走 TaoToken 的统一 API 通道。这样做的直接好处是 POC 启动时间从“等机器、配环境、调推理参数”变成“创建 Key、填 Base URL、跑 StateGraph”。审批 Agent 的核心仍然是 LangGraph 状态机和 MCP 工具边界模型接入只是被替换成更轻的通道。2. init→rule_check让 Qwen3 先判断能不能自动审批2.1 ApprovalState 要用强类型字段别让 Graph 在运行时报错原文的ApprovalState用TypedDict定义包含request_id、applicant、amount、category、attachments、current_step、decision、human_feedback、audit_log。这套字段设计是对的但在生产环境里TypedDict不会做运行时校验。只要某个节点返回了不存在的字段或者amount从字符串变成浮点数后面节点就可能静默出错。审批 Agent 最怕的不是模型答错而是状态字段类型漂移导致audit记录不完整。POC 阶段可以继续用TypedDict快速起步上线前建议把关键状态换成Pydantic BaseModel或者在节点入口增加assert。特别是decision字段建议固定枚举值例如auto_approve、reject、escalate、waiting_human。不要让 Qwen3 自由输出中文“同意”“拒绝”“转人工”否则add_conditional_edges的分支判断很容易漏掉。模型输出 JSON 后先用解析函数做规范化再写回 State。from typing import TypedDict, Annotated, Literal import operator class ApprovalState(TypedDict): request_id: str applicant: str amount: float category: str attachments: list[str] current_step: str decision: Literal[auto_approve, reject, escalate, waiting_human, None] human_feedback: str | None audit_log: Annotated[list[str], operator.add]audit_log用Annotated[list[str], operator.add]是原文留下的好设计多个节点返回日志时会自动追加而不是覆盖。这个字段在后续排障时非常有用rule_check为什么判自动通过、execute_tool调了哪个 MCP 工具、human_review是谁在什么时候点了同意都能靠它串起来。注意它只能追加不能依赖它做业务判断业务判断仍然看decision和current_step。2.2 rule_check_node 调 Qwen3 时不要把规则写死在 prompt原文的rule_check_node先从 MCP 调oa_approval.get_auto_rules拿动态规则再让 Qwen3 根据申请人、金额、类别、附件数量输出 JSON。这个顺序在 API 路线下不变。不要把审批规则硬编码进 prompt否则每次业务调整都要改代码、重新部署 LangGraph 服务。规则从 MCP 工具取模型只负责在当前规则下做判断和解释这样 Qwen3 的职责更清晰。调用 Qwen3 时建议在 prompt 里明确要求“只输出 JSON”并在代码里做容错解析。模型可能返回 Markdown 代码块、前后解释文字或者尾随逗号。解析失败时不要把原始文本直接塞进decision而是返回escalate并追加审计日志让人工接管。审批 Agent 的底线是模型不可信输出不能直接触发资金动作。import json from datetime import datetime async def rule_check_node(state: ApprovalState) - dict: rules await mcp_client.call_tool(oa_approval.get_auto_rules, {}) prompt f你是企业审批规则引擎。根据以下规则和申请信息判断是否可自动审批。 【自动审批规则】 {rules[description]} 【申请信息】 申请人: {state[applicant]} 金额: {state[amount]}元 类别: {state[category]} 附件: {len(state[attachments])}个 请只输出 JSON: {{decision: auto_approve|escalate, reason: ...}} response await llm.ainvoke(prompt) text response.content.strip() try: if text.startswith(): text text.strip().replace(json\n, , 1).strip() result json.loads(text) decision result[decision] reason result.get(reason, ) except Exception: decision escalate reason Qwen3 输出解析失败转人工复核 return { decision: decision, current_step: rule_checked, audit_log: [f[{datetime.now()}] Rule check: {reason}], }这段节点能直接放进原文的 StateGraph。llm用上一节的ChatOpenAI实例mcp_client继续走 MCP Proxy。注意 Qwen3 只输出判断和理由不直接调用approve或reject。真正的工具调用在execute_tool节点并且只能调用 MCP Server 声明过的工具。3. execute_tool→human_reviewMCP 工具层和中断恢复怎么接3.1 mcp-servers.yaml 只暴露审批最必要的工具原文的 MCP 适配层用mcp-servers.yaml聚合多个 MCP Serveroa_approval暴露submit_approval、get_approval_status、approve、rejecterp_budget暴露check_budget、reserve_fund。这个粒度适合审批 Agent。不要为了“能力全”把 OA、ERP 的所有 API 都挂上去。工具越多模型选错工具的概率越高权限边界也越难审计。MCP Server 应运行在独立容器或独立进程里只暴露最小必要 API。Agent 通过 MCP 发起的只是工具调用请求真正执行由 MCP Server 内部调用企业系统接口完成企业系统仍然做鉴权和业务校验。不要把 MCP Server 写成能直连生产库、任意执行 SQL 或批量操作的通道。审批场景里approve这类动作必须绑定审批人身份、租户 ID 和申请单 ID不能只凭模型输出的参数就执行。servers: oa_approval: command: python args: [./mcp-oa/server.py] env: OA_API_KEY: ${OA_API_KEY} allowed_tools: - submit_approval - get_approval_status - approve - reject erp_budget: command: python args: [./mcp-erp/server.py] allowed_tools: - check_budget - reserve_fundexecute_tool_node里要做三件事校验decision是否允许自动执行、检查工具是否在allowed_tools、把tenant_id和request_id透传给 MCP。如果decision是escalate应该直接转human_review而不是让模型再尝试一次工具调用。原文在execute_tool_node里加入重试和超时升级这个逻辑值得保留MCP 超时重试 3 次后转人工权限不足直接抛异常并停止流程。async def execute_tool_node(state: ApprovalState) - dict: if state[decision] ! auto_approve: return {current_step: skipped_tool, audit_log: [非自动审批跳过工具执行]} tool_name oa_approval.approve params { request_id: state[request_id], tenant_id: state.get(tenant_id), comment: LangGraph 自动审批通过, } try: result await mcp_client.call_tool(tool_name, params) return { current_step: tool_executed, audit_log: [fMCP {tool_name} 返回: {result}], } except MCPTimeoutError: if state.get(retry_count, 0) 3: return {decision: escalate, audit_log: [MCP 超时 3 次转人工]} return {retry_count: state.get(retry_count, 0) 1} except MCPPermissionDenied: raise SecurityException(f越权工具调用: {tool_name})3.2 human_review 用 interrupt_before 暂停审批回调再 resume原文的human_review_node生成审批卡片推送到企业微信LangGraph 在interrupt_before[human_review]处暂停等外部回调 resume。这个设计是审批 Agent 能上线的关键。模型可以自动判断、自动调用低风险工具但涉及资金、合同、人员权限的动作必须留人工确认点。interrupt_before不是失败而是流程的一部分。编译图时要把 checkpointer 配上否则进程重启后审批进度会丢。原文用 PostgreSQL 做检查点生产环境也可以用 Redis 或企业已有的持久化层。thread_id必须和申请单绑定并且在 State 里强制写入tenant_id。否则 A 部门的审批实例可能被 B 部门看到。回调接口收到审批结果后用app.ainvoke带同一个thread_id恢复执行把human_feedback写回 State再进入audit节点。workflow StateGraph(ApprovalState) workflow.add_node(init, init_node) workflow.add_node(rule_check, rule_check_node) workflow.add_node(execute_tool, execute_tool_node) workflow.add_node(human_review, human_review_node) workflow.add_node(audit, audit_node) workflow.set_entry_point(init) workflow.add_edge(init, rule_check) workflow.add_conditional_edges( rule_check, lambda state: execute_tool if state[decision] auto_approve else human_review, {execute_tool: execute_tool, human_review: human_review}, ) workflow.add_edge(execute_tool, audit) workflow.add_edge(human_review, audit) workflow.add_edge(audit, END) app workflow.compile( checkpointerPostgresSaver.from_conn_string(postgresql://user:passlocalhost:5432/approval), interrupt_before[human_review], )回调接口可以像原文一样用 FastAPI 接收企业微信回调。要注意的是回调 payload 里的thread_id、decision、comment都要做校验不能让未认证请求直接 resume 审批流。审批人身份、租户 ID、申请单状态都要在恢复前再查一次。LangGraph 负责状态恢复企业系统负责业务鉴权两者不要混在一起。4. audit 节点与审计日志审批 Agent 上线前必须补的三件事4.1 全链路 Trace 和字段级脱敏原文在audit节点记录完整操作链每个节点输入输出、MCP 调用参数和结果、Qwen3 原始响应。这个做法对排障很有用但直接落库会把敏感信息一起写进去。审批数据里的身份证号、银行卡号、手机号、附件里的合同金额都需要在 State 进入 LLM 前做脱敏或者在audit写入前做字段级过滤。脱敏逻辑不要放在 prompt 里让模型做要放在代码里用规则处理。审计日志建议 append-only。audit_log字段在 State 里是追加语义落到数据库时也要禁止 UPDATE 和 DELETE。每次审批动作、每次 MCP 调用、每次人工确认都生成一条独立记录带上request_id、tenant_id、operator、timestamp、action、result。这样后续对账时能还原这条审批到底是 Qwen3 自动判断的还是人工在human_review节点点的。4.2 租户隔离和检查点恢复多租户场景下thread_id不能只用一个自增 ID。建议用tenant_id:request_id组合或者在 State 中强制注入tenant_idMCP 调用时透传。检查点恢复时先根据thread_id查出租户再校验当前操作人是否有权限 resume。原文提到“A 部门看到 B 部门审批”的问题根因就是thread_id没绑定租户。LangGraph 的 checkpointer 只负责保存状态不负责租户隔离这部分必须在业务层补。检查点写入频率也要注意。长会话审批流可能包含多次模型调用和工具调用如果每次节点结束都写完整 State数据库压力会很大。可以只持久化必要字段大附件存对象存储State 里只传引用。原文提到用 msgpack 替代 JSON、大附件存 OSS 只传引用这类优化在 POC 之后再做但字段设计要提前留好扩展位。5. 跑通后的验证与排障Qwen3 调用、MCP 超时、模型 ID5.1 先用一段最小 ChatOpenAI 代码测 Qwen3配完 LangGraph 不要直接跑完整审批流先用最小代码测 Qwen3 是否走通。同一个 Key、同一个 Base URL、同一个模型名发一条简单消息。如果这一步不通问题在 Key、Base URL 或模型 ID不在 StateGraph。测试通过后再把llm实例注入rule_check_node。# test_qwen3.py import asyncio from langchain_openai import ChatOpenAI test_llm ChatOpenAI( modelQwen3, api_keyYOUR_API_KEY, base_urlhttps://taotoken.net/api, temperature0, ) async def main(): resp await test_llm.ainvoke(只输出 JSON: {\ok\: true}) print(resp.content) asyncio.run(main())这一步也可以通过 TaoToken 模型对话 手动发一条消息验证。模型对话里能选到的 Qwen3 模型 ID直接复制到ChatOpenAI(model...)。不要把网上抄来的模型名、日期后缀当成正式配置模型广场当时列表里有什么就用什么。5.2 常见报错对照401、404、多了 /v1、MCP timeout401 通常看 Key。检查YOUR_API_KEY是否完整是否从 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_content 创建是否被空格或换行截断。不要把 Key 写在代码里提交到 Git环境变量和 secrets 管理工具更合适。401 还有一种情况是 SDK 发了不同的认证头确认ChatOpenAI使用的是Authorization: Bearer风格。404 先看 Base URL。填进工具的是https://taotoken.net/api末尾不要加/v1。有些库会自动补/v1这时要看最终请求路径不要盲目在前面再加一层。另一个 404 来源是模型 ID 写错比如把Qwen3写成不存在的日期版本。去模型广场复制准确 ID再跑最小测试代码。MCP timeout 则是工具层问题和 Qwen3 无关。检查 MCP Server 日志、企业 OA/ERP 接口响应时间、超时阈值。重试 3 次仍失败就转人工不要无限重试。human_review卡住不动先看interrupt_before是否配置在编译阶段再看 checkpointer 是否正常写入。没有 checkpointer进程重启后状态会丢thread_id写错回调 resume 会找不到原实例。审批回调里decision的值要和 State 里定义的枚举一致不要一边写approve一边写auto_approve。这类不一致不会报编译错误但会让条件边走到意外分支。6. 从审批 POC 到长会话多工具下一步看控制台和模型广场6.1 查看这次审批流的调用记录审批流跑通后回到 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_content 的控制台看用量和调用记录。确认每个审批实例是否按预期触发 Qwen3 调用rule_check调了几次、execute_tool失败后有没有额外模型调用、human_review恢复后有没有做结果归纳。用量异常通常意味着某个节点在循环里反复调模型或者add_conditional_edges的分支写错导致状态回退。模型广场也要定期看。Qwen3 的可用模型 ID、上下文长度、并发限制以当时列表为准。POC 时能跑通的模型名上线前要固定到配置中心不要散落在各节点代码里。团队多人开发时建议把model、base_url、api_key集中到环境变量或配置管理避免有人本地能跑、测试环境 404。6.2 把开发工具也接到同一把 Key写 LangGraph 代码、改 MCP Server、调 StateGraph 分支时如果你用 Claude Code 或 Codex 做辅助开发也可以把同一个 Base URL 配进去。注意区分Claude Code 用ANTHROPIC_BASE_URL时值仍然是https://taotoken.net/api不要带 UTMCodex 用~/.codex/config.toml时改成对应的model_provider和base_url。但这些工具只负责生成、解释、对照代码不会直接连你的生产 OA 或审批库。诊断 SQL、审批工具的实际执行仍然由读者在本地或企业系统里跑再把结果贴回对话。长会话、多工具场景下下一步重点不是继续加模型而是把状态机边界收紧哪些节点允许自动执行、哪些必须人工确认、哪些 MCP 工具只读、哪些工具需要二次鉴权。Qwen3 走 TaoToken 解决了推理基座的接入问题但审批 Agent 能不能上线取决于审计、租户隔离、超时升级和人工中断点有没有做扎实。配完这套审批流后先用 TaoToken 模型对话 发一条消息看 Qwen3 模型 ID 和 Base URL 是否对得上如果准备把 LangGraph 审批 Agent 长期跑在测试环境去 Coding Plan 看额度Key 在 控制台 API Keys 创建和轮换用 Claude Code 改 StateGraph 代码时环境变量对照看 Claude Code 接入文档。