构建完整工作流

前两篇我们学了"状态"和"节点/边"。本篇把它们整合，从零搭建一个含两个条件分支的完整工作流，并演示调试技巧。

一、目标：一个"问答 + 评分"工作流

我们要做一个把用户问题翻译成英文、再让模型回答、并对答案质量打分的工作流。流程如下：

flowchart TD
    START([START]) --> translate[翻译问题]
    translate --> answer[模型回答]
    answer --> grade[评分]
    grade -->|分数高| enough{分数>=0.8?}
    grade -->|分数低| rewrite[改写问题]
    rewrite --> translate
    enough -->|是| END([END])
    enough -->|否| rewrite

关键点：两条条件分支——一条决定"评分够不够去输出"，一条决定"要不要回头改写重试"。

二、第一步：定义状态

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

class QAState(TypedDict):
    user_input: str          # 原始中文问题
    en_question: str         # 翻译后英文问题
    answer: str              # 模型回答
    score: float             # 评分
    iteration_count: int     # 重试次数，防止死循环

注意 iteration_count 字段，重试循环必须配计数器，否则可能无限循环。详见 错误处理与重试。

三、第二步：写节点

为了让例子不依赖真实 API 也能跑，节点里用"假装"逻辑。你替换成真实 LLM 调用即可。

def translate(state: QAState) -> dict:
    # 实际项目：调用 ChatOpenAI 做翻译
    q = state["user_input"]
    en = f"[EN]{q}"  # 假装翻译成了英文
    return {"en_question": en}

def answer_node(state: QAState) -> dict:
    en = state["en_question"]
    ans = f"Answer to {en}"
    # 故意让前两次分低，演示重试
    n = state.get("iteration_count", 0)
    score = 0.5 if n < 1 else 0.9
    return {"answer": ans, "score": score}

def rewrite(state: QAState) -> dict:
    # 改写问题，并累加重试次数
    return {"en_question": state["en_question"] + " (refined)",
            "iteration_count": state.get("iteration_count", 0) + 1}

四、第三步：连边 + 条件分支

def route_after_grade(state: QAState) -> str:
    """评分后路由：够好就结束，不够就改写重试。"""
    if state["score"] >= 0.8:
        return "end"
    return "rewrite"

# 第二个条件分支：限制最多重试 3 次，超过也结束，避免死循环
def route_after_rewrite(state: QAState) -> str:
    if state.get("iteration_count", 0) >= 3:
        return "end"        # 兜底退出
    return "translate"      # 否则重新翻译

graph = StateGraph(QAState)
graph.add_node("translate", translate)
graph.add_node("answer", answer_node)
graph.add_node("rewrite", rewrite)

graph.add_edge(START, "translate")
graph.add_edge("translate", "answer")
# 条件分支 1：评分后决定去 END 还是 rewrite
graph.add_conditional_edges("answer", route_after_grade, ["end", "rewrite"])
# rewrite 后条件分支 2：决定重试还是退出
graph.add_conditional_edges("rewrite", route_after_rewrite, ["end", "translate"])

app = graph.compile()

注意条件边返回的 "end" 是个特殊字符串，会自动映射到 END。也可以直接返回 END 常量，效果一样。

五、第四步：运行并逐步打印状态

方式 1：看最终结果

result = app.invoke({"user_input": "什么是状态机"})
print(result["answer"], "score=", result["score"])

方式 2：逐步看每节点增量（推荐调试）

for chunk in app.stream({"user_input": "什么是状态机"}, stream_mode="updates"):
    for node, update in chunk.items():
        print(f"--- {node} ---")
        print(update)

输出大致：

--- translate ---
{'en_question': '[EN]什么是状态机'}
--- answer ---
{'answer': 'Answer to [EN]什么是状态机', 'score': 0.5}
--- rewrite ---
{'en_question': '[EN]什么是状态机 (refined)', 'iteration_count': 1}
--- translate ---
{'en_question': '[EN]什么是状态机 (refined)'}
--- answer ---
{'answer': 'Answer to ...', 'score': 0.9}

第二种 stream_mode="updates" 是排查工作流最直观的工具。

六、stream_mode 参数速览

取值	输出内容	适合场景
values	每步的完整状态快照	想看状态全貌
updates	每步的增量更新（默认推荐）	调试单节点行为
messages	LLM 逐 token 输出	做打字机效果
debug	最详细的执行日志	排查调度问题

可以同时订阅多个：

for chunk in app.stream(inputs, stream_mode=["updates", "values"]):
    print(chunk)

更多详见 流式输出 Streaming。

七、把"假装"换成真实 LLM

把 answer_node 换成真实调用即可。先安装：

pip install langchain-openai -i https://pypi.tuna.tsinghua.edu.cn/simple

Windows 用户：在 PowerShell 里直接用上面命令即可（-i 指定国内镜像源加速）。macOS/Linux 同命令。

import os
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def answer_node(state: QAState) -> dict:
    resp = llm.invoke(state["en_question"])
    return {"answer": resp.content, "score": 0.9}  # 评分也可让另一个 LLM 调用做

八、可视化图结构

把图导出成 mermaid 字符串，方便确认连边对不对：

print(app.get_graph().draw_mermaid())

把输出贴到任何 mermaid 渲染器就能看到拓扑。

九、常见踩坑汇总

踩坑 1：条件边返回的节点名拼错

route_after_grade 返回 "End"（大写），但 add_conditional_edges 第三参数写的是 "end"，匹配不上会报错。统一用小写，或直接用 END 常量。

踩坑 2：循环没有兜底出口

只写了"不够分就 rewrite"，没限制次数，模型一直给低分就死循环。任何带回头边的图都要配计数器出口。

踩坑 3：状态字段在重试中被意外累积

rewrite 同时改了 en_question 和 iteration_count，但 answer 字段没清空，可能残留上一轮答案。重试路径里要显式重置会被读取的中间字段。

踩坑 4：忘了 add_edge(START, ...)

每个图必须有一条 add_edge(START, 某节点) 作为入口，否则 compile 通过但 invoke 报"没有入口节点"。

踩坑 5：节点返回 None

节点不返回任何更新时应返回 {}，而不是 None。返回 None 会在合并时报错。

十、小结

• 完整工作流四步走：定义状态 → 写节点 → 连边（含条件分支）→ compile。
• 任何回路都要有终止条件（计数器/分数阈值），防止死循环。
• 调试三件套：stream_mode="updates" 看增量、get_graph().draw_mermaid() 看拓扑、print 中间状态。
• 真实项目把"假装"逻辑替换成 LLM/检索调用即可。

下一篇 多轮对话状态管理 讲如何让工作流"记住"对话历史。