原文链接:https://docs.langchain.com/oss/python/langgraph/quickstart#full-code-example

本快速入门演示了如何使用 LangGraph 的图 API 或函数式 API 构建一个计算器代理。

正在使用 AI 编程助手?
安装 LangChain Docs MCP 服务器,让您的代理能够访问最新的 LangChain 文档和示例。
安装 LangChain Skills,以提高您的代理在 LangChain 生态系统任务上的性能。

如果您倾向于将代理定义为由节点和边组成的图,请使用图 API。
如果您倾向于将代理定义为单个函数,请使用函数式 API。

有关概念信息,请参阅图 API 概述和函数式 API 概述。

对于此示例,您需要设置一个 Claude (Anthropic) 帐户并获取 API 密钥。然后,在终端中设置 ANTHROPIC_API_KEY 环境变量。有关所有可用提供商,请参阅聊天模型集成。如果您使用 LangSmith Gateway,您可以自带提供商密钥,或使用 Gateway Credits 访问模型而无需提供商密钥。

使用图 API
使用函数式 API

​1. 定义工具和模型

在此示例中,我们将使用 Claude Sonnet 4.5 模型,并定义用于加法、乘法和除法的工具。

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 multiply(a: int, b: int) -> int:
    """将 `a` 和 `b` 相乘。

    参数:
        a: 第一个整数
        b: 第二个整数
    """
    return a * b


@tool
def add(a: int, b: int) -> int:
    """将 `a` 和 `b` 相加。

    参数:
        a: 第一个整数
        b: 第二个整数
    """
    return a + b


@tool
def divide(a: int, b: int) -> float:
    """将 `a` 除以 `b`。

    参数:
        a: 第一个整数
        b: 第二个整数
    """
    return a / b


# 使用工具增强 LLM
tools = [add, multiply, divide]
tools_by_name = {tool.name: tool for tool in tools}
model_with_tools = model.bind_tools(tools)

​2. 定义状态

图的状态用于存储消息和 LLM 调用次数。
LangGraph 中的状态在代理执行期间持续存在。
使用 operator.addAnnotated 类型确保新消息被追加到现有列表中,而不是替换它。

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. 定义模型节点

模型节点用于调用 LLM 并决定是否调用工具。

from langchain.messages import SystemMessage


def llm_call(state: dict):
    """LLM 决定是否调用工具"""

    return {
        "messages": [
            model_with_tools.invoke(
                [
                    SystemMessage(
                        content="你是一个乐于助人的助手,任务是对一组输入执行算术运算。"
                    )
                ]
                + state["messages"]
            )
        ],
        "llm_calls": state.get('llm_calls', 0) + 1
    }

​4. 定义工具节点

工具节点用于调用工具并返回结果。

from langchain.messages import ToolMessage


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}

​5. 定义结束逻辑

条件边函数用于根据 LLM 是否进行了工具调用来路由到工具节点或结束。

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


def should_continue(state: MessagesState) -> Literal["tool_node", END]:
    """根据 LLM 是否进行了工具调用,决定是继续循环还是停止"""

    messages = state["messages"]
    last_message = messages[-1]

    # 如果 LLM 进行了工具调用,则执行操作
    if last_message.tool_calls:
        return "tool_node"

    # 否则,我们停止(回复用户)
    return END

​6. 构建并编译代理

使用 StateGraph 类构建代理,并使用 compile 方法进行编译。

# 构建工作流
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 IPython.display import Image, display
display(Image(agent.get_graph(xray=True).draw_mermaid_png()))

# 调用
from langchain.messages import HumanMessage
messages = [HumanMessage(content="将 3 和 4 相加。")]
messages = agent.invoke({"messages": messages})
for m in messages["messages"]:
    m.pretty_print()

使用 LangSmith 跟踪和调试您的代理。请遵循跟踪快速入门进行设置。为生产环境准备就绪后,请参阅部署以获取托管选项。
我们建议您也设置 LangSmith Engine,它会监控您的跟踪、检测问题并提出修复建议。

恭喜!您已经使用 LangGraph 图 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 multiply(a: int, b: int) -> int:
    """将 `a` 和 `b` 相乘。

    参数:
        a: 第一个整数
        b: 第二个整数
    """
    return a * b


@tool
def add(a: int, b: int) -> int:
    """将 `a` 和 `b` 相加。

    参数:
        a: 第一个整数
        b: 第二个整数
    """
    return a + b


@tool
def divide(a: int, b: int) -> float:
    """将 `a` 除以 `b`。

    参数:
        a: 第一个整数
        b: 第二个整数
    """
    return a / b


# 使用工具增强 LLM
tools = [add, multiply, divide]
tools_by_name = {tool.name: tool for tool in tools}
model_with_tools = model.bind_tools(tools)

# 步骤 2:定义状态

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


def llm_call(state: MessagesState):
    """LLM 决定是否调用工具"""

    return {
        "messages": [
            model_with_tools.invoke(
                [
                    SystemMessage(
                        content="你是一个乐于助人的助手,任务是对一组输入执行算术运算。"
                    )
                ]
                + state["messages"]
            )
        ],
        "llm_calls": state.get('llm_calls', 0) + 1
    }


# 步骤 4:定义工具节点

from langchain.messages import ToolMessage


def tool_node(state: MessagesState):
    """执行工具调用"""

    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}

# 步骤 5:定义确定是否结束的逻辑

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


# 条件边函数,根据 LLM 是否进行了工具调用来路由到工具节点或结束
def should_continue(state: MessagesState) -> Literal["tool_node", END]:
    """根据 LLM 是否进行了工具调用,决定是继续循环还是停止"""

    messages = state["messages"]
    last_message = messages[-1]

    # 如果 LLM 进行了工具调用,则执行操作
    if last_message.tool_calls:
        return "tool_node"

    # 否则,我们停止(回复用户)
    return END

# 步骤 6:构建代理

# 构建工作流
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 IPython.display import Image, display
# 显示代理
display(Image(agent.get_graph(xray=True).draw_mermaid_png()))

# 调用
from langchain.messages import HumanMessage
messages = [HumanMessage(content="将 3 和 4 相加。")]
messages = agent.invoke({"messages": messages})
for m in messages["messages"]:
    m.pretty_print()
Logo

汇聚全球AI编程工具,助力开发者即刻编程。

更多推荐