★ FEATURED ARTICLE 1. 从链式调用到状态图多智能体工作流为什么需要 LangGraph如果你用 LangChain 写过稍微复杂一点的东西大概率经历过这个阶段一开始用LLMChain串几个步骤跑得挺顺后来需求变成用户可能追问、可能要求重来、可能中途插入人工审核代码里开始出现while True、if retry_count 3、手动维护的history列表最后整个函数变成一坨谁都不敢动的面条。这不是你写得不优雅而是 LangChain 的抽象模型决定的。LangChain 的核心是链——数据从 A 流到 B 再到 C单向、固定、无环。它把 Prompt、Model、Parser、Retriever 这些组件封装得很好但它不负责流程控制。一旦流程里出现循环、条件跳转、状态回溯你就得自己在链外面套一层控制逻辑而这层逻辑 LangChain 不管。LangGraph 补的正是这一块。它把工作流建模成有向图节点是执行单元调 LLM、跑工具、做判断边是流转关系可以是条件边、循环边所有节点共享一个全局 State。这个 State 贯穿整个执行过程节点读它、改它、把它传给下一个节点。循环重试、条件分支、断点续跑、多 Agent 协作都是图模型天然能表达的东西。我试过把同一个客服问答 审核的任务分别用 LangChain 和 LangGraph 实现LangChain 版本大概 80 行其中一半在处理重试和状态LangGraph 版本 60 行状态管理全部交给框架逻辑清晰得多。这不是说 LangGraph 一定更好而是说当你的流程开始不线性的时候图模型的心智负担明显更低。这篇文章要解决的具体问题是如何用 TaoToken 统一 Key 和 API 通道把 LangChain 的链式调用平滑迁移到 LangGraph 的状态图并跑通一个多智能体工作流。适合已经用过 LangChain、想上手 LangGraph 但被状态检查点条件边这些概念卡住的开发者。下面从环境配置开始一步步给出可复制的代码和验证步骤。2. TaoToken 前置配置统一 Base URL 与依赖清单在写任何 LangGraph 代码之前先把模型接入这一层统一掉。多智能体工作流里经常要切换模型——检索 Agent 用便宜快的审核 Agent 用推理强的如果每个框架、每个 Agent 都单独配 Key管理成本会很高。TaoToken 的作用就是提供一个统一的 API 通道LangChain 和 LangGraph 都通过同一个 Base URL 和 Key 访问模型。先装依赖。LangGraph 依赖 LangChain 的核心组件所以两个都要装pip install langchain langchain-openai langgraph python-dotenv版本上langchain-core建议 1.5.x 以上langgraph用 2.0 系列。装完可以用pip show langgraph确认一下。接下来配置环境变量。在项目根目录建一个.env文件TAOTOKEN_API_KEY你的Key TAOTOKEN_BASE_URLhttps://taotoken.net/api注意 Base URL 这里不带任何路径后缀OpenAI 兼容接口会自动拼/v1/chat/completions。Key 的获取路径是登录后在控制台创建具体入口在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite 。然后在代码里统一读取。我习惯写一个config.py把模型客户端集中管理import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI load_dotenv() def get_model(model_id: str gpt-4o-mini, temperature: float 0.3) - ChatOpenAI: return ChatOpenAI( modelmodel_id, temperaturetemperature, api_keyos.getenv(TAOTOKEN_API_KEY), base_urlos.getenv(TAOTOKEN_BASE_URL), )这里的关键是base_url参数。LangChain 的ChatOpenAI底层走 OpenAI SDK只要把base_url指到 TaoToken 的 API 地址所有请求就会走统一通道。LangGraph 不直接调模型它调用的是 LangChain 的组件所以这一层配置对两个框架同时生效。如果你用的是 Claude 系列模型模型 ID 写claude-3-5-sonnet-20241022这类即可TaoToken 的通道会做协议适配。想先确认某个模型 ID 能不能用可以到模型对话页面手动发一条消息测试https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite 。依赖清单汇总一下方便你对照包名建议版本作用langchain1.x组件层Prompt/Model/Parserlangchain-openai最新OpenAI 兼容客户端langgraph2.0.x状态图编排引擎python-dotenv最新读取 .env配置完成后先跑一个最小验证确认通道是通的from config import get_model model get_model() resp model.invoke(用一句话说明什么是状态机) print(resp.content)能打印出内容说明 Base URL 和 Key 都对了。这一步别跳过后面 LangGraph 报错的时候你至少能确定不是接入层的问题。3. 可复制配置从 LangChain 链迁移到 LangGraph 状态图这一节是核心。我用同一个任务——用户提问 → 检索 → 生成回答 → 审核 → 不通过则重写——分别给出 LangChain 和 LangGraph 的实现你能直接看到差异在哪。先看 LangChain 的写法。链式调用下审核不通过要重写只能靠外层循环from config import get_model from langchain_core.prompts import ChatPromptTemplate model get_model() answer_prompt ChatPromptTemplate.from_template( 根据资料回答问题{context}\n问题{question} ) review_prompt ChatPromptTemplate.from_template( 审核以下回答是否准确只回复 PASS 或 FAIL\n{answer} ) def chain_workflow(question: str, context: str, max_retry: int 3) - str: for i in range(max_retry): messages answer_prompt.format_messages(contextcontext, questionquestion) answer model.invoke(messages).content review model.invoke( review_prompt.format_messages(answeranswer) ).content.strip() if PASS in review: return answer return answer # 重试耗尽返回最后一次这段代码能跑但问题很明显重试逻辑、状态当前是第几次、上一次的答案全在函数里手动管。如果再加一个检索 Agent和人工审核中断这个函数会迅速膨胀。换成 LangGraph同样的任务用状态图表达。先定义全局 Statefrom typing import TypedDict, Annotated import operator class WorkflowState(TypedDict): question: str context: str answer: str review: str retry_count: int history: Annotated[list, operator.add]Annotated[list, operator.add]表示这个字段用追加的方式合并多个节点往里写不会覆盖。这是 LangGraph 状态管理的一个细节新手容易在这里踩坑——不加operator.add后写的节点会覆盖前面的。然后定义节点函数。每个节点接收 State返回要更新的字段def retrieve_node(state: WorkflowState): # 实际项目里这里接向量库这里用占位 ctx f关于「{state[question]}」的检索结果 return {context: ctx, history: [retrieve]} def answer_node(state: WorkflowState): messages answer_prompt.format_messages( contextstate[context], questionstate[question] ) ans model.invoke(messages).content return {answer: ans, history: [answer]} def review_node(state: WorkflowState): messages review_prompt.format_messages(answerstate[answer]) result model.invoke(messages).content.strip() return {review: result, history: [review]} def rewrite_node(state: WorkflowState): return {retry_count: state[retry_count] 1, history: [rewrite]}接下来是图的核心——条件边。审核节点跑完后根据review决定是结束还是回到重写from langgraph.graph import StateGraph, START, END def route_after_review(state: WorkflowState) - str: if PASS in state[review]: return end if state[retry_count] 3: return end return rewrite graph StateGraph(WorkflowState) graph.add_node(retrieve, retrieve_node) graph.add_node(answer, answer_node) graph.add_node(review, review_node) graph.add_node(rewrite, rewrite_node) graph.add_edge(START, retrieve) graph.add_edge(retrieve, answer) graph.add_edge(answer, review) graph.add_conditional_edges( review, route_after_review, {rewrite: rewrite, end: END}, ) graph.add_edge(rewrite, answer) # 重写后回到 answer形成循环 app graph.compile()注意graph.add_edge(rewrite, answer)这一行——它让图形成了环。LangChain 的链做不到这一点而 LangGraph 天然支持。整个流程的重试不再是一段for循环而是图里的一条边。运行result app.invoke({ question: 如何申请退款, context: , answer: , review: , retry_count: 0, history: [], }) print(result[answer]) print(result[history])history会打印出节点执行顺序比如[retrieve, answer, review, rewrite, answer, review]你能清楚看到循环发生了几次。这个可观测性是链式调用给不了的。如果你要把这段配置持久化LangGraph 支持检查点。加一行from langgraph.checkpoint.memory import MemorySaver编译时传checkpointerMemorySaver()就能在中断后从上次状态恢复。生产环境换成 SQLite 或 Postgres 的 checkpointer 即可。4. 验证请求与成功结果多智能体协作跑通单 Agent 的图跑通后往上加多智能体。LangGraph 2.0 内置了 Supervisor 模式一个调度 Agent 决定把任务分给哪个子 Agent。这里我用两个子 Agent一个负责检索一个负责回答Supervisor 做路由。先定义子 Agent 的节点。为了演示清晰检索和回答各用一个模型调用def search_agent(state: WorkflowState): model get_model(gpt-4o-mini) ans model.invoke(f检索并总结{state[question]}).content return {context: ans, history: [search_agent]} def writer_agent(state: WorkflowState): model get_model(gpt-4o) ans model.invoke( f基于资料写回答{state[context]}\n问题{state[question]} ).content return {answer: ans, history: [writer_agent]}Supervisor 节点负责决策下一步走谁def supervisor_node(state: WorkflowState): model get_model(gpt-4o-mini) decision model.invoke( f当前已有资料{state[context][:100]}\n f当前回答{state[answer][:100]}\n 下一步应该 search 还是 write只回复一个词。 ).content.strip().lower() return {history: [fsupervisor-{decision}]} def route_supervisor(state: WorkflowState) - str: last state[history][-1] if search in last: return search return write组装图g StateGraph(WorkflowState) g.add_node(supervisor, supervisor_node) g.add_node(search, search_agent) g.add_node(write, writer_agent) g.add_node(review, review_node) g.add_edge(START, supervisor) g.add_conditional_edges( supervisor, route_supervisor, {search: search, write: write}, ) g.add_edge(search, supervisor) # 检索完回到调度 g.add_edge(write, review) g.add_conditional_edges( review, route_after_review, {rewrite: write, end: END}, ) multi_app g.compile()跑起来out multi_app.invoke({ question: LangGraph 和 LangChain 有什么区别, context: , answer: , review: , retry_count: 0, history: [], }) print(最终回答, out[answer]) print(执行轨迹, out[history])成功的话history会显示类似[supervisor-search, search_agent, supervisor-write, writer_agent, review]的轨迹。你能看到 Supervisor 先派检索拿到资料后再派写作最后进审核。整个过程的状态流转全部由框架管理你不需要手动传任何中间变量。这里有个实测下来的经验Supervisor 的决策提示词要写得足够窄只让它输出search或write两个词。如果提示词太开放模型可能返回一整句话route_supervisor里的字符串匹配就会失效导致路由错误。我一开始就踩过这个坑后来在提示词里加了只回复一个词才稳定。验证成功的标准有三个一是最终answer非空且内容合理二是history里能看到多个 Agent 的交替执行三是如果审核不通过能看到rewrite后重新进入write。三个都满足说明多智能体工作流跑通了。5. 本篇常见错误排查401、local proxy failed 与 OAuth 报错配置和运行过程中报错基本集中在这几类。我按实际遇到的频率排一下。401 Unauthorized / invalid api key最常见。原因通常是.env没被加载或者 Key 复制时带了空格。先确认load_dotenv()在读取环境变量之前执行再打印一下os.getenv(TAOTOKEN_API_KEY)[:8]看前几位对不对。如果 Key 本身没问题检查base_url是不是写成了https://taotoken.net/api/末尾多了斜杠有些版本会因此拼出双斜杠导致鉴权失败。正确写法是https://taotoken.net/api不带尾斜杠。local proxy failed / connection error这个报错说明请求根本没发出去卡在本地网络层。先确认你的运行环境能正常访问外网然后检查有没有全局的HTTP_PROXY环境变量在干扰。有些开发机配了系统级代理OpenAI SDK 会自动读取导致请求被劫持。可以在代码开头临时清掉import os os.environ.pop(HTTP_PROXY, None) os.environ.pop(HTTPS_PROXY, None)清完再跑如果通了说明就是代理环境变量的问题。reading choices / KeyError: choices这个报错通常不是网络问题而是返回体结构不对。可能是模型 ID 写错了通道返回了一个错误 JSON而 LangChain 还在按正常响应解析choices字段。解决办法是把原始响应打出来看import httpx, os r httpx.post( f{os.getenv(TAOTOKEN_BASE_URL)}/v1/chat/completions, headers{Authorization: fBearer {os.getenv(TAOTOKEN_API_KEY)}}, json{model: gpt-4o-mini, messages: [{role: user, content: hi}]}, timeout30, ) print(r.status_code, r.text[:500])看r.text里的error.message一般会直接告诉你模型不存在还是参数不对。OAuth / authentication 相关报错如果你在 Claude Code 或某些 CLI 工具里看到 OAuth 报错那是工具自己的登录态和 API Key 模式冲突了。这类工具要么走 OAuth 登录要么走 API Key不能混用。切到 API Key 模式时需要同时配好三件套Base URL、Key、Model ID。以 Claude Code 为例配置文件里要写全{ env: { ANTHROPIC_BASE_URL: https://taotoken.net/api, ANTHROPIC_API_KEY: 你的Key, ANTHROPIC_MODEL: claude-3-5-sonnet-20241022 } }三个字段缺一个都可能触发 OAuth 回退逻辑然后报鉴权失败。Cline 的 MCP 配置、Codex 的auth.json也是同理Base URL、Key、Model ID 三件套必须齐全。配置细节可以参考接入文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite 。图跑起来但 history 为空 / 状态没更新这不是接入问题是 LangGraph 的状态合并问题。检查你的 State 字段有没有加Annotated[list, operator.add]。没加的话节点返回的history会覆盖而不是追加看起来就像没更新。另外确认节点函数返回的是 dict且 key 名和 State 字段名完全一致拼写错了会被静默忽略。6. 把统一 Key 用在长期编码与 Agent 工作流上走到这里你应该已经能用 TaoToken 的统一 Key 同时驱动 LangChain 组件和 LangGraph 状态图了。回顾一下迁移路径LangChain 负责怎么调模型——Prompt 模板、模型客户端、输出解析LangGraph 负责怎么编排流程——状态、条件边、循环、多 Agent 调度。两者不是替代关系LangGraph 的节点内部照样调用 LangChain 的组件统一 Key 让这一层接入对两个框架透明。如果你打算把这个工作流长期跑下去比如做成一个常驻的编码助手或 Agent 服务建议关注 Coding Plan 这类长期方案比按量计费更适合高频调用场景https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite 。配置入口和控制台在 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite Key 管理在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite 。最后留一个实用技巧LangGraph 的 checkpointer 配合统一 Key可以做到服务重启后从上次中断的节点继续跑。在多 Agent 长流程里这个能力比想象中重要——一次任务可能跑十几分钟中途挂了不用从头再来。把MemorySaver换成持久化 checkpointerState 会自动落盘下次invoke时传入相同的thread_id就能恢复。这一步做完你的工作流才算真正具备生产可用的韧性。
阅读完成 · 觉得有帮助？
