1. 为什么线性 Chain 撑不住多步骤 AI 任务如果你用 LangChain 的线性 Chain 写过稍微复杂一点的大模型应用大概率踩过这几个坑代码生成助手需要反复修改直到运行通过智能客服要根据问题类型走不同处理流程文档分析要同时提取摘要、关键词、核心观点。这些场景的共同点是——执行路径不是一条直线而是带环、带岔路、带并发的图。线性 Chain 的问题在于它假设「A 完了必然到 BB 完了必然到 C」。一旦你需要「校验失败就回到上一步重来」就只能在外面套 while 循环状态散落在各个变量里调试时根本不知道第几轮出了什么问题。一旦你需要「根据分类结果走不同分支」就是一堆 if/else 嵌套加一个分支就要动主流程。一旦你需要「四个维度同时提取」只能串行调用本来 4 秒能跑完的活硬生生拖到 16 秒。LangGraph 把工作流抽象成状态机全局共享一个 State节点是读写 State 的函数边决定下一步去哪。循环靠条件边回指上游节点分支靠条件边返回不同目标并行靠 Send API 分发多个节点同时跑再合并结果。三种模式组合起来就能覆盖绝大多数多步骤 AI 任务编排场景。这篇面向已经会写 LangChain 线性链、想往复杂工作流进阶的开发者。我会给出循环、分支、并行三类模式的完整配置骨架和节点路由示例每一步都配可运行的验证动作帮你确认循环是否真的终止、分支是否走对、并行是否真的汇合。文中模型调用统一走 TaoToken 的兼容接口你换成自己的 Key 也能直接跑。2. TaoToken 前置把模型调用这层先铺好LangGraph 本身只负责编排真正干活的是节点里的大模型调用。为了让后面的代码能直接复制运行先把模型接入这层配置好。TaoToken 提供 OpenAI 兼容的接口意味着你原来用ChatOpenAI写的节点函数几乎不用改只需要把base_url和api_key换掉。官网地址是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 端点是 https://taotoken.net/api 。你需要先拿到一个 API Key。登录后进入控制台在 API Keys 页面创建一个新 Key复制出来存到环境变量里。这个 Key 就是后面所有节点调用模型的凭证。# .env 文件 TAOTOKEN_API_KEYsk-你的key TAOTOKEN_BASE_URLhttps://taotoken.net/api然后在 Python 里这样初始化模型客户端import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI load_dotenv() llm ChatOpenAI( modelgpt-4o-mini, # 按你账号可用的模型名填 api_keyos.getenv(TAOTOKEN_API_KEY), base_urlos.getenv(TAOTOKEN_BASE_URL), temperature0, )这里有个容易忽略的点base_url末尾不要带/v1TaoToken 的兼容层会自动处理路径。如果你填成https://taotoken.net/api/v1部分 SDK 版本会拼出重复路径导致 404。实测下来直接用https://taotoken.net/api最稳。模型选型上循环里的代码生成节点建议用推理能力强的模型分支里的分类节点用便宜快速的小模型就够并行里的提取节点按维度复杂度分别选。这样能在保证效果的同时把成本压下来。3. 循环模式配置骨架与终止验证循环模式的核心是「条件边回指上游节点」。我们以代码生成助手为例用户给需求模型生成代码执行校验有错就带着错误信息回到生成节点重写直到通过或达到最大轮数。3.1 定义带循环计数的 State关键点在于循环次数要用Annotated加自定义 reducer否则每轮节点返回的loop_count会直接覆盖旧值永远停在 1循环就停不下来了。from typing import TypedDict, Annotated def increment_loop_count(old: int, new: int) - int: return old new class CodeGenState(TypedDict): requirement: str code: str error_msg: str loop_count: Annotated[int, increment_loop_count] max_loops: intincrement_loop_count的作用是节点返回{loop_count: 1}时LangGraph 不是覆盖而是把旧值和新值相加。这样每跑一轮生成节点计数就加一。3.2 节点函数与条件路由from langchain_core.prompts import ChatPromptTemplate def generate_code(state: CodeGenState) - dict: prompt ChatPromptTemplate.from_messages([ (system, 你是Python开发者根据需求生成可运行代码只输出代码。 如果提供了错误信息修复它。\n错误信息{error_msg}), (user, 需求{requirement}) ]) chain prompt | llm code chain.invoke({ requirement: state[requirement], error_msg: state.get(error_msg, ) }).content code code.replace(python, ).replace(, ).strip() return {code: code, loop_count: 1} def validate_code(state: CodeGenState) - dict: try: exec(state[code]) return {error_msg: } except Exception as e: return {error_msg: str(e)} def should_loop(state: CodeGenState) - str: if state[error_msg] or state[loop_count] state[max_loops]: return end return generate_codeshould_loop是循环的刹车片。两个终止条件必须同时存在业务上成功了或者轮数到顶了。只写前者模型输出不稳定时就会无限循环烧钱。3.3 构建图并验证终止from langgraph.graph import StateGraph, END builder StateGraph(CodeGenState) builder.add_node(generate_code, generate_code) builder.add_node(validate_code, validate_code) builder.set_entry_point(generate_code) builder.add_edge(generate_code, validate_code) builder.add_conditional_edges( validate_code, should_loop, {generate_code: generate_code, end: END} ) graph builder.compile()验证动作故意给一个会报错的需求观察loop_count是否递增到max_loops后停止。result graph.invoke({ requirement: 写一个函数故意调用一个不存在的库 foo_bar, loop_count: 0, max_loops: 3 }) print(循环次数, result[loop_count]) # 期望输出 3 print(最终错误, result[error_msg]) # 期望非空如果loop_count停在 1说明 reducer 没生效检查Annotated是否写对。如果循环次数超过max_loops说明条件函数里的判断用了而不是。4. 分支模式路由配置与走向验证分支模式的核心是「条件边返回不同目标节点」。以智能客服路由为例先分类问题再路由到对应处理节点。4.1 State 与分类节点class ServiceState(TypedDict): question: str question_type: str answer: str def classify_question(state: ServiceState) - dict: prompt ChatPromptTemplate.from_messages([ (system, 将问题分为四类product_use、bill、after_sale、other 只输出分类英文标识不要其他内容。), (user, 问题{question}) ]) qtype (prompt | llm).invoke({question: state[question]}).content.strip() return {question_type: qtype}4.2 路由函数与默认分支def route_question(state: ServiceState) - str: mapping { product_use: handle_product_use, bill: handle_bill, after_sale: handle_after_sale, } return mapping.get(state[question_type], handle_other)用dict.get加默认值比一长串 if/elif 更不容易漏。模型偶尔会输出没见过的分类词没有默认分支的话图会直接抛异常中断。4.3 构建图并验证走向builder StateGraph(ServiceState) builder.add_node(classify_question, classify_question) builder.add_node(handle_product_use, handle_product_use) builder.add_node(handle_bill, handle_bill) builder.add_node(handle_after_sale, handle_after_sale) builder.add_node(handle_other, handle_other) builder.set_entry_point(classify_question) builder.add_conditional_edges( classify_question, route_question, { handle_product_use: handle_product_use, handle_bill: handle_bill, handle_after_sale: handle_after_sale, handle_other: handle_other, } ) for node in [handle_product_use, handle_bill, handle_after_sale, handle_other]: builder.add_edge(node, END) graph builder.compile()验证动作用三类不同问题各跑一次检查question_type和最终answer是否匹配。for q in [怎么修改密码, 账单多扣了钱, 我要退货, 你们公司在哪里]: r graph.invoke({question: q}) print(q, -, r[question_type], -, r[answer][:20])如果所有问题都走到handle_other多半是分类节点返回的字符串带了多余空格或标点在classify_question里加.strip()和.lower()处理。5. 并行模式Send 分发与汇合验证并行模式的核心是「Send API 分发多个节点同时执行再合并结果」。以文档多维度分析为例同时提取摘要、关键词、核心观点。5.1 带合并规则的 State并行节点会同时写 State如果多个节点写同一个字段默认覆盖会丢数据。所以列表类字段要加追加型 reducer。from typing import List def merge_list(old: List, new: List) - List: return (old or []) (new or []) class DocState(TypedDict): content: str summary: str keywords: Annotated[List[str], merge_list] key_points: Annotated[List[str], merge_list]5.2 并行节点与 Send 分发from langgraph.constants import Send def extract_summary(state: DocState) - dict: prompt ChatPromptTemplate.from_messages([ (system, 提取摘要不超过100字只输出摘要。), (user, 文档{content}) ]) return {summary: (prompt | llm).invoke({content: state[content]}).content.strip()} def extract_keywords(state: DocState) - dict: import json prompt ChatPromptTemplate.from_messages([ (system, 提取3-5个关键词输出JSON数组不要其他内容。), (user, 文档{content}) ]) raw (prompt | llm).invoke({content: state[content]}).content.strip() return {keywords: json.loads(raw)} def parallel_dispatch(state: DocState) - List[Send]: return [ Send(extract_summary, state), Send(extract_keywords, state), Send(extract_key_points, state), ]parallel_dispatch返回一个 Send 列表LangGraph 会为每个 Send 启动一个目标节点它们共享同一份输入 State各自返回的增量按 reducer 合并。5.3 构建图并验证汇合builder StateGraph(DocState) builder.add_node(extract_summary, extract_summary) builder.add_node(extract_keywords, extract_keywords) builder.add_node(extract_key_points, extract_key_points) builder.set_entry_point(parallel_dispatch) builder.add_node(parallel_dispatch, parallel_dispatch) builder.add_edge(extract_summary, END) builder.add_edge(extract_keywords, END) builder.add_edge(extract_key_points, END) graph builder.compile()验证动作跑一次检查三个字段是否都有值且keywords和key_points没有被覆盖成空。doc LangGraph 是状态机编排框架支持循环、分支、并行。它被广泛用于多智能体场景。 r graph.invoke({content: doc}) print(摘要, r.get(summary)) print(关键词, r.get(keywords)) print(核心观点, r.get(key_points))如果某个字段是空的检查对应节点是否真的被 Send 触发了。可以在节点函数里加一行print(running extract_keywords)看输出顺序是否交错——交错说明真并行了顺序执行说明 Send 没生效。6. 本篇常见错排查循环停不下来九成是loop_count的 reducer 没配。检查 State 里是否写成Annotated[int, increment_loop_count]以及节点返回的是不是{loop_count: 1}而不是{loop_count: state[loop_count] 1}。后者配合 reducer 会翻倍增长。分支走到默认分支分类节点返回的字符串和路由字典的 key 对不上。模型可能返回Bill或bill 在路由函数里统一.strip().lower()再匹配。并行结果被覆盖多个节点写同一个字段且没加 reducer。列表用追加型 reducer字典用合并型 reducer标量字段尽量让不同节点写不同字段。Send 分发后图直接结束parallel_dispatch节点本身没有出边它的作用是返回 Send 列表不需要add_edge。但每个被 Send 的目标节点必须有出边指向 END 或下游节点否则图不知道汇合点在哪。状态序列化报错State 里存了函数、文件句柄、数据库连接这类不可序列化对象。把它们放到节点外部State 里只存 ID 或 URL。模型调用 401检查TAOTOKEN_API_KEY是否加载成功base_url是否写成https://taotoken.net/api。可以用一行代码单独测llm.invoke(hi)能返回就说明接入层没问题。7. 下一步把三种模式组合起来单模式跑通后真正的价值在组合。一个典型的智能论文分析助手可以这样编排PDF 解析节点 → 并行分发提取摘要/关键词/实验结果 → 合并 → 分支判断领域AI 领域走额外分析其他走通用→ 生成报告 → 循环评分直到达标或到顶。三种模式在同一个图里各司其职State 贯穿始终。调试组合图时建议先单独验证每个子图再拼起来。LangGraph 支持把编译好的子图作为节点嵌入父图这样循环、分支、并行的边界清晰出问题也好定位。如果你在接入模型这层想省点事可以直接用 TaoToken 的兼容端点把base_url指向 https://taotoken.net/api Key 在控制台创建。接入文档里有各语言 SDK 的配置示例遇到 404 或 401 先对照文档检查路径和鉴权头。模型对话入口可以用来快速验证 Key 是否可用长期跑编码类 Agent 任务的话可以看看 Coding Plan 的额度方案。