智能体
langgraph.prebuilt.chat_agent_executor ¶
| 函数 | 描述 |
|---|---|
create_react_agent |
创建一个代理图,该图循环调用工具,直到满足停止条件。 |
AgentState 已弃用 ¶
基类:TypedDict
已弃用
AgentState 已移至 langchain.agents。请将您的导入更新为 from langchain.agents import AgentState。
代理的状态。
create_react_agent 已弃用 ¶
create_react_agent(
model: 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: Sequence[BaseTool | Callable | dict[str, Any]] | ToolNode,
*,
prompt: Prompt | None = None,
response_format: StructuredResponseSchema
| tuple[str, StructuredResponseSchema]
| None = None,
pre_model_hook: RunnableLike | None = None,
post_model_hook: RunnableLike | None = None,
state_schema: StateSchemaType | None = None,
context_schema: type[Any] | None = None,
checkpointer: Checkpointer | None = None,
store: BaseStore | None = None,
interrupt_before: list[str] | None = None,
interrupt_after: list[str] | None = None,
debug: bool = False,
version: Literal["v1", "v2"] = "v2",
name: str | None = None,
**deprecated_kwargs: Any,
) -> CompiledStateGraph
已弃用
create_react_agent 已移至 langchain.agents。请将您的导入更新为 from langchain.agents import create_agent。
创建一个代理图,该图循环调用工具,直到满足停止条件。
有关使用 create_react_agent 的更多详细信息,请访问代理文档。
| 参数 | 描述 |
|---|---|
model
|
代理的语言模型。支持静态和动态模型选择。
动态函数接收图状态和运行时信息,从而实现依赖于上下文的模型选择。必须返回一个 动态模型 动态模型要求 确保返回的模型已通过
类型: |
工具
|
一个工具列表或一个
类型: |
prompt
|
一个可选的 LLM 提示。可以采用以下几种形式:
类型: |
response_format
|
一个可选的最终代理输出模式。 如果提供,输出将被格式化以匹配给定的模式,并返回在 'structured_response' 状态键中。 如果不提供, 可以作为以下形式传入:
重要
注意 代理循环结束后,图将对 LLM 进行一次单独的调用以生成结构化响应。这不是获取结构化响应的唯一策略,更多选项请参见本指南。
类型: |
pre_model_hook
|
一个可选节点,添加在 重要 必须提供
类型: |
post_model_hook
|
一个可选节点,添加在 注意 仅在
类型: |
state_schema
|
一个可选的状态模式,用于定义图状态。必须包含 注意
类型: |
context_schema
|
一个可选的运行时上下文模式。 |
checkpointer
|
一个可选的检查点保存器对象。这用于为单个线程(例如,单个对话)持久化图的状态(例如,作为聊天内存)。
类型: |
store
|
一个可选的存储对象。这用于在多个线程(例如,多个对话/用户)之间持久化数据。
类型: |
interrupt_before
|
一个可选的节点名称列表,用于在执行前中断。应为以下之一: 如果您想在执行操作之前添加用户确认或其他中断,这将非常有用。 |
interrupt_after
|
一个可选的节点名称列表,用于在执行后中断。应为以下之一: 如果您想直接返回或对输出进行额外处理,这将非常有用。 |
debug
|
一个指示是否启用调试模式的标志。
类型: |
版本
|
确定要创建的图的版本。 可以是以下之一:
类型: |
name
|
一个可选的
类型: |
config_schema 已弃用
config_schema 参数在 v0.6.0 中已弃用,并将在 v2.0.0 中移除支持。请改用 context_schema 来指定运行范围内的上下文模式。
| 返回 | 描述 |
|---|---|
CompiledStateGraph
|
一个已编译的 LangChain |
"agent" 节点使用消息列表(应用提示后)调用语言模型。如果生成的 AIMessage 包含 tool_calls,图将调用 "tools" 节点。"tools" 节点执行工具(每个 tool_call 一个工具),并将响应作为 ToolMessage 对象添加到消息列表中。然后,代理节点再次调用语言模型。该过程重复进行,直到响应中不再出现 tool_calls。然后,代理返回完整的消息列表,形式为一个包含键 '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)
langgraph.prebuilt.tool_node.ToolNode ¶
基类:RunnableCallable
一个用于在 LangGraph 工作流中执行工具的节点。
处理工具执行模式,包括函数调用、状态注入、持久化存储和控制流。管理并行执行和错误处理。
输入格式
-
包含
messages键(其值为消息列表)的图状态- 代理工作流的通用表示
- 通过
messages_key参数支持自定义消息键
-
消息列表:
[AIMessage(..., tool_calls=[...])]- 最后一条 AIMessage 中包含工具调用的消息列表
-
直接工具调用:
[{"name": "tool", "args": {...}, "id": "1", "type": "tool_call"}]- 绕过消息解析,直接执行工具
- 用于编程式工具调用和测试
输出格式
输出格式取决于输入类型和工具行为
对于常规工具:
- 字典输入 →
{"messages": [ToolMessage(...)]} - 列表输入 →
[ToolMessage(...)]
对于命令工具:
- 返回
[Command(...)]或与常规工具输出混合的列表 Command可以更新状态、触发导航或发送消息
| 参数 | 描述 |
|---|---|
工具
|
此节点可以调用的一系列工具。 支持:
|
name
|
此节点在图中的名称标识符。用于调试和可视化。
类型: |
tags
|
可选的元数据标签,与节点关联以进行筛选和组织。 |
handle_tool_errors
|
工具执行期间的错误处理配置。支持多种策略:
默认为一个可调用对象,该对象:
类型: |
messages_key
|
状态字典中包含消息列表的键。此键也将用于输出的 允许使用具有不同消息字段名称的自定义状态模式。
类型: |
示例
基本用法
from langchain.tools 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])
状态注入
from typing_extensions import Annotated
from langchain.tools import InjectedState
@tool
def context_tool(query: str, state: Annotated[dict, InjectedState]) -> str:
"""Some tool that uses state."""
return f"Query: {query}, Messages: {len(state['messages'])}"
tool_node = ToolNode([context_tool])
错误处理
def handle_errors(e: ValueError) -> str:
return "Invalid input provided"
tool_node = ToolNode([my_tool], handle_tool_errors=handle_errors)
langgraph.prebuilt.tool_node ¶
用于 LangGraph 工作流的工具执行节点。
此模块提供了在 LangGraph 中执行工具的预构建功能。
工具是模型可以调用以与外部系统、API、数据库交互或执行计算的函数。
该模块实现了以下设计模式:
- 为提高效率而并行执行多个工具调用
- 具有可自定义错误消息的健壮错误处理
- 需要访问图状态的工具的状态注入
- 需要持久化存储的工具的存储注入
- 用于高级控制流的基于命令的状态更新
关键组件
ToolNode: 在 LangGraph 工作流中执行工具的主要类InjectedState: 用于将图状态注入工具的注解InjectedStore: 用于将持久化存储注入工具的注解ToolRuntime: 工具的运行时信息,将state,context,config,stream_writer,tool_call_id和store捆绑在一起tools_condition: 用于基于工具调用进行条件路由的实用函数
典型用法
| 函数 | 描述 |
|---|---|
tools_condition |
用于工具调用工作流的条件路由函数。 |
InjectedState ¶
用于将图状态注入工具参数的注解。
此注解使工具能够访问图状态,而无需向语言模型暴露状态管理细节。使用 InjectedState 注解的工具在执行期间自动接收状态数据,同时对模型的工具调用接口保持不可见。
| 参数 | 描述 |
|---|---|
field
|
从状态字典中提取的可选键。如果为
类型: |
示例
from typing import List
from typing_extensions import Annotated, TypedDict
from langchain_core.messages import BaseMessage, AIMessage
from langchain.tools import InjectedState, ToolNode, tool
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在执行期间处理注入过程- 工具可以混合使用常规参数(由模型控制)和注入参数(由系统控制)
- 状态注入发生在模型生成工具调用之后、工具执行之前
| 方法 | 描述 |
|---|---|
__init__ |
初始化 |
InjectedStore ¶
用于将持久化存储注入工具参数的注解。
此注解使工具能够访问 LangGraph 的持久化存储系统,而无需向语言模型暴露存储细节。使用 InjectedStore 注解的工具在执行期间自动接收存储实例,同时对模型的工具调用接口保持不可见。
存储提供持久的、跨会话的数据存储,工具可用于维护上下文、用户偏好或任何需要超越单个工作流执行而持久化的数据。
警告
InjectedStore 注解需要 langchain-core >= 0.3.8
示例
from typing_extensions import Annotated
from langgraph.store.memory import InMemoryStore
from langchain.tools import InjectedStore, ToolNode, tool
@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: list[AnyMessage] | dict[str, Any] | BaseModel, messages_key: str = "messages"
) -> Literal["tools", "__end__"]
用于工具调用工作流的条件路由函数。
此实用函数实现了 ReAct 风格代理的标准条件逻辑:如果最后的 AIMessage 包含工具调用,则路由到工具执行节点;否则,结束工作流。这种模式是大多数工具调用代理架构的基础。
该函数处理 LangGraph 应用程序中常用的多种状态格式,使其在保持一致行为的同时,能够灵活适应不同的图设计。
| 参数 | 描述 |
|---|---|
state
|
用于检查工具调用的当前图状态。支持的格式: - 包含 messages 键的字典(用于 |
messages_key
|
状态中包含消息列表的键或属性名称。这允许对使用不同状态模式的图进行自定义。
类型: |
| 返回 | 描述 |
|---|---|
Literal['tools', '__end__']
|
如果最后的 |
| 引发 | 描述 |
|---|---|
ValueError
|
如果在提供的状态格式中找不到任何消息。 |
示例
在 ReAct 代理中的基本用法
from langgraph.graph import StateGraph
from langchain.tools import ToolNode
from langchain.tools.tool_node import 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,这是工具调用语言模型的标准输出格式。
langgraph.prebuilt.tool_validator.ValidationNode 已弃用 ¶
基类:RunnableCallable
已弃用
ValidationNode 已被弃用。请使用 langchain.agents 中的 create_agent 并配合自定义工具错误处理。
一个验证来自最后一条 AIMessage 的所有工具请求的节点。
它可以在 StateGraph 中与 'messages' 键一起使用。
注意
此节点实际上并不运行工具,它只验证工具调用,这对于提取和其他需要生成符合复杂模式的结构化输出,同时不丢失原始消息和工具 ID(用于多轮对话)的用例非常有用。
| 返回 | 描述 |
|---|---|
Union[Dict[str, List[ToolMessage]], Sequence[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()
langgraph.prebuilt.interrupt ¶
HumanInterruptConfig 已弃用 ¶
基类:TypedDict
已弃用
HumanInterruptConfig 已移至 langchain.agents.interrupt。请将您的导入更新为 from langchain.agents.interrupt import HumanInterruptConfig。
定义了人类中断时允许的操作的配置。
这控制了当图为等待人类输入而暂停时可用的交互选项。
| 属性 | 描述 |
|---|---|
allow_ignore |
人类是否可以选择忽略/跳过当前步骤
|
allow_respond |
人类是否可以提供文本响应/反馈
|
allow_edit |
人类是否可以编辑提供的内容/状态
|
allow_accept |
人类是否可以接受/批准当前状态
|
ActionRequest 已弃用 ¶
基类:TypedDict
已弃用
ActionRequest 已移至 langchain.agents.interrupt。请将您的导入更新为 from langchain.agents.interrupt import ActionRequest。
表示在图执行过程中请求人类操作。
包含操作类型以及该操作所需的任何相关参数。
| 属性 | 描述 |
|---|---|
action |
被请求的操作的类型或名称(例如,
|
args |
操作所需的键值对参数
|
HumanInterrupt 已弃用 ¶
基类:TypedDict
已弃用
HumanInterrupt 已移至 langchain.agents.interrupt。请将您的导入更新为 from langchain.agents.interrupt import HumanInterrupt。
表示由图触发的、需要人类干预的中断。
当执行暂停以等待人类输入时,此对象被传递给 interrupt 函数。
| 属性 | 描述 |
|---|---|
action_request |
向人类请求的具体操作
|
配置 |
定义了允许哪些操作的配置
|
描述 |
关于需要什么输入的可选详细描述
|
示例
# 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 |
响应的类型
类型: |
args |
响应的有效负载
类型: |