LangGraph vs LangChain：用TaoToken统一Key跑通多智能体工作流

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_URL=https://taotoken.net/api

注意 Base URL 这里不带任何路径后缀，OpenAI 兼容接口会自动拼/v1/chat/completions。Key 的获取路径是登录后在控制台创建，具体入口在 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite 。

然后在代码里统一读取。我习惯写一个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( model=model_id, temperature=temperature, api_key=os.getenv("TAOTOKEN_API_KEY"), base_url=os.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_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite 。

依赖清单汇总一下，方便你对照：

配置完成后，先跑一个最小验证，确认通道是通的：

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(context=context, question=question) answer = model.invoke(messages).content review = model.invoke( review_prompt.format_messages(answer=answer) ).content.strip() if "PASS" in review: return answer return answer # 重试耗尽，返回最后一次

这段代码能跑，但问题很明显：重试逻辑、状态（当前是第几次、上一次的答案）全在函数里手动管。如果再加一个"检索 Agent"和"人工审核中断"，这个函数会迅速膨胀。

换成 LangGraph，同样的任务用状态图表达。先定义全局 State：

from 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( context=state["context"], question=state["question"] ) ans = model.invoke(messages).content return {"answer": ans, "history": ["answer"]} def review_node(state: WorkflowState): messages = review_prompt.format_messages(answer=state["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，编译时传checkpointer=MemorySaver()，就能在中断后从上次状态恢复。生产环境换成 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": [f"supervisor->{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": f"Bearer {os.getenv('TAOTOKEN_API_KEY')}"}, json={"model": "gpt-4o-mini", "messages": [{"role": "user", "content": "hi"}]}, timeout=30, ) 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_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite 。

图跑起来但 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_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite 。配置入口和控制台在 https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite ，Key 管理在 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite 。

最后留一个实用技巧：LangGraph 的 checkpointer 配合统一 Key，可以做到"服务重启后从上次中断的节点继续跑"。在多 Agent 长流程里，这个能力比想象中重要——一次任务可能跑十几分钟，中途挂了不用从头再来。把MemorySaver换成持久化 checkpointer，State 会自动落盘，下次invoke时传入相同的thread_id就能恢复。这一步做完，你的工作流才算真正具备生产可用的韧性。