如何将传统的 LangChain 智能体迁移到 LangGraph

前置条件

本指南假设您熟悉以下概念：

我们重点介绍如何从传统的 LangChain 智能体迁移到更灵活的 LangGraph 智能体。 LangChain 智能体（特别是其中的 AgentExecutor）拥有多个配置参数。在本笔记本中，我们将展示这些参数如何通过 create_react_agent 预构建辅助方法映射到 LangGraph react 智能体执行器。

注意

在 LangGraph 中，图替代了 LangChain 的代理执行器。它管理代理的循环，并在其状态中将便签（scratchpad）跟踪为消息。LangChain 的“代理”对应于您提供的提示词和大型语言模型（LLM）。

前置条件

本指南使用 OpenAI 作为大语言模型（LLM）。请安装依赖项以运行。

%%capture --no-stderr
%pip install -U langgraph langchain langchain-openai

然后，设置您的 OpenAI API 密钥。

import getpass
import os

if "OPENAI_API_KEY" not in os.environ:
    os.environ["OPENAI_API_KEY"] = getpass.getpass("OpenAI API key:\n")

基本用法

对于基于工具调用的 ReAct 风格代理的基本创建和使用，其功能是一样的。首先，让我们定义一个模型和工具（或工具集），然后我们将使用它们来创建一个代理。

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o")

@tool
def magic_function(input: int) -> int:
    """Applies a magic function to an input."""
    return input + 2

tools = [magic_function]

query = "what is the value of magic_function(3)?"

API 参考:工具 |ChatOpenAI

对于 LangChain AgentExecutor，我们定义了一个包含智能体草稿占位符的提示。智能体的调用方式如下：

from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant"),
        ("human", "{input}"),
        # Placeholders fill up a **list** of messages
        ("placeholder", "{agent_scratchpad}"),
    ]
)


agent = create_tool_calling_agent(model, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools)

agent_executor.invoke({"input": query})

API 参考:代理执行器 |创建工具调用代理 |ChatPromptTemplate

{'input': 'what is the value of magic_function(3)?',
 'output': 'The value of `magic_function(3)` is 5.'}

LangGraph 的 react 代理执行器管理一个由消息列表定义的状态。它将持续处理该列表，直到代理输出中不再包含工具调用。要启动它，我们输入一个消息列表。输出将包含图的全部状态——在本例中为对话历史。

from langgraph.prebuilt import create_react_agent

langgraph_agent_executor = create_react_agent(model, tools)


messages = langgraph_agent_executor.invoke({"messages": [("human", query)]})
{
    "input": query,
    "output": messages["messages"][-1].content,
}

API 参考:create_react_agent

{'input': 'what is the value of magic_function(3)?',
 'output': 'The value of `magic_function(3)` is 5.'}

message_history = messages["messages"]

new_query = "Pardon?"

messages = langgraph_agent_executor.invoke(
    {"messages": message_history + [("human", new_query)]}
)
{
    "input": new_query,
    "output": messages["messages"][-1].content,
}

{'input': 'Pardon?',
 'output': 'The result of applying `magic_function` to the input value 3 is 5.'}

提示模板

使用传统的 LangChain 智能体时，您必须传入一个提示模板。您可以利用它来控制智能体的行为。

使用 LangGraph react 智能体执行器，默认情况下没有提示词。您可以通过以下几种方式实现对智能体的类似控制：

将系统消息作为输入传入
使用系统消息初始化智能体
使用函数初始化代理，该函数用于在将消息传递到模型之前转换图状态中的消息。
使用 Runnable 初始化智能体，以便在将消息传递到模型之前转换图状态中的消息。这包括传递提示模板。

让我们来看看下面所有这些内容。我们将传入自定义指令，使代理用西班牙语响应。

首先，使用 AgentExecutor：

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant. Respond only in Spanish."),
        ("human", "{input}"),
        # Placeholders fill up a **list** of messages
        ("placeholder", "{agent_scratchpad}"),
    ]
)


agent = create_tool_calling_agent(model, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools)

agent_executor.invoke({"input": query})

{'input': 'what is the value of magic_function(3)?',
 'output': 'El valor de magic_function(3) es 5.'}

现在，让我们向 react agent executor 传递一个自定义的系统消息。

LangGraph 的预构建create_react_agent不直接将提示模板作为参数，而是接受一个prompt参数。该参数会在调用 LLM 之前修改图状态，其值可以是以下四种之一：

一个 SystemMessage，它被添加到消息列表的开头。
一个 string，它被转换为 SystemMessage 并添加到消息列表的开头。
一个 Callable，它应该接收完整的图状态。然后输出被传递给语言模型。
或一个 Runnable，它应接收完整的图状态。输出随后被传递给语言模型。

以下是它在实际操作中的样子：

