用 LangGraph 定义可分支的工作流图
用 StateGraph 定义节点与边,跑通一个带条件分支的最小工作流。
基于 LangGraph Docs 整理 · LangGraph · 官方资料
不是所有任务都需要完全自主的 Agent。很多产品只要可控的工作流:分类 → 分支 → 汇总。
这篇按官方 workflows 思路定义节点与边,跑通一次 invoke。
定义状态、节点和边
用 TypedDict / dataclass 定义状态,用 add_node / add_edge / add_conditional_edges 连接。先让图跑通,再替换节点内部的模型调用。
结构示意
python
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
class State(TypedDict):
topic: str
route: str
answer: str
def classify(state: State) -> State:
route = "docs" if "文档" in state["topic"] else "chat"
return {**state, "route": route}
def answer_docs(state: State) -> State:
return {**state, "answer": "走文档分支"}
def answer_chat(state: State) -> State:
return {**state, "answer": "走闲聊分支"}
graph = StateGraph(State)
graph.add_node("classify", classify)
graph.add_node("docs", answer_docs)
graph.add_node("chat", answer_chat)
graph.add_edge(START, "classify")
graph.add_conditional_edges(
"classify",
lambda s: s["route"],
{"docs": "docs", "chat": "chat"},
)
graph.add_edge("docs", END)
graph.add_edge("chat", END)
app = graph.compile()
print(app.invoke({"topic": "这份文档讲什么", "route": "", "answer": ""}))最容易踩的坑
条件边走不到目标节点
确认路由函数返回值与 mapping 的 key 完全一致。