使用 AI 编码助手?
- 安装 LangChain Docs MCP 服务器 让您的 Agent 访问最新的 LangChain 文档和示例。
- 安装 LangChain Skills 以提高 Agent 在 LangChain 生态系统任务上的性能。
- 如果您更喜欢将 Agent 定义为节点和边的图,请使用 Graph API。
- 如果您更喜欢将 Agent 定义为单个函数,请使用 Functional API。
对于此示例,您需要设置一个 Claude (Anthropic) 账户并获取 API 密钥。然后,在终端中设置
ANTHROPIC_API_KEY 环境变量。- 使用 Graph API
- 使用 Functional 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` 相乘。
Args:
a: 第一个整数
b: 第二个整数
"""
return a * b
@tool
def add(a: int, b: int) -> int:
"""将 `a` 和 `b` 相加。
Args:
a: 第一个整数
b: 第二个整数
"""
return a + b
@tool
def divide(a: int, b: int) -> float:
"""将 `a` 除以 `b`。
Args:
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. 定义状态
图的 state 用于存储消息和 LLM 调用次数。LangGraph 中的 state 在 Agent 执行过程中持续存在。带
operator.add 的 Annotated 类型确保新消息被追加到现有列表,而不是替换它。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="You are a helpful assistant tasked with performing arithmetic on a set of inputs."
)
]
+ 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. 构建并编译 Agent
Agent 使用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 = agent_builder.compile()
# 显示 Agent
from IPython.display import Image, display
display(Image(agent.get_graph(xray=True).draw_mermaid_png()))
# 调用
from langchain.messages import HumanMessage
messages = [HumanMessage(content="Add 3 and 4.")]
messages = agent.invoke({"messages": messages})
for m in messages["messages"]:
m.pretty_print()
要了解如何使用 LangSmith 跟踪您的 Agent,请参阅 LangSmith 文档。
完整代码示例
完整代码示例
# 步骤 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` 相乘。
Args:
a: 第一个整数
b: 第二个整数
"""
return a * b
@tool
def add(a: int, b: int) -> int:
"""将 `a` 和 `b` 相加。
Args:
a: 第一个整数
b: 第二个整数
"""
return a + b
@tool
def divide(a: int, b: int) -> float:
"""将 `a` 除以 `b`。
Args:
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: dict):
"""LLM 决定是否调用工具"""
return {
"messages": [
model_with_tools.invoke(
[
SystemMessage(
content="You are a helpful assistant tasked with performing arithmetic on a set of inputs."
)
]
+ 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:定义是否结束的决定逻辑
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
# 构建工作流
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 = agent_builder.compile()
from IPython.display import Image, display
# 显示 Agent
display(Image(agent.get_graph(xray=True).draw_mermaid_png()))
# 调用
from langchain.messages import HumanMessage
messages = [HumanMessage(content="Add 3 and 4.")]
messages = agent.invoke({"messages": messages})
for m in messages["messages"]:
m.pretty_print()
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` 相乘。
Args:
a: 第一个整数
b: 第二个整数
"""
return a * b
@tool
def add(a: int, b: int) -> int:
"""将 `a` 和 `b` 相加。
Args:
a: 第一个整数
b: 第二个整数
"""
return a + b
@tool
def divide(a: int, b: int) -> float:
"""将 `a` 除以 `b`。
Args:
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)
from langgraph.graph import add_messages
from langchain.messages import (
SystemMessage,
HumanMessage,
ToolCall,
)
from langchain_core.messages import BaseMessage
from langgraph.func import entrypoint, task
2. 定义模型节点
模型节点用于调用 LLM 并决定是否调用工具。@task 装饰器将函数标记为可作为 Agent 一部分执行的任务。任务可以在入口点函数中同步或异步调用。@task
def call_llm(messages: list[BaseMessage]):
"""LLM 决定是否调用工具"""
return model_with_tools.invoke(
[
SystemMessage(
content="You are a helpful assistant tasked with performing arithmetic on a set of inputs."
)
]
+ messages
)
3. 定义工具节点
工具节点用于调用工具并返回结果。@task
def call_tool(tool_call: ToolCall):
"""执行工具调用"""
tool = tools_by_name[tool_call["name"]]
return tool.invoke(tool_call)
4. 定义 Agent
Agent 使用@entrypoint 函数构建。在 Functional API 中,您不需要显式定义节点和边,而是可以在单个函数中编写标准控制流逻辑(循环、条件)。
@entrypoint()
def agent(messages: list[BaseMessage]):
model_response = call_llm(messages).result()
while True:
if not model_response.tool_calls:
break
# 执行工具
tool_result_futures = [
call_tool(tool_call) for tool_call in model_response.tool_calls
]
tool_results = [fut.result() for fut in tool_result_futures]
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="Add 3 and 4.")]
for chunk in agent.stream(messages, stream_mode="updates"):
print(chunk)
print("\n")
要了解如何使用 LangSmith 跟踪您的 Agent,请参阅 LangSmith 文档。
完整代码示例
完整代码示例
# 步骤 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` 相乘。
Args:
a: 第一个整数
b: 第二个整数
"""
return a * b
@tool
def add(a: int, b: int) -> int:
"""将 `a` 和 `b` 相加。
Args:
a: 第一个整数
b: 第二个整数
"""
return a + b
@tool
def divide(a: int, b: int) -> float:
"""将 `a` 除以 `b`。
Args:
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)
from langgraph.graph import add_messages
from langchain.messages import (
SystemMessage,
HumanMessage,
ToolCall,
)
from langchain_core.messages import BaseMessage
from langgraph.func import entrypoint, task
# 步骤 2:定义模型节点
@task
def call_llm(messages: list[BaseMessage]):
"""LLM 决定是否调用工具"""
return model_with_tools.invoke(
[
SystemMessage(
content="You are a helpful assistant tasked with performing arithmetic on a set of inputs."
)
]
+ messages
)
# 步骤 3:定义工具节点
@task
def call_tool(tool_call: ToolCall):
"""执行工具调用"""
tool = tools_by_name[tool_call["name"]]
return tool.invoke(tool_call)
# 步骤 4:定义 Agent
@entrypoint()
def agent(messages: list[BaseMessage]):
model_response = call_llm(messages).result()
while True:
if not model_response.tool_calls:
break
# 执行工具
tool_result_futures = [
call_tool(tool_call) for tool_call in model_response.tool_calls
]
tool_results = [fut.result() for fut in tool_result_futures]
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="Add 3 and 4.")]
for chunk in agent.stream(messages, stream_mode="updates"):
print(chunk)
print("\n")

