LangGraph 是一个用于构建有状态、多参与者应用的工作流库。创建基础图的思路可以概括为以下步骤：

1. 定义状态（State）

状态是工作流中流动的数据容器，通常用 TypedDict 或 Pydantic 模型定义：
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END

class State(TypedDict):
    messages: Annotated[list, "对话消息列表"]
    summary: str
    step: int

2. 创建状态图实例

workflow = StateGraph(State)  # 传入状态类型

3. 定义节点函数

节点是工作流的基本执行单元，每个节点：

• 接收当前状态作为输入
• 执行特定逻辑
• 返回状态更新（可以是部分更新）

def node1(state: State):
    print(f"节点1: 处理消息 {state['messages']}")
    return {"step": 1}  # 只更新step字段

def node2(state: State):
    return {"step": 2, "summary": "处理完成"}

4. 添加节点到图中

workflow.add_node("node_1", node1)
workflow.add_node("node_2", node2)

5. 设置入口点

workflow.set_entry_point("node_1")  # 指定从哪里开始

6. 连接节点（添加边）

定义节点间的流转关系：
# 简单线性流
workflow.add_edge("node_1", "node_2")
workflow.add_edge("node_2", END)  # 结束

# 或使用条件边
workflow.add_conditional_edges(
    "node_1",
    decide_next,  # 决策函数
    {"continue": "node_2", "end": END}
)

7. 编译和执行

# 编译图
app = workflow.compile()

# 执行
result = app.invoke({"messages": [], "summary": "", "step": 0})

完整示例

from typing import TypedDict
from langgraph.graph import StateGraph, END

# 1. 定义状态
class MyState(TypedDict):
    input: str
    processed: str
    step: int

# 2. 创建图
graph = StateGraph(MyState)

# 3. 定义节点
def input_node(state):
    print(f"输入: {state['input']}")
    return {"step": 1}

def process_node(state):
    processed = state['input'].upper()
    return {"processed": processed, "step": 2}

# 4. 添加节点
graph.add_node("input_node", input_node)
graph.add_node("process_node", process_node)

# 5. 设置入口
graph.set_entry_point("input_node")

# 6. 连接节点
graph.add_edge("input_node", "process_node")
graph.add_edge("process_node", END)

# 7. 编译执行
app = graph.compile()
result = app.invoke({"input": "hello", "processed": "", "step": 0})
print(result) # {'input': 'hello', 'processed': 'HELLO', 'step': 2}

示例执行结果

C:\Users\xiayu\miniconda3\envs\langchain03\python.exe C:\Users\xiayu\PyCharmMiscProject\AI-Agent-Dev-Practices-Code\第8章代码\test.py 
输入: hello
{'input': 'hello', 'processed': 'HELLO', 'step': 2}

Process finished with exit code 0
 

设计思路要点

状态驱动：所有数据都通过状态传递
函数式更新：节点返回状态的部分更新，系统会自动合并
声明式连接：用边明确声明节点间关系
可组合：小图可以组合成大图
可视化：可以导出为可视化图表

常见模式

# 线性流

A → B → C → END

# 分支流

     → B
A → 
     → C

# 循环流

A → B → (条件判断) → A 或 END

# 并行流

     → B
A →         → D
     → C

这种设计让复杂的工作流变得清晰、可维护，特别适合构建AI应用、数据处理流水线等场景。