TodayAI

用 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 完全一致。

官方资料

LangGraph DocsWorkflows and agents