企业官网建设流程全解析

1. 为什么线性 Chain 撑不住多步骤 AI 任务

如果你用 LangChain 的线性 Chain 写过稍微复杂一点的大模型应用，大概率踩过这几个坑：代码生成助手需要反复修改直到运行通过，智能客服要根据问题类型走不同处理流程，文档分析要同时提取摘要、关键词、核心观点。这些场景的共同点是——执行路径不是一条直线，而是带环、带岔路、带并发的图。

线性 Chain 的问题在于它假设「A 完了必然到 B，B 完了必然到 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_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ，API 端点是 https://taotoken.net/api 。

你需要先拿到一个 API Key。登录后进入控制台，在 API Keys 页面创建一个新 Key，复制出来存到环境变量里。这个 Key 就是后面所有节点调用模型的凭证。

# .env 文件 TAOTOKEN_API_KEY=sk-你的key TAOTOKEN_BASE_URL=https://taotoken.net/api

然后在 Python 里这样初始化模型客户端：

import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI load_dotenv() llm = ChatOpenAI( model="gpt-4o-mini", # 按你账号可用的模型名填 api_key=os.getenv("TAOTOKEN_API_KEY"), base_url=os.getenv("TAOTOKEN_BASE_URL"), temperature=0, )

这里有个容易忽略的点：base_url末尾不要带/v1，TaoToken 的兼容层会自动处理路径。如果你填成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: int

increment_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_code"

should_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 的额度方案。