智能体¶
类
名称 | 描述 |
---|---|
AgentState |
智能体的状态。 |
函数
名称 | 描述 |
---|---|
create_react_agent |
创建一个智能体图,该图会循环调用工具,直到满足停止条件。 |
AgentState ¶
基类:TypedDict
智能体的状态。
create_react_agent ¶
create_react_agent(
model: Union[
str,
LanguageModelLike,
Callable[
[StateSchema, Runtime[ContextT]], BaseChatModel
],
Callable[
[StateSchema, Runtime[ContextT]],
Awaitable[BaseChatModel],
],
Callable[
[StateSchema, Runtime[ContextT]],
Runnable[LanguageModelInput, BaseMessage],
],
Callable[
[StateSchema, Runtime[ContextT]],
Awaitable[
Runnable[LanguageModelInput, BaseMessage]
],
],
],
tools: Union[
Sequence[Union[BaseTool, Callable, dict[str, Any]]],
ToolNode,
],
*,
prompt: Optional[Prompt] = None,
response_format: Optional[
Union[
StructuredResponseSchema,
tuple[str, StructuredResponseSchema],
]
] = None,
pre_model_hook: Optional[RunnableLike] = None,
post_model_hook: Optional[RunnableLike] = None,
state_schema: Optional[StateSchemaType] = None,
context_schema: Optional[Type[Any]] = None,
checkpointer: Optional[Checkpointer] = None,
store: Optional[BaseStore] = None,
interrupt_before: Optional[list[str]] = None,
interrupt_after: Optional[list[str]] = None,
debug: bool = False,
version: Literal["v1", "v2"] = "v2",
name: Optional[str] = None,
**deprecated_kwargs: Any
) -> CompiledStateGraph
创建一个智能体图,该图会循环调用工具,直到满足停止条件。
有关使用 create_react_agent
的更多详情,请访问智能体文档。
参数
名称 | 类型 | 描述 | 默认值 |
---|---|---|---|
model
|
Union[str, LanguageModelLike, Callable[[StateSchema, Runtime[ContextT]], BaseChatModel], Callable[[StateSchema, Runtime[ContextT]], Awaitable[BaseChatModel]], Callable[[StateSchema, Runtime[ContextT]], Runnable[LanguageModelInput, BaseMessage]], Callable[[StateSchema, Runtime[ContextT]], Awaitable[Runnable[LanguageModelInput, BaseMessage]]]]
|
智能体的语言模型。支持静态和动态模型选择。
动态函数接收图状态和运行时,从而实现依赖于上下文的模型选择。必须返回一个 动态模型示例
动态模型要求 确保返回的模型通过 |
必填 |
工具
|
Union[Sequence[Union[BaseTool, Callable, dict[str, Any]]], ToolNode]
|
工具列表或 ToolNode 实例。如果提供了一个空列表,智能体将只包含一个没有工具调用的 LLM 节点。 |
必填 |
prompt
|
Optional[Prompt]
|
一个可选的 LLM 提示。可以采用几种不同的形式
|
None
|
response_format
|
Optional[Union[StructuredResponseSchema, tuple[str, StructuredResponseSchema]]]
|
一个可选的最终智能体输出模式。 如果提供,输出将被格式化以匹配给定的模式,并返回在 'structured_response' 状态键中。如果未提供,
重要
注意 图将在智能体循环结束后,单独调用 LLM 来生成结构化响应。这并非获取结构化响应的唯一策略,更多选项请参阅本指南。 |
None
|
pre_model_hook
|
Optional[RunnableLike]
|
一个可选的节点,用于在
重要 必须提供 |
None
|
post_model_hook
|
Optional[RunnableLike]
|
一个可选的节点,用于在 注意 仅在 |
None
|
state_schema
|
Optional[StateSchemaType]
|
一个可选的状态模式,用于定义图状态。必须包含 |
None
|
context_schema
|
Optional[Type[Any]]
|
运行时上下文的可选模式。 |
None
|
checkpointer
|
Optional[Checkpointer]
|
一个可选的检查点保存器对象。这用于为单个线程(例如,单个对话)持久化图的状态(例如,作为聊天内存)。 |
None
|
store
|
Optional[BaseStore]
|
一个可选的存储对象。这用于在多个线程(例如,多个对话/用户)之间持久化数据。 |
None
|
interrupt_before
|
Optional[list[str]]
|
一个可选的节点名称列表,在这些节点之前中断。应为以下之一:"agent", "tools"。如果您想在执行操作之前添加用户确认或其他中断,这将非常有用。 |
None
|
interrupt_after
|
Optional[list[str]]
|
一个可选的节点名称列表,在这些节点之后中断。应为以下之一:"agent", "tools"。如果您想直接返回或对输出运行额外的处理,这将非常有用。 |
None
|
debug
|
bool
|
一个指示是否启用调试模式的标志。 |
False
|
version
|
Literal['v1', 'v2']
|
确定要创建的图的版本。可以是以下之一
|
'v2'
|
name
|
Optional[str]
|
CompiledStateGraph 的可选名称。当将 ReAct 智能体图作为子图节点添加到另一个图中时,此名称将自动使用——这对于构建多智能体系统特别有用。 |
None
|
config_schema
已弃用
config_schema
参数在 v0.6.0 中已弃用,并将在 v2.0.0 中移除支持。请改用 context_schema
来指定运行范围上下文的模式。
返回
类型 | 描述 |
---|---|
CompiledStateGraph
|
一个已编译的 LangChain 可运行对象,可用于聊天交互。 |
“agent”节点使用消息列表(应用提示后)调用语言模型。如果生成的 AIMessage 包含 tool_calls
,图将接着调用 “tools”。“tools”节点执行工具(每个 tool_call
一个工具),并将响应作为 ToolMessage
对象添加到消息列表中。然后,agent 节点再次调用语言模型。该过程重复进行,直到响应中不再出现 tool_calls
。然后,agent 将返回完整的消息列表,形式为一个包含键 "messages" 的字典。
sequenceDiagram
participant U as User
participant A as LLM
participant T as Tools
U->>A: Initial input
Note over A: Prompt + LLM
loop while tool_calls present
A->>T: Execute tools
T-->>A: ToolMessage for each tool_calls
end
A->>U: Return final state
示例
from langgraph.prebuilt import create_react_agent
def check_weather(location: str) -> str:
'''Return the weather forecast for the specified location.'''
return f"It's always sunny in {location}"
graph = create_react_agent(
"anthropic:claude-3-7-sonnet-latest",
tools=[check_weather],
prompt="You are a helpful assistant",
)
inputs = {"messages": [{"role": "user", "content": "what is the weather in sf"}]}
for chunk in graph.stream(inputs, stream_mode="updates"):
print(chunk)
ToolNode ¶
基类:RunnableCallable
一个运行上一条 AIMessage 中调用的工具的节点。
它可以在 StateGraph 中使用,带有一个 "messages" 状态键(或通过 ToolNode 的 'messages_key' 传递的自定义键)。如果请求了多个工具调用,它们将并行运行。输出将是一个 ToolMessages 列表,每个工具调用对应一个。
工具调用也可以直接作为 ToolCall
字典的列表传递。
参数
名称 | 类型 | 描述 | 默认值 |
---|---|---|---|
工具
|
Sequence[Union[BaseTool, Callable]]
|
此节点可以调用的一系列工具。工具可以是 BaseTool 实例或将被转换为工具的普通函数。 |
必填 |
name
|
str
|
图中此节点的名称标识符。用于调试和可视化。默认为 "tools"。 |
'tools'
|
tags
|
Optional[list[str]]
|
与节点关联的可选元数据标签,用于过滤和组织。默认为 None。 |
None
|
handle_tool_errors
|
Union[bool, str, Callable[..., str], tuple[type[Exception], ...]]
|
工具执行期间错误处理的配置。默认为 True。支持多种策略
|
True
|
messages_key
|
str
|
状态字典中包含消息列表的键。此键也将用于输出的 ToolMessages。默认为 "messages"。 |
'messages'
|
示例
使用简单工具的基本用法
from langgraph.prebuilt import ToolNode
from langchain_core.tools import tool
@tool
def calculator(a: int, b: int) -> int:
"""Add two numbers."""
return a + b
tool_node = ToolNode([calculator])
自定义错误处理
def handle_math_errors(e: ZeroDivisionError) -> str:
return "Cannot divide by zero!"
tool_node = ToolNode([calculator], handle_tool_errors=handle_math_errors)
直接执行工具调用
注意
ToolNode 期望输入为以下三种格式之一: 1. 包含一个 messages 键的字典,该键的值为消息列表 2. 直接的消息列表 3. 工具调用字典的列表
使用消息格式时,最后一条消息必须是填充了 tool_calls 的 AIMessage。节点会自动提取并并发处理这些工具调用。
对于涉及状态注入或存储访问的高级用例,可以使用 InjectedState 或 InjectedStore 注解工具,以自动接收图上下文。
方法
名称 | 描述 |
---|---|
inject_tool_args |
将图状态和存储注入到工具调用参数中。 |
inject_tool_args ¶
inject_tool_args(
tool_call: ToolCall,
input: Union[
list[AnyMessage], dict[str, Any], BaseModel
],
store: Optional[BaseStore],
) -> ToolCall
将图状态和存储注入到工具调用参数中。
此方法使工具能够访问不应由模型控制的图上下文。工具可以使用 InjectedState 和 InjectedStore 注解来声明对图状态或持久存储的依赖。此方法会自动识别这些依赖项并注入适当的值。
注入过程在保留原始工具调用结构的同时,添加了必要的上下文参数。这使得工具既能被模型调用,又能感知上下文,而无需向模型暴露内部状态管理。
参数
名称 | 类型 | 描述 | 默认值 |
---|---|---|---|
tool_call
|
ToolCall
|
要用注入的参数来增强的工具调用字典。必须包含 'name', 'args', 'id' 和 'type' 字段。 |
必填 |
input
|
Union[list[AnyMessage], dict[str, Any], BaseModel]
|
要注入到需要状态访问的工具中的当前图状态。可以是一个消息列表、状态字典或 BaseModel 实例。 |
必填 |
store
|
Optional[BaseStore]
|
要注入到需要存储的工具中的持久存储实例。如果图没有配置存储,则为 None。 |
必填 |
返回
类型 | 描述 |
---|---|
ToolCall
|
一个新的 ToolCall 字典,其结构与输入相同,但根据工具的注解要求注入了额外的参数。 |
ToolCall
|
其他参数已注入。 |
抛出
类型 | 描述 |
---|---|
ValueError
|
如果工具需要存储注入但未提供存储,或者无法满足状态注入要求。 |
注意
此方法在工具执行期间自动调用,但也可以在使用 Send API 或自定义路由逻辑时手动使用。注入是在工具调用的副本上执行的,以避免改变原始调用。
LangGraph 工作流的工具执行节点。
该模块为在 LangGraph 中执行工具提供了预构建的功能。
工具是模型可以调用的函数,用于与外部系统、API、数据库交互或执行计算。
该模块实现了几个关键的设计模式: - 并行执行多个工具调用以提高效率 - 具有可定制错误消息的健壮错误处理 - 为需要访问图状态的工具提供状态注入 - 为需要持久存储的工具提供存储注入 - 用于高级控制流的基于命令的状态更新
关键组件
ToolNode:在 LangGraph 工作流中执行工具的主类 InjectedState:用于将图状态注入工具的注解 InjectedStore:用于将持久存储注入工具的注解 tools_condition:基于工具调用进行条件路由的实用函数
典型用法
类
名称 | 描述 |
---|---|
InjectedState |
用于将图状态注入工具参数的注解。 |
InjectedStore |
用于将持久化存储注入工具参数的注解。 |
函数
名称 | 描述 |
---|---|
tools_condition |
用于工具调用工作流的条件路由函数。 |
InjectedState ¶
用于将图状态注入工具参数的注解。
此注解使工具能够访问图状态,而无需向语言模型暴露状态管理细节。使用 InjectedState 注解的工具在执行期间自动接收状态数据,同时对模型的工具调用接口保持不可见。
参数
名称 | 类型 | 描述 | 默认值 |
---|---|---|---|
field
|
Optional[str]
|
从状态字典中提取的可选键。如果为 None,则注入整个状态。如果指定,则仅注入该字段的值。这允许工具请求特定的状态组件,而不是处理整个状态结构。 |
None
|
示例
from typing import List
from typing_extensions import Annotated, TypedDict
from langchain_core.messages import BaseMessage, AIMessage
from langchain_core.tools import tool
from langgraph.prebuilt import InjectedState, ToolNode
class AgentState(TypedDict):
messages: List[BaseMessage]
foo: str
@tool
def state_tool(x: int, state: Annotated[dict, InjectedState]) -> str:
'''Do something with state.'''
if len(state["messages"]) > 2:
return state["foo"] + str(x)
else:
return "not enough messages"
@tool
def foo_tool(x: int, foo: Annotated[str, InjectedState("foo")]) -> str:
'''Do something else with state.'''
return foo + str(x + 1)
node = ToolNode([state_tool, foo_tool])
tool_call1 = {"name": "state_tool", "args": {"x": 1}, "id": "1", "type": "tool_call"}
tool_call2 = {"name": "foo_tool", "args": {"x": 1}, "id": "2", "type": "tool_call"}
state = {
"messages": [AIMessage("", tool_calls=[tool_call1, tool_call2])],
"foo": "bar",
}
node.invoke(state)
注意
- InjectedState 参数会自动从呈现给语言模型的工具模式中排除
- ToolNode 在执行期间处理注入过程
- 工具可以混合使用常规参数(由模型控制)和注入参数(由系统控制)
- 状态注入发生在模型生成工具调用之后,但在工具执行之前
InjectedStore ¶
用于将持久化存储注入工具参数的注解。
此注解使工具能够访问 LangGraph 的持久存储系统,而无需向语言模型暴露存储细节。使用 InjectedStore 注解的工具在执行期间自动接收存储实例,同时对模型的工具调用接口保持不可见。
该存储提供持久的、跨会话的数据存储,工具可以使用它来维护上下文、用户偏好或任何其他需要超越单个工作流执行而持久存在的数据。
警告
InjectedStore
注解需要 langchain-core >= 0.3.8
示例
from typing_extensions import Annotated
from langchain_core.tools import tool
from langgraph.store.memory import InMemoryStore
from langgraph.prebuilt import InjectedStore, ToolNode
@tool
def save_preference(
key: str,
value: str,
store: Annotated[Any, InjectedStore()]
) -> str:
"""Save user preference to persistent storage."""
store.put(("preferences",), key, value)
return f"Saved {key} = {value}"
@tool
def get_preference(
key: str,
store: Annotated[Any, InjectedStore()]
) -> str:
"""Retrieve user preference from persistent storage."""
result = store.get(("preferences",), key)
return result.value if result else "Not found"
与 ToolNode 和图编译一起使用
from langgraph.graph import StateGraph
from langgraph.store.memory import InMemoryStore
store = InMemoryStore()
tool_node = ToolNode([save_preference, get_preference])
graph = StateGraph(State)
graph.add_node("tools", tool_node)
compiled_graph = graph.compile(store=store) # Store is injected automatically
跨会话持久性
注意
- InjectedStore 参数会自动从呈现给语言模型的工具模式中排除
- 存储实例由 ToolNode 在执行期间自动注入
- 工具可以使用存储的 get/put 方法访问命名空间存储
- 存储注入要求图在编译时带有一个存储实例
- 多个工具可以共享同一个存储实例以实现数据一致性
tools_condition ¶
tools_condition(
state: Union[
list[AnyMessage], dict[str, Any], BaseModel
],
messages_key: str = "messages",
) -> Literal["tools", "__end__"]
用于工具调用工作流的条件路由函数。
此实用函数实现了 ReAct 风格智能体的标准条件逻辑:如果最后一条 AI 消息包含工具调用,则路由到工具执行节点;否则,结束工作流。这种模式是大多数工具调用智能体架构的基础。
该函数处理 LangGraph 应用程序中常用的多种状态格式,使其能够灵活适应不同的图设计,同时保持一致的行为。
参数
名称 | 类型 | 描述 | 默认值 |
---|---|---|---|
state
|
Union[list[AnyMessage], dict[str, Any], BaseModel]
|
用于检查工具调用的当前图状态。支持的格式: - 消息列表(用于 MessageGraph) - 包含 messages 键的字典(用于 StateGraph) - 带有 messages 属性的 BaseModel 实例 |
必填 |
messages_key
|
str
|
状态中包含消息列表的键或属性名称。这允许对使用不同状态模式的图进行自定义。默认为 "messages"。 |
'messages'
|
返回
类型 | 描述 |
---|---|
Literal['tools', '__end__']
|
如果最后一条 AI 消息中存在工具调用,则为 "tools",否则为 "end" |
Literal['tools', '__end__']
|
以终止工作流。这些是 |
Literal['tools', '__end__']
|
工具调用条件边的标准路由目标。 |
抛出
类型 | 描述 |
---|---|
ValueError
|
如果在提供的状态格式中找不到任何消息。 |
示例
在 ReAct 智能体中的基本用法
from langgraph.graph import StateGraph
from langgraph.prebuilt import ToolNode, tools_condition
from typing_extensions import TypedDict
class State(TypedDict):
messages: list
graph = StateGraph(State)
graph.add_node("llm", call_model)
graph.add_node("tools", ToolNode([my_tool]))
graph.add_conditional_edges(
"llm",
tools_condition, # Routes to "tools" or "__end__"
{"tools": "tools", "__end__": "__end__"}
)
自定义消息键
注意
此函数旨在与 ToolNode 和标准的 LangGraph 模式无缝协作。当存在工具调用时,它期望最后一条消息是 AIMessage,这是工具调用语言模型的标准输出格式。
ValidationNode ¶
基类:RunnableCallable
一个验证上一条 AIMessage 中所有工具请求的节点。
它既可以在带有“messages”键的 StateGraph 中使用,也可以在 MessageGraph 中使用。
注意
此节点实际上并不运行工具,它只验证工具调用,这对于提取和其他需要生成符合复杂模式的结构化输出而又不丢失原始消息和工具ID(用于多轮对话)的用例非常有用。
参数
名称 | 类型 | 描述 | 默认值 |
---|---|---|---|
schemas
|
Sequence[Union[BaseTool, Type[BaseModel], Callable]]
|
用于验证工具调用的模式列表。可以是以下任何一种: - 一个 pydantic BaseModel 类 - 一个 BaseTool 实例(将使用 args_schema) - 一个函数(将根据函数签名创建模式) |
必填 |
format_error
|
Optional[Callable[[BaseException, ToolCall, Type[BaseModel]], str]]
|
一个函数,它接受一个异常、一个 ToolCall 和一个模式,并返回一个格式化的错误字符串。默认情况下,它返回异常的 repr 和一条消息,提示在修复验证错误后进行响应。 |
None
|
name
|
str
|
节点的名称。 |
'validation'
|
tags
|
Optional[list[str]]
|
要添加到节点的标签列表。 |
None
|
返回
类型 | 描述 |
---|---|
Union[Dict[str, List[ToolMessage]], Sequence[ToolMessage]]
|
包含已验证内容或错误消息的 ToolMessage 列表。 |
示例
from typing import Literal, Annotated
from typing_extensions import TypedDict
from langchain_anthropic import ChatAnthropic
from pydantic import BaseModel, field_validator
from langgraph.graph import END, START, StateGraph
from langgraph.prebuilt import ValidationNode
from langgraph.graph.message import add_messages
class SelectNumber(BaseModel):
a: int
@field_validator("a")
def a_must_be_meaningful(cls, v):
if v != 37:
raise ValueError("Only 37 is allowed")
return v
builder = StateGraph(Annotated[list, add_messages])
llm = ChatAnthropic(model="claude-3-5-haiku-latest").bind_tools([SelectNumber])
builder.add_node("model", llm)
builder.add_node("validation", ValidationNode([SelectNumber]))
builder.add_edge(START, "model")
def should_validate(state: list) -> Literal["validation", "__end__"]:
if state[-1].tool_calls:
return "validation"
return END
builder.add_conditional_edges("model", should_validate)
def should_reprompt(state: list) -> Literal["model", "__end__"]:
for msg in state[::-1]:
# None of the tool calls were errors
if msg.type == "ai":
return END
if msg.additional_kwargs.get("is_error"):
return "model"
return END
builder.add_conditional_edges("validation", should_reprompt)
graph = builder.compile()
res = graph.invoke(("user", "Select a number, any number"))
# Show the retry logic
for msg in res:
msg.pretty_print()
类
名称 | 描述 |
---|---|
HumanInterruptConfig |
定义了人类中断时允许的操作的配置。 |
ActionRequest |
表示在图执行过程中对人类操作的请求。 |
HumanInterrupt |
表示由图触发的、需要人类干预的中断。 |
HumanResponse |
人类对中断提供的响应,在图执行恢复时返回。 |
HumanInterruptConfig ¶
ActionRequest ¶
HumanInterrupt ¶
基类:TypedDict
表示由图触发的、需要人类干预的中断。
当执行暂停以等待人类输入时,此对象将传递给 interrupt
函数。
属性
名称 | 类型 | 描述 |
---|---|---|
action_request |
ActionRequest
|
向人类请求的具体操作 |
config |
HumanInterruptConfig
|
定义允许的操作的配置 |
description |
Optional[str]
|
关于需要什么输入的可选详细描述 |
示例
# Extract a tool call from the state and create an interrupt request
request = HumanInterrupt(
action_request=ActionRequest(
action="run_command", # The action being requested
args={"command": "ls", "args": ["-l"]} # Arguments for the action
),
config=HumanInterruptConfig(
allow_ignore=True, # Allow skipping this step
allow_respond=True, # Allow text feedback
allow_edit=False, # Don't allow editing
allow_accept=True # Allow direct acceptance
),
description="Please review the command before execution"
)
# Send the interrupt request and get the response
response = interrupt([request])[0]
HumanResponse ¶
基类:TypedDict
人类对中断提供的响应,在图执行恢复时返回。
属性
名称 | 类型 | 描述 |
---|---|---|
type |
Literal['accept', 'ignore', 'response', 'edit']
|
响应类型: - "accept":批准当前状态,不做更改 - "ignore":跳过/忽略当前步骤 - "response":提供文本反馈或指令 - "edit":修改当前状态/内容 |
args |
Union[None, str, ActionRequest]
|
响应载荷: - None:用于忽略/接受操作 - str:用于文本响应 - ActionRequest:用于带有更新内容的编辑操作 |