图 API
LangGraph 的核心是将 Agent 工作流建模为图（Graph）。你可以使用三个关键组件来定义 Agent 的行为：
- State（状态）：一个共享的数据结构，代表应用的当前快照。可以是一个
TypedDict或 Pydantic 模型。 - Nodes（节点）：编码 Agent 逻辑的函数。它们接收当前状态作为输入，执行计算或副作用，并返回更新后的状态。
- Edges（边）：根据当前状态确定下一步执行哪个节点的函数。可以是条件分支或固定转换。
通过组合 Nodes 和 Edges，你可以创建复杂的、循环的工作流，随时间演进状态。Nodes 和 Edges 本质上就是函数——它们可以包含 LLM，也可以是纯代码。
简而言之：节点负责执行工作，边负责告诉下一步做什么。
StateGraph
StateGraph 是主要的图类，由用户定义的 State 对象参数化。
定义状态
状态的模式可以是 TypedDict、dataclass 或 Pydantic BaseModel。每个键可以指定一个 reducer 函数，定义如何应用对该键的更新。如果未指定 reducer，则默认是覆盖（override）。
python
from typing import Annotated
from typing_extensions import TypedDict
from operator import add
from langgraph.graph.message import add_messages
from langchain.messages import AnyMessage
class AgentState(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]
next_agent: str
input_files: Annotated[list[str], add]添加节点
节点是普通的 Python 函数，接收当前状态并返回状态更新：
python
from langgraph.graph import StateGraph
def collect_input(state: AgentState) -> AgentState:
"""收集用户输入。"""
return {"input_files": state.get("input_files", [])}
def process_with_llm(state: AgentState) -> AgentState:
"""将消息发送给 LLM 处理。"""
# LLM 调用逻辑
return state添加边
边定义了图中节点之间的连接。有两种类型：
普通边：固定路径，从源节点到目标节点。
python
builder = StateGraph(AgentState)
builder.add_node("collect", collect_input)
builder.add_node("process", process_with_llm)
builder.add_edge("collect", "process")条件边：根据当前状态动态选择目标节点。
python
from langgraph.graph import START, END
def router(state: AgentState) -> str:
"""根据状态决定下一步去哪个节点。"""
if state.get("next_agent") == "researcher":
return "research_node"
elif state.get("next_agent") == "writer":
return "write_node"
return END
builder.add_conditional_edges(
"process",
router,
{
"research_node": "research",
"write_node": "write",
END: END,
},
)编译图
编译图会验证结构（没有孤立节点等），并允许指定运行时参数，如 checkpointers 和断点：
python
from langgraph.checkpoint.memory import InMemorySaver
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)必须先编译图，然后才能使用它。
完整示例
以下是一个完整的图 API 示例，实现了一个简单的 Agent 循环：
python
from typing import Annotated, Literal
from typing_extensions import TypedDict
from operator import add
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import InMemorySaver
from langchain.messages import AnyMessage
from langchain.agents import create_agent
class AgentState(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]
next_steps: Annotated[list[str], add]
def call_model(state: AgentState) -> dict:
"""调用 LLM 生成响应。"""
# 这里是模型调用逻辑的简化占位符
messages = state["messages"]
return {
"messages": [{"role": "assistant", "content": f"Processed: {messages[-1].content}"}],
}
def check_needs_tool(state: AgentState) -> Literal["tools", END]:
"""检查是否需要调用工具。"""
last_message = state["messages"][-1]
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "tools"
return END
def run_tools(state: AgentState) -> dict:
"""执行工具调用。"""
return {"next_steps": ["tool_executed"]}
builder = StateGraph(AgentState)
# 添加节点
builder.add_node("call_model", call_model)
builder.add_node("tools", run_tools)
# 添加边
builder.add_edge(START, "call_model")
builder.add_conditional_edges(
"call_model",
check_needs_tool,
{"tools": "tools", END: END},
)
builder.add_edge("tools", "call_model")
# 编译
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
# 使用
result = graph.invoke(
{"messages": [{"role": "user", "content": "Hello!"}]},
config={"configurable": {"thread_id": "1"}},
)Reducer 详解
Reducer 是理解节点更新如何应用到状态的关键。状态中的每个键都有自己的 reducer 函数。
默认 reducer（覆盖）：
python
class State(TypedDict):
foo: int
bar: list[str]
# 初始: {"foo": 1, "bar": ["hi"]}
# 节点返回: {"foo": 2}
# 结果: {"foo": 2, "bar": ["hi"]}
# 节点返回: {"bar": ["bye"]}
# 结果: {"foo": 2, "bar": ["bye"]}使用 operator.add reducer：
python
from typing import Annotated
from operator import add
class State(TypedDict):
foo: int
bar: Annotated[list[str], add]
# 初始: {"foo": 1, "bar": ["hi"]}
# 节点返回: {"foo": 2}
# 结果: {"foo": 2, "bar": ["hi"]}
# 节点返回: {"bar": ["bye"]}
# 结果: {"foo": 2, "bar": ["hi", "bye"]}使用 create_agent 构建图
顶层 create_agent 工厂函数内部也是基于图 API 构建的：
python
from langchain.agents import create_agent
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
@tool
def search(query: str) -> str:
"""搜索信息。"""
return f"Results for: {query}"
agent = create_agent(
model="claude-sonnet-4-6",
tools=[search],
system_prompt="You are a helpful assistant.",
checkpointer=InMemorySaver(),
)
# agent 本身就是一个已编译的图
result = agent.invoke(
{"messages": [{"role": "user", "content": "Search for AI news"}]},
config={"configurable": {"thread_id": str(uuid7())}},
)