from langchain_core.messages import SystemMessage
from langgraph.prebuilt import create_react_agent

system_message = "You are a helpful assistant. Respond only in Spanish."
# This could also be a SystemMessage object
# system_message = SystemMessage(content="You are a helpful assistant. Respond only in Spanish.")

langgraph_agent_executor = create_react_agent(model, tools, prompt=system_message)


messages = langgraph_agent_executor.invoke({"messages": [("user", query)]})

API 参考:系统消息 |create_react_agent

我们还可以传入任意函数或可运行对象。该函数/可运行对象应接收图状态作为输入，并输出一个消息列表。在此处我们可以对所有类型的消息进行任意格式化。在本例中，让我们向消息列表的开头添加一个 SystemMessage，并在末尾追加另一个用户消息。

from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.prebuilt import create_react_agent
from langgraph.prebuilt.chat_agent_executor import AgentState

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant. Respond only in Spanish."),
        ("placeholder", "{messages}"),
        ("user", "Also say 'Pandamonium!' after the answer."),
    ]
)

# alternatively, this can be passed as a function, e.g.
# def prompt(state: AgentState):
#     return (
#         [SystemMessage(content="You are a helpful assistant. Respond only in Spanish.")] +
#         state["messages"] +
#         [HumanMessage(content="Also say 'Pandamonium!' after the answer.")]
#     )


langgraph_agent_executor = create_react_agent(model, tools, prompt=prompt)


messages = langgraph_agent_executor.invoke({"messages": [("human", query)]})
print(
    {
        "input": query,
        "output": messages["messages"][-1].content,
    }
)

API 参考:人类消息 |系统消息 |create_react_agent

{'input': 'what is the value of magic_function(3)?', 'output': 'El valor de magic_function(3) es 5. ¡Pandamonium!'}

内存

在 LangChain

借助 LangChain 的 AgentExecutor，您可以添加聊天 Memory，使其能够参与多轮对话。

from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o")
memory = InMemoryChatMessageHistory(session_id="test-session")
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant."),
        # First put the history
        ("placeholder", "{chat_history}"),
        # Then the new input
        ("human", "{input}"),
        # Finally the scratchpad
        ("placeholder", "{agent_scratchpad}"),
    ]
)


@tool
def magic_function(input: int) -> int:
    """Applies a magic function to an input."""
    return input + 2


tools = [magic_function]


agent = create_tool_calling_agent(model, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools)

agent_with_chat_history = RunnableWithMessageHistory(
    agent_executor,
    # This is needed because in most real world scenarios, a session id is needed
    # It isn't really used here because we are using a simple in memory ChatMessageHistory
    lambda session_id: memory,
    input_messages_key="input",
    history_messages_key="chat_history",
)

config = {"configurable": {"session_id": "test-session"}}
print(
    agent_with_chat_history.invoke(
        {"input": "Hi, I'm polly! What's the output of magic_function of 3?"}, config
    )["output"]
)
print("---")
print(agent_with_chat_history.invoke({"input": "Remember my name?"}, config)["output"])
print("---")
print(
    agent_with_chat_history.invoke({"input": "what was that output again?"}, config)[
        "output"
    ]
)

The output of the magic function when the input is 3 is 5.
---
Yes, you mentioned your name is Polly.
---
The output of the magic function when the input is 3 is 5.

在 LangGraph

记忆只是持久化，即检查点。

向 agent 添加一个 checkpointer，即可免费获得聊天记忆功能。

from langgraph.checkpoint.memory import MemorySaver  # an in-memory checkpointer
from langgraph.prebuilt import create_react_agent

system_message = "You are a helpful assistant."
# This could also be a SystemMessage object
# system_message = SystemMessage(content="You are a helpful assistant. Respond only in Spanish.")

memory = MemorySaver()
langgraph_agent_executor = create_react_agent(
    model, tools, prompt=system_message, checkpointer=memory
)

config = {"configurable": {"thread_id": "test-thread"}}
print(
    langgraph_agent_executor.invoke(
        {
            "messages": [
                ("user", "Hi, I'm polly! What's the output of magic_function of 3?")
            ]
        },
        config,
    )["messages"][-1].content
)
print("---")
print(
    langgraph_agent_executor.invoke(
        {"messages": [("user", "Remember my name?")]}, config
    )["messages"][-1].content
)
print("---")
print(
    langgraph_agent_executor.invoke(
        {"messages": [("user", "what was that output again?")]}, config
    )["messages"][-1].content
)

API 参考:内存保存器 |create_react_agent

The output of the magic function for the input 3 is 5.
---
Yes, you mentioned that your name is Polly.
---
The output of the magic function for the input 3 was 5.

遍历步骤

在 LangChain

