快速开始

本教程演示如何用 LangGraph 构建一个计算器 Agent，支持 Graph API 和 Functional API 两种方式。

准备工作

安装依赖并设置 API Key：

pip install -U langgraph langchain-anthropic
export ANTHROPIC_API_KEY=your-api-key

方式一：Graph API（图结构）

1. 定义工具和模型

from langchain.tools import tool
from langchain.chat_models import init_chat_model

model = init_chat_model("claude-sonnet-4-6", temperature=0)

@tool
def add(a: int, b: int) -> int:
    """加法"""
    return a + b

@tool
def multiply(a: int, b: int) -> int:
    """乘法"""
    return a * b

@tool
def divide(a: int, b: int) -> float:
    """除法"""
    return a / b

tools = [add, multiply, divide]
tools_by_name = {tool.name: tool for tool in tools}
model_with_tools = model.bind_tools(tools)

2. 定义 State

from langchain.messages import AnyMessage
from typing_extensions import TypedDict, Annotated
import operator

class MessagesState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    llm_calls: int

3. 定义节点

from langchain.messages import SystemMessage, ToolMessage

def llm_call(state: dict):
    """LLM 决定是否调用工具"""
    return {
        "messages": [
            model_with_tools.invoke(
                [SystemMessage(content="你是一个计算助手")] + state["messages"]
            )
        ],
        "llm_calls": state.get('llm_calls', 0) + 1
    }

def tool_node(state: dict):
    """执行工具调用"""
    result = []
    for tool_call in state["messages"][-1].tool_calls:
        tool = tools_by_name[tool_call["name"]]
        observation = tool.invoke(tool_call["args"])
        result.append(ToolMessage(content=observation, tool_call_id=tool_call["id"]))
    return {"messages": result}

4. 路由逻辑

from typing import Literal
from langgraph.graph import StateGraph, START, END

def should_continue(state: MessagesState) -> Literal["tool_node", END]:
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tool_node"
    return END

5. 构建和编译

agent_builder = StateGraph(MessagesState)
agent_builder.add_node("llm_call", llm_call)
agent_builder.add_node("tool_node", tool_node)
agent_builder.add_edge(START, "llm_call")
agent_builder.add_conditional_edges("llm_call", should_continue, ["tool_node", END])
agent_builder.add_edge("tool_node", "llm_call")
agent = agent_builder.compile()

# 执行
from langchain.messages import HumanMessage
messages = [HumanMessage(content="3 加 4 等于多少?")]
result = agent.invoke({"messages": messages})
for m in result["messages"]:
    m.pretty_print()

方式二：Functional API（函数式）

LangGraph 3.0+ 支持 Functional API，用 @entrypoint 和 @task 装饰器编写：

from langgraph.func import entrypoint, task
from langgraph.graph import add_messages
from langchain.messages import SystemMessage, HumanMessage, ToolCall
from langchain_core.messages import BaseMessage

@task
def call_llm(messages: list[BaseMessage]):
    return model_with_tools.invoke(
        [SystemMessage(content="你是一个计算助手")] + messages
    )

@task
def call_tool(tool_call: ToolCall):
    tool = tools_by_name[tool_call["name"]]
    return tool.invoke(tool_call)

@entrypoint()
def agent(messages: list[BaseMessage]):
    model_response = call_llm(messages).result()
    while True:
        if not model_response.tool_calls:
            break
        tool_results = [call_tool(tc).result() for tc in model_response.tool_calls]
        messages = add_messages(messages, [model_response, *tool_results])
        model_response = call_llm(messages).result()
    messages = add_messages(messages, model_response)
    return messages

# 执行
messages = [HumanMessage(content="3 加 4 等于多少?")]
for chunk in agent.stream(messages, stream_mode="updates"):
    print(chunk)

下一步

• 了解 Graph API 的 StateGraph 核心概念
• 了解 Functional API 的详细用法
• 添加 持久化 Checkpoint 支持对话记忆

参考

• 官方文档：Quickstart
• API 参考：StateGraph