Skip to main content
Open In ColabOpen on GitHub

如何从旧版LangChain代理迁移到LangGraph

Prerequisites

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

在这里，我们重点讨论如何从传统的LangChain代理转向更灵活的LangGraph代理。 LangChain代理（特别是AgentExecutor）有多个配置参数。 在本笔记本中，我们将展示这些参数如何映射到使用create_react_agent预构建辅助方法的LangGraph反应代理执行器。

note

在LangGraph中，图取代了LangChain的代理执行器。它管理代理的循环，并将草稿作为消息跟踪在其状态中。LangChain的“代理”对应于您提供的state_modifier和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 Reference:tool | 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})
{'input': 'what is the value of magic_function(3)?',
 'output': 'The value of `magic_function(3)` is 5.'}

LangGraph的react agent executor管理一个由消息列表定义的状态。它会继续处理该列表，直到代理的输出中没有工具调用为止。为了启动它，我们输入一个消息列表。输出将包含图的整个状态——在这种情况下，是对话历史。

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 Reference: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 agent executor，默认情况下没有提示。您可以通过几种方式实现对代理的类似控制：

  1. 传入系统消息作为输入
  2. 使用系统消息初始化代理
  3. 初始化代理时使用一个函数，在将消息传递给模型之前对其进行转换。

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

首先，使用 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 不直接接受提示模板作为参数，而是接受一个 state_modifier 参数。这会在调用 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, state_modifier=system_message
)


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

我们也可以传入一个任意函数。这个函数应该接收一个消息列表并输出一个消息列表。 我们可以在这里进行各种任意格式的消息处理。在这种情况下，我们只需在消息列表的开头添加一个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}"),
    ]
)


def _modify_state_messages(state: AgentState):
    return prompt.invoke({"messages": state["messages"]}).to_messages() + [
        ("user", "Also say 'Pandamonium!' after the answer.")
    ]


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


messages = langgraph_agent_executor.invoke({"messages": [("human", query)]})
print(
    {
        "input": query,
        "output": messages["messages"][-1].content,
    }
)
API Reference: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中​

内存只是持久化，也就是检查点。

向代理添加一个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, state_modifier=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
)
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)
{'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中，事情是使用stream或异步的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}"),
    ]
)


def _modify_state_messages(state: AgentState):
    return prompt.invoke({"messages": state["messages"]}).to_messages()


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

for step in langgraph_agent_executor.stream(
    {"messages": [("human", query)]}, stream_mode="updates"
):
    print(step)
API Reference: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 agent executor会将所有消息附加到中央状态。因此，只需查看完整状态，就可以轻松看到任何中间步骤。

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 Reference: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})


> Entering new AgentExecutor chain...
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?

> Finished chain.
{'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 Reference: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})


> Entering new AgentExecutor chain...
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.

> Finished chain.
{'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 Reference: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 Reference: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，以返回一个字符串说“代理因迭代限制或时间限制而停止。”（"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"])
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 Reference: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
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中​

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

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, state_modifier=_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 Reference: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 操作指南。

这个页面有帮助吗？