使用 LangChain 的 AgentExecutor，您可以通过 stream（或异步 astream）方法或 iter 方法来迭代步骤。LangGraph 支持通过 stream 进行逐步迭代。

from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o")


prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant."),
        ("human", "{input}"),
        # Placeholders fill up a **list** of messages
        ("placeholder", "{agent_scratchpad}"),
    ]
)


@tool
def magic_function(input: int) -> int:
    """Applies a magic function to an input."""
    return input + 2


tools = [magic_function]

agent = create_tool_calling_agent(model, tools, prompt=prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools)

for step in agent_executor.stream({"input": query}):
    print(step)

API 参考:代理执行器 |创建工具调用代理 |聊天提示模板 |工具 |ChatOpenAI

{'actions': [ToolAgentAction(tool='magic_function', tool_input={'input': 3}, log="\nInvoking: `magic_function` with `{'input': 3}`\n\n\n", message_log=[AIMessageChunk(content='', additional_kwargs={'tool_calls': [{'index': 0, 'id': 'call_yyetzabaDBRX9Ml2KyqfKzZM', 'function': {'arguments': '{"input":3}', 'name': 'magic_function'}, 'type': 'function'}]}, response_metadata={'finish_reason': 'tool_calls', 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_a7d06e42a7'}, id='run-7a3a5ada-52ec-4df0-bf7d-81e5051b01b4', tool_calls=[{'name': 'magic_function', 'args': {'input': 3}, 'id': 'call_yyetzabaDBRX9Ml2KyqfKzZM', 'type': 'tool_call'}], tool_call_chunks=[{'name': 'magic_function', 'args': '{"input":3}', 'id': 'call_yyetzabaDBRX9Ml2KyqfKzZM', 'index': 0, 'type': 'tool_call_chunk'}])], tool_call_id='call_yyetzabaDBRX9Ml2KyqfKzZM')], 'messages': [AIMessageChunk(content='', additional_kwargs={'tool_calls': [{'index': 0, 'id': 'call_yyetzabaDBRX9Ml2KyqfKzZM', 'function': {'arguments': '{"input":3}', 'name': 'magic_function'}, 'type': 'function'}]}, response_metadata={'finish_reason': 'tool_calls', 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_a7d06e42a7'}, id='run-7a3a5ada-52ec-4df0-bf7d-81e5051b01b4', tool_calls=[{'name': 'magic_function', 'args': {'input': 3}, 'id': 'call_yyetzabaDBRX9Ml2KyqfKzZM', 'type': 'tool_call'}], tool_call_chunks=[{'name': 'magic_function', 'args': '{"input":3}', 'id': 'call_yyetzabaDBRX9Ml2KyqfKzZM', 'index': 0, 'type': 'tool_call_chunk'}])]}
{'steps': [AgentStep(action=ToolAgentAction(tool='magic_function', tool_input={'input': 3}, log="\nInvoking: `magic_function` with `{'input': 3}`\n\n\n", message_log=[AIMessageChunk(content='', additional_kwargs={'tool_calls': [{'index': 0, 'id': 'call_yyetzabaDBRX9Ml2KyqfKzZM', 'function': {'arguments': '{"input":3}', 'name': 'magic_function'}, 'type': 'function'}]}, response_metadata={'finish_reason': 'tool_calls', 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_a7d06e42a7'}, id='run-7a3a5ada-52ec-4df0-bf7d-81e5051b01b4', tool_calls=[{'name': 'magic_function', 'args': {'input': 3}, 'id': 'call_yyetzabaDBRX9Ml2KyqfKzZM', 'type': 'tool_call'}], tool_call_chunks=[{'name': 'magic_function', 'args': '{"input":3}', 'id': 'call_yyetzabaDBRX9Ml2KyqfKzZM', 'index': 0, 'type': 'tool_call_chunk'}])], tool_call_id='call_yyetzabaDBRX9Ml2KyqfKzZM'), observation=5)], 'messages': [FunctionMessage(content='5', additional_kwargs={}, response_metadata={}, name='magic_function')]}
{'output': 'The value of `magic_function(3)` is 5.', 'messages': [AIMessage(content='The value of `magic_function(3)` is 5.', additional_kwargs={}, response_metadata={})]}

在 LangGraph

在 LangGraph 中，事情使用流式传输或异步 astream 方法原生处理。

from langgraph.prebuilt import create_react_agent
from langgraph.prebuilt.chat_agent_executor import AgentState

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant."),
        ("placeholder", "{messages}"),
    ]
)

langgraph_agent_executor = create_react_agent(model, tools, prompt=prompt)

for step in langgraph_agent_executor.stream(
    {"messages": [("human", query)]}, stream_mode="updates"
):
    print(step)

API 参考:create_react_agent

{'agent': {'messages': [AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_IHTMrjvIHn8gFOX42FstIpr9', 'function': {'arguments': '{"input":3}', 'name': 'magic_function'}, 'type': 'function'}], 'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 14, 'prompt_tokens': 61, 'total_tokens': 75, 'completion_tokens_details': {'audio_tokens': None, 'reasoning_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_a7d06e42a7', 'finish_reason': 'tool_calls', 'logprobs': None}, id='run-1a6970da-163a-4e4d-b9b7-7e73b1057f42-0', tool_calls=[{'name': 'magic_function', 'args': {'input': 3}, 'id': 'call_IHTMrjvIHn8gFOX42FstIpr9', 'type': 'tool_call'}], usage_metadata={'input_tokens': 61, 'output_tokens': 14, 'total_tokens': 75, 'input_token_details': {'cache_read': 0}, 'output_token_details': {'reasoning': 0}})]}}
{'tools': {'messages': [ToolMessage(content='5', name='magic_function', id='51a9d3e4-734d-426f-a5a1-c6597e4efe25', tool_call_id='call_IHTMrjvIHn8gFOX42FstIpr9')]}}
{'agent': {'messages': [AIMessage(content='The value of `magic_function(3)` is 5.', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 14, 'prompt_tokens': 84, 'total_tokens': 98, 'completion_tokens_details': {'audio_tokens': None, 'reasoning_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_a20a4ee344', 'finish_reason': 'stop', 'logprobs': None}, id='run-73001576-a3dc-4552-8d81-c9ce8aec05b3-0', usage_metadata={'input_tokens': 84, 'output_tokens': 14, 'total_tokens': 98, 'input_token_details': {'cache_read': 0}, 'output_token_details': {'reasoning': 0}})]}}

`return_intermediate_steps`

在 LangChain

在 AgentExecutor 中设置此参数允许用户访问 intermediate_steps，该步骤将智能体的操作（例如工具调用）与其结果配对。

agent_executor = AgentExecutor(agent=agent, tools=tools, return_intermediate_steps=True)
result = agent_executor.invoke({"input": query})
print(result["intermediate_steps"])

[(ToolAgentAction(tool='magic_function', tool_input={'input': 3}, log="\nInvoking: `magic_function` with `{'input': 3}`\n\n\n", message_log=[AIMessageChunk(content='', additional_kwargs={'tool_calls': [{'index': 0, 'id': 'call_njTvl2RsVf4q1aMUxoYnJuK1', 'function': {'arguments': '{"input":3}', 'name': 'magic_function'}, 'type': 'function'}]}, response_metadata={'finish_reason': 'tool_calls', 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_a7d06e42a7'}, id='run-c9dfe3ab-2db6-4592-851e-89e056aeab32', tool_calls=[{'name': 'magic_function', 'args': {'input': 3}, 'id': 'call_njTvl2RsVf4q1aMUxoYnJuK1', 'type': 'tool_call'}], tool_call_chunks=[{'name': 'magic_function', 'args': '{"input":3}', 'id': 'call_njTvl2RsVf4q1aMUxoYnJuK1', 'index': 0, 'type': 'tool_call_chunk'}])], tool_call_id='call_njTvl2RsVf4q1aMUxoYnJuK1'), 5)]

在 LangGraph

默认情况下，LangGraph 应用中的 react 代理执行器会将所有消息追加到中央状态。因此，只需查看完整状态即可轻松观察任何中间步骤。

from langgraph.prebuilt import create_react_agent

langgraph_agent_executor = create_react_agent(model, tools=tools)

messages = langgraph_agent_executor.invoke({"messages": [("human", query)]})

messages

API 参考:create_react_agent

{'messages': [HumanMessage(content='what is the value of magic_function(3)?', additional_kwargs={}, response_metadata={}, id='1abb52c2-4bc2-4d82-bd32-5a24c3976b0f'),
  AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_XfQD6C7rAalcmicQubkhJVFq', 'function': {'arguments': '{"input":3}', 'name': 'magic_function'}, 'type': 'function'}], 'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 14, 'prompt_tokens': 55, 'total_tokens': 69, 'completion_tokens_details': {'audio_tokens': None, 'reasoning_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_a20a4ee344', 'finish_reason': 'tool_calls', 'logprobs': None}, id='run-34f02786-5b5c-4bb1-bd9e-406c81944a24-0', tool_calls=[{'name': 'magic_function', 'args': {'input': 3}, 'id': 'call_XfQD6C7rAalcmicQubkhJVFq', 'type': 'tool_call'}], usage_metadata={'input_tokens': 55, 'output_tokens': 14, 'total_tokens': 69, 'input_token_details': {'cache_read': 0}, 'output_token_details': {'reasoning': 0}}),
  ToolMessage(content='5', name='magic_function', id='cbc9fadf-1962-4ed7-b476-348c774652be', tool_call_id='call_XfQD6C7rAalcmicQubkhJVFq'),
  AIMessage(content='The value of `magic_function(3)` is 5.', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 14, 'prompt_tokens': 78, 'total_tokens': 92, 'completion_tokens_details': {'audio_tokens': None, 'reasoning_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_a7d06e42a7', 'finish_reason': 'stop', 'logprobs': None}, id='run-547e03d2-872d-4008-a38d-b7f739a77df5-0', usage_metadata={'input_tokens': 78, 'output_tokens': 14, 'total_tokens': 92, 'input_token_details': {'cache_read': 0}, 'output_token_details': {'reasoning': 0}})]}

`max_iterations`

在 LangChain

AgentExecutor 实现了一个 max_iterations 参数，允许用户中止超过指定迭代次数的运行。

@tool
def magic_function(input: str) -> str:
    """Applies a magic function to an input."""
    return "Sorry, there was an error. Please try again."


tools = [magic_function]

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant. Respond only in Spanish."),
        ("human", "{input}"),
        # Placeholders fill up a **list** of messages
        ("placeholder", "{agent_scratchpad}"),
    ]
)

agent = create_tool_calling_agent(model, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=3,
)

agent_executor.invoke({"input": query})

[1m> Entering new AgentExecutor chain...[0m
[32;1m[1;3mLo siento, no puedo decirte directamente el valor de `magic_function(3)`. Si deseas, puedo usar la función mágica para calcularlo. ¿Te gustaría que lo hiciera?[0m

[1m> Finished chain.[0m

{'input': 'what is the value of magic_function(3)?',
 'output': 'Lo siento, no puedo decirte directamente el valor de `magic_function(3)`. Si deseas, puedo usar la función mágica para calcularlo. ¿Te gustaría que lo hiciera?'}

在 LangGraph

在 LangGraph 中，这是通过 recursion_limit 配置参数控制的。

请注意，在AgentExecutor中，“一次迭代”包含完整的工具调用和执行过程。在 LangGraph 中，每一步都会贡献到递归限制，因此我们需要将其乘以二（并加一）以获得等效结果。

如果达到递归限制，LangGraph 会抛出一个特定的异常类型，我们可以像处理 AgentExecutor 一样捕获并管理它。

from langgraph.errors import GraphRecursionError
from langgraph.prebuilt import create_react_agent

RECURSION_LIMIT = 2 * 3 + 1

langgraph_agent_executor = create_react_agent(model, tools=tools)

try:
    for chunk in langgraph_agent_executor.stream(
        {"messages": [("human", query)]},
        {"recursion_limit": RECURSION_LIMIT},
        stream_mode="values",
    ):
        print(chunk["messages"][-1])
except GraphRecursionError:
    print({"input": query, "output": "Agent stopped due to max iterations."})

API 参考:create_react_agent

content='what is the value of magic_function(3)?' additional_kwargs={} response_metadata={} id='c2489fe8-e69c-4163-876d-3cce26b28521'
content='' additional_kwargs={'tool_calls': [{'id': 'call_OyNTcO6SDAvZcBlIEknPRrTR', 'function': {'arguments': '{"input":"3"}', 'name': 'magic_function'}, 'type': 'function'}], 'refusal': None} response_metadata={'token_usage': {'completion_tokens': 14, 'prompt_tokens': 55, 'total_tokens': 69, 'completion_tokens_details': {'audio_tokens': None, 'reasoning_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_a7d06e42a7', 'finish_reason': 'tool_calls', 'logprobs': None} id='run-b65504bb-fa23-4f8a-8d6c-7edb6d16e7ff-0' tool_calls=[{'name': 'magic_function', 'args': {'input': '3'}, 'id': 'call_OyNTcO6SDAvZcBlIEknPRrTR', 'type': 'tool_call'}] usage_metadata={'input_tokens': 55, 'output_tokens': 14, 'total_tokens': 69, 'input_token_details': {'cache_read': 0}, 'output_token_details': {'reasoning': 0}}
content='Sorry, there was an error. Please try again.' name='magic_function' id='f00e0bff-54fe-4726-a1a7-127a59d8f7ed' tool_call_id='call_OyNTcO6SDAvZcBlIEknPRrTR'
content="It seems there was an error when trying to compute the value of the magic function with input 3. Let's try again." additional_kwargs={'tool_calls': [{'id': 'call_Q020rQoJh4cnh8WglIMnDm4z', 'function': {'arguments': '{"input":"3"}', 'name': 'magic_function'}, 'type': 'function'}], 'refusal': None} response_metadata={'token_usage': {'completion_tokens': 40, 'prompt_tokens': 88, 'total_tokens': 128, 'completion_tokens_details': {'audio_tokens': None, 'reasoning_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_a7d06e42a7', 'finish_reason': 'tool_calls', 'logprobs': None} id='run-556d8cb2-b47a-4826-b17d-b520982c2475-0' tool_calls=[{'name': 'magic_function', 'args': {'input': '3'}, 'id': 'call_Q020rQoJh4cnh8WglIMnDm4z', 'type': 'tool_call'}] usage_metadata={'input_tokens': 88, 'output_tokens': 40, 'total_tokens': 128, 'input_token_details': {'cache_read': 0}, 'output_token_details': {'reasoning': 0}}
content='Sorry, there was an error. Please try again.' name='magic_function' id='777212cd-8381-44db-9762-3f81951ea73e' tool_call_id='call_Q020rQoJh4cnh8WglIMnDm4z'
content="It seems there is a persistent issue in computing the value of the magic function with the input 3. Unfortunately, I can't provide the value at this time. If you have any other questions or need further assistance, feel free to ask!" additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 49, 'prompt_tokens': 150, 'total_tokens': 199, 'completion_tokens_details': {'audio_tokens': None, 'reasoning_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_a7d06e42a7', 'finish_reason': 'stop', 'logprobs': None} id='run-92ec0b90-bc8e-4851-9139-f1d976145ab7-0' usage_metadata={'input_tokens': 150, 'output_tokens': 49, 'total_tokens': 199, 'input_token_details': {'cache_read': 0}, 'output_token_details': {'reasoning': 0}}

`max_execution_time`

在 LangChain

AgentExecutor 实现了一个 max_execution_time 参数，允许用户中止超过总时间限制的运行。

import time


@tool
def magic_function(input: str) -> str:
    """Applies a magic function to an input."""
    time.sleep(2.5)
    return "Sorry, there was an error. Please try again."


tools = [magic_function]

agent = create_tool_calling_agent(model, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    max_execution_time=2,
    verbose=True,
)

agent_executor.invoke({"input": query})

[1m> Entering new AgentExecutor chain...[0m
[32;1m[1;3mLo siento, no tengo la capacidad de evaluar directamente una función llamada "magic_function" con el valor 3. Sin embargo, si me proporcionas más detalles sobre qué hace la función o cómo está definida, podría intentar ayudarte a comprender su comportamiento o resolverlo de otra manera.[0m

[1m> Finished chain.[0m

{'input': 'what is the value of magic_function(3)?',
 'output': 'Lo siento, no tengo la capacidad de evaluar directamente una función llamada "magic_function" con el valor 3. Sin embargo, si me proporcionas más detalles sobre qué hace la función o cómo está definida, podría intentar ayudarte a comprender su comportamiento o resolverlo de otra manera.'}

在 LangGraph

使用 LangGraph 的 react 代理，您可以在两个级别上控制超时。

您可以设置 step_timeout 来限制每个步骤：

from langgraph.prebuilt import create_react_agent

langgraph_agent_executor = create_react_agent(model, tools=tools)
# Set the max timeout for each step here
langgraph_agent_executor.step_timeout = 2

try:
    for chunk in langgraph_agent_executor.stream({"messages": [("human", query)]}):
        print(chunk)
        print("------")
except TimeoutError:
    print({"input": query, "output": "Agent stopped due to a step timeout."})

API 参考:create_react_agent

{'agent': {'messages': [AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_UuxSgpGaqzX84sNlKzCVOiRO', 'function': {'arguments': '{"input":"3"}', 'name': 'magic_function'}, 'type': 'function'}], 'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 14, 'prompt_tokens': 55, 'total_tokens': 69, 'completion_tokens_details': {'audio_tokens': None, 'reasoning_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_a7d06e42a7', 'finish_reason': 'tool_calls', 'logprobs': None}, id='run-24c94cbd-2962-48cf-a447-af888eb6ef86-0', tool_calls=[{'name': 'magic_function', 'args': {'input': '3'}, 'id': 'call_UuxSgpGaqzX84sNlKzCVOiRO', 'type': 'tool_call'}], usage_metadata={'input_tokens': 55, 'output_tokens': 14, 'total_tokens': 69, 'input_token_details': {'cache_read': 0}, 'output_token_details': {'reasoning': 0}})]}}
------
{'input': 'what is the value of magic_function(3)?', 'output': 'Agent stopped due to a step timeout.'}

设置整个运行单次最大超时值的另一种方法是直接使用 Python 标准库 asyncio。

import asyncio

from langgraph.prebuilt import create_react_agent

langgraph_agent_executor = create_react_agent(model, tools=tools)


async def stream(langgraph_agent_executor, inputs):
    async for chunk in langgraph_agent_executor.astream(
        {"messages": [("human", query)]}
    ):
        print(chunk)
        print("------")


try:
    task = asyncio.create_task(
        stream(langgraph_agent_executor, {"messages": [("human", query)]})
    )
    await asyncio.wait_for(task, timeout=3)
except asyncio.TimeoutError:
    print("Task Cancelled.")

API 参考:create_react_agent

{'agent': {'messages': [AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_km17xvoY7wJ5yNnXhb5V9D3I', 'function': {'arguments': '{"input":"3"}', 'name': 'magic_function'}, 'type': 'function'}], 'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 14, 'prompt_tokens': 55, 'total_tokens': 69, 'completion_tokens_details': {'audio_tokens': None, 'reasoning_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_45c6de4934', 'finish_reason': 'tool_calls', 'logprobs': None}, id='run-b44a04e5-9b68-4020-be36-98de1593eefc-0', tool_calls=[{'name': 'magic_function', 'args': {'input': '3'}, 'id': 'call_km17xvoY7wJ5yNnXhb5V9D3I', 'type': 'tool_call'}], usage_metadata={'input_tokens': 55, 'output_tokens': 14, 'total_tokens': 69, 'input_token_details': {'cache_read': 0}, 'output_token_details': {'reasoning': 0}})]}}
------
Task Cancelled.

`early_stopping_method`

在 LangChain

借助 LangChain 的 AgentExecutor，您可以配置一个 early_stopping_method，使其要么返回字符串 "Agent stopped due to iteration limit or time limit."（"force"），要么提示 LLM 最后一次进行响应（"generate"）。

from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o")


prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant."),
        ("human", "{input}"),
        # Placeholders fill up a **list** of messages
        ("placeholder", "{agent_scratchpad}"),
    ]
)


@tool
def magic_function(input: int) -> int:
    """Applies a magic function to an input."""
    return "Sorry there was an error, please try again."


tools = [magic_function]

agent = create_tool_calling_agent(model, tools, prompt=prompt)
agent_executor = AgentExecutor(
    agent=agent, tools=tools, early_stopping_method="force", max_iterations=1
)

result = agent_executor.invoke({"input": query})
print("Output with early_stopping_method='force':")
print(result["output"])

API 参考:代理执行器 |创建工具调用代理 |聊天提示模板 |工具 |ChatOpenAI

Output with early_stopping_method='force':
Agent stopped due to max iterations.

在 LangGraph

在 LangGraph 中，您可以显式地在代理外部处理响应行为，因为可以访问完整状态。

from langgraph.errors import GraphRecursionError
from langgraph.prebuilt import create_react_agent

RECURSION_LIMIT = 2 * 1 + 1

langgraph_agent_executor = create_react_agent(model, tools=tools)

try:
    for chunk in langgraph_agent_executor.stream(
        {"messages": [("human", query)]},
        {"recursion_limit": RECURSION_LIMIT},
        stream_mode="values",
    ):
        print(chunk["messages"][-1])
except GraphRecursionError:
    print({"input": query, "output": "Agent stopped due to max iterations."})

API 参考:create_react_agent

content='what is the value of magic_function(3)?' additional_kwargs={} response_metadata={} id='81fd2e50-1e6a-4871-87aa-b7c1225913a4'
content='' additional_kwargs={'tool_calls': [{'id': 'call_aaEzj3aO1RTnB0uoc9rYUIhi', 'function': {'arguments': '{"input":3}', 'name': 'magic_function'}, 'type': 'function'}], 'refusal': None} response_metadata={'token_usage': {'completion_tokens': 14, 'prompt_tokens': 55, 'total_tokens': 69, 'completion_tokens_details': {'audio_tokens': None, 'reasoning_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_a7d06e42a7', 'finish_reason': 'tool_calls', 'logprobs': None} id='run-476bc4b1-b7bf-4607-a31c-ddf09dc814c5-0' tool_calls=[{'name': 'magic_function', 'args': {'input': 3}, 'id': 'call_aaEzj3aO1RTnB0uoc9rYUIhi', 'type': 'tool_call'}] usage_metadata={'input_tokens': 55, 'output_tokens': 14, 'total_tokens': 69, 'input_token_details': {'cache_read': 0}, 'output_token_details': {'reasoning': 0}}
content='Sorry there was an error, please try again.' name='magic_function' id='dcbe7e3e-0ed4-467d-a729-2f45916ff44f' tool_call_id='call_aaEzj3aO1RTnB0uoc9rYUIhi'
content="It seems there was an error when trying to compute the value of `magic_function(3)`. Let's try that again." additional_kwargs={'tool_calls': [{'id': 'call_jr4R8uJn2pdXF5GZC2Dg3YWS', 'function': {'arguments': '{"input":3}', 'name': 'magic_function'}, 'type': 'function'}], 'refusal': None} response_metadata={'token_usage': {'completion_tokens': 40, 'prompt_tokens': 87, 'total_tokens': 127, 'completion_tokens_details': {'audio_tokens': None, 'reasoning_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_a7d06e42a7', 'finish_reason': 'tool_calls', 'logprobs': None} id='run-d94b8932-6e9e-4ab1-99f7-7dca89887ffe-0' tool_calls=[{'name': 'magic_function', 'args': {'input': 3}, 'id': 'call_jr4R8uJn2pdXF5GZC2Dg3YWS', 'type': 'tool_call'}] usage_metadata={'input_tokens': 87, 'output_tokens': 40, 'total_tokens': 127, 'input_token_details': {'cache_read': 0}, 'output_token_details': {'reasoning': 0}}
{'input': 'what is the value of magic_function(3)?', 'output': 'Agent stopped due to max iterations.'}

`trim_intermediate_steps`

在 LangChain

借助 LangChain 的 AgentExecutor，您可以使用 trim_intermediate_steps 来缩减长运行代理的中间步骤，该参数可以是整数（表示代理应保留最后 N 个步骤）或自定义函数。

例如，我们可以截断值，以便智能体仅查看最新的中间步骤。

from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o")


prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant."),
        ("human", "{input}"),
        # Placeholders fill up a **list** of messages
        ("placeholder", "{agent_scratchpad}"),
    ]
)


magic_step_num = 1


@tool
def magic_function(input: int) -> int:
    """Applies a magic function to an input."""
    global magic_step_num
    print(f"Call number: {magic_step_num}")
    magic_step_num += 1
    return input + magic_step_num


tools = [magic_function]

agent = create_tool_calling_agent(model, tools, prompt=prompt)


def trim_steps(steps: list):
    # Let's give the agent amnesia
    return []


agent_executor = AgentExecutor(
    agent=agent, tools=tools, trim_intermediate_steps=trim_steps
)


query = "Call the magic function 4 times in sequence with the value 3. You cannot call it multiple times at once."

for step in agent_executor.stream({"input": query}):
    pass

API 参考:代理执行器 |创建工具调用代理 |聊天提示模板 |工具 |ChatOpenAI

Call number: 1
Call number: 2
Call number: 3
Call number: 4
Call number: 5
Call number: 6
Call number: 7
Call number: 8
Call number: 9
Call number: 10
Call number: 11
Call number: 12
Call number: 13
Call number: 14
``````output
Stopping agent prematurely due to triggering stop condition
``````output
Call number: 15

在 LangGraph

我们可以像之前一样使用 prompt，当传入提示模板时。

from langgraph.errors import GraphRecursionError
from langgraph.prebuilt import create_react_agent
from langgraph.prebuilt.chat_agent_executor import AgentState

magic_step_num = 1


@tool
def magic_function(input: int) -> int:
    """Applies a magic function to an input."""
    global magic_step_num
    print(f"Call number: {magic_step_num}")
    magic_step_num += 1
    return input + magic_step_num


tools = [magic_function]


def _modify_state_messages(state: AgentState):
    # Give the agent amnesia, only keeping the original user query
    return [("system", "You are a helpful assistant"), state["messages"][0]]


langgraph_agent_executor = create_react_agent(
    model, tools, prompt=_modify_state_messages
)

try:
    for step in langgraph_agent_executor.stream(
        {"messages": [("human", query)]}, stream_mode="updates"
    ):
        pass
except GraphRecursionError as e:
    print("Stopping agent prematurely due to triggering stop condition")

API 参考:create_react_agent

Call number: 1
Call number: 2
Call number: 3
Call number: 4
Call number: 5
Call number: 6
Call number: 7
Call number: 8
Call number: 9
Call number: 10
Call number: 11
Call number: 12
Stopping agent prematurely due to triggering stop condition

下一步

您现在已了解如何将 LangChain 代理执行器迁移到 LangGraph。

接下来，查看其他 LangGraph 操操作指南。

前置条件​

基本用法​

提示模板​

内存​

在 LangChain​

在 LangGraph​

遍历步骤​

在 LangChain​

在 LangGraph​

return_intermediate_steps​

在 LangChain​

在 LangGraph​

max_iterations​

在 LangChain​

在 LangGraph​

max_execution_time​

在 LangChain​

在 LangGraph​

early_stopping_method​

在 LangChain​

在 LangGraph​

trim_intermediate_steps​

在 LangChain​

在 LangGraph​

下一步​

前置条件

基本用法

提示模板

内存

在 LangChain

在 LangGraph

遍历步骤

在 LangChain

在 LangGraph

`return_intermediate_steps`

在 LangChain

在 LangGraph

`max_iterations`

在 LangChain

在 LangGraph

`max_execution_time`

在 LangChain

在 LangGraph

`early_stopping_method`

在 LangChain

在 LangGraph

`trim_intermediate_steps`

在 LangChain

在 LangGraph

下一步