Any way to store intermediate steps as part of AgentExecutor's invoke in history?

[ni-todo-spot]

Hey all,
Consider the following snippet emulating my project's code:

    tools = ...
    llm = ChatOpenAI(
        model=...,
        temperature=0,
        timeout=120,
        seed=1,
    ).bind_tools(tools)

    prompt = ChatPromptTemplate.from_messages(
        [
            ("system", ...),
            MessagesPlaceholder(variable_name="history", optional=True),
            (
                "human",
                """

            {input}


            {agent_scratchpad}""",
            ),
        ]
    )

    agent = create_openai_tools_agent(llm, tools, prompt)
    agent_executor = AgentExecutor(
        agent=agent,
        tools=tools,
        return_intermediate_steps=True,
        max_iterations=10,
    )
    runnable = RunnableWithMessageHistory(
        agent_executor,
        get_session_history,
        input_messages_key="input",
        history_messages_key="history",
        # https://python.langchain.com/v0.2/docs/how_to/message_history/#customization
        history_factory_config=[
            ConfigurableFieldSpec(
                id="session_id",
                annotation=int,
                name="Conversation ID",
                description="Unique identifier for the conversation.",
                default="",
                is_shared=True,
            ),
        ],
    )

The following setup works in terms of plain functionality.
The thing is the intermediate steps are not saved to history, so I do it explicitly in my code:

# other module in project calling the code above as function
history = runnable.get_session_history(session_id)

# the following row fetches the result & stores the input & output messages in history - but not intermediate steps:
result = runnable.invoke(....)

# so, adding that messages explicitly - this chronologically wrong order:
history.add_messages(format_to_tool_messages(result["intermediate_steps"]))

Any ideas for how to save the intermediate steps as part of the invoke call?
Ideally they'll be stored after the input human message but before final output ai message.

Using latest langchain version.
Using SQLChatMessageHistory with the agent.

Many thanks in advance!

[dosubot[bot]]

Hey there, @ni-todo-spot! 👋 I'm here to help you out with any bugs, questions, or contributions you need. Let's tackle this issue together!

To store intermediate steps as part of the AgentExecutor's invoke call in the message history, you can use the agent_scratchpad to format and place these steps between the input human message and the final output AI message. The provided code already includes a mechanism to handle this through the agent_scratchpad.

Here is the relevant part of the code that achieves this:

agent = (
    {
        "input": lambda x: x["input"],
        "chat_history": lambda x: _format_chat_history(x["chat_history"]),
        "agent_scratchpad": lambda x: format_to_openai_function_messages(
            x["intermediate_steps"]
        ),
    }
    | prompt
    | llm_with_tools
    | OpenAIFunctionsAgentOutputParser()
)

In this setup:

• chat_history is formatted using _format_chat_history.
• agent_scratchpad is used to format the intermediate steps with format_to_openai_function_messages.

This ensures that the intermediate steps are placed in the message history after the input human message but before the final output AI message.

Additionally, you can wrap the AgentExecutor in a RunnableWithMessageHistory class to internally manage history messages. This approach allows you to handle the message history more effectively:

from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

demo_ephemeral_chat_history_for_chain = ChatMessageHistory()

conversational_agent_executor = RunnableWithMessageHistory(
    agent_executor,
    lambda session_id: demo_ephemeral_chat_history_for_chain,
    input_messages_key="messages",
    output_messages_key="output",
)

conversational_agent_executor.invoke(
    {"messages": [HumanMessage("I'm Nemo!")]},
    {"configurable": {"session_id": "unused"}},
)

This setup ensures that the intermediate steps are properly managed and stored in the message history, maintaining the conversational context.

To further illustrate how intermediate steps can be formatted and added to the message history, consider the following example from the LangChain tests:

from langchain_core.agents import AgentAction
from langchain_core.messages import AIMessage, HumanMessage

from langchain.agents.format_scratchpad.log_to_messages import format_log_to_messages

def test_single_intermediate_step_default_response() -> None:
    intermediate_steps = [
        (AgentAction(tool="Tool1", tool_input="input1", log="Log1"), "Observation1")
    ]
    expected_result = [AIMessage(content="Log1"), HumanMessage(content="Observation1")]
    assert format_log_to_messages(intermediate_steps) == expected_result

def test_multiple_intermediate_steps_default_response() -> None:
    intermediate_steps = [
        (AgentAction(tool="Tool1", tool_input="input1", log="Log1"), "Observation1"),
        (AgentAction(tool="Tool2", tool_input="input2", log="Log2"), "Observation2"),
        (AgentAction(tool="Tool3", tool_input="input3", log="Log3"), "Observation3"),
    ]
    expected_result = [
        AIMessage(content="Log1"),
        HumanMessage(content="Observation1"),
        AIMessage(content="Log2"),
        HumanMessage(content="Observation2"),
        AIMessage(content="Log3"),
        HumanMessage(content="Observation3"),
    ]
    assert format_log_to_messages(intermediate_steps) == expected_result

This example demonstrates how intermediate steps can be converted into a sequence of AIMessage and HumanMessage objects, which can then be inserted into the message history.

Additionally, the LangChain tests provide further examples of how to manage message history using the RunnableWithMessageHistory class:

def test_input_messages() -> None:
    runnable = RunnableLambda(
        lambda messages: "you said: "
        + "\n".join(str(m.content) for m in messages if isinstance(m, HumanMessage))
    )
    store: Dict = {}
    get_session_history = _get_get_session_history(store=store)
    with_history = RunnableWithMessageHistory(runnable, get_session_history)
    config: RunnableConfig = {"configurable": {"session_id": "1"}}
    output = with_history.invoke([HumanMessage(content="hello")], config)
    assert output == "you said: hello"
    output = with_history.invoke([HumanMessage(content="good bye")], config)
    assert output == "you said: hello\ngood bye"
    output = [*with_history.stream([HumanMessage(content="hi again")], config)]
    assert output == ["you said: hello\ngood bye\nhi again"]
    assert store == {
        "1": InMemoryChatMessageHistory(
            messages=[
                HumanMessage(content="hello"),
                AIMessage(content="you said: hello"),
                HumanMessage(content="good bye"),
                AIMessage(content="you said: hello\ngood bye"),
                HumanMessage(content="hi again"),
                AIMessage(content="you said: hello\ngood bye\nhi again"),
            ]
        )
    }

This test shows how the RunnableWithMessageHistory class can be used to manage and store message history, ensuring that intermediate steps are correctly placed in the sequence of messages ^[1][2][3][4].

To continue talking to Dosu, mention @dosu.

[dosubot[bot]]

The _format_chat_history function is designed to convert a list of tuples representing chat history into a list of HumanMessage and AIMessage objects. This formatting is essential for maintaining a coherent conversation flow within the LangChain framework.

Here is the implementation of the _format_chat_history function:

def _format_chat_history(chat_history: List[Tuple[str, str]]):
    buffer = []
    for human, ai in chat_history:
        buffer.append(HumanMessage(content=human))
        buffer.append(AIMessage(content=ai))
    return buffer

Explanation:

• Input Parameter:

  • chat_history: A list of tuples, where each tuple contains two strings. The first string represents the human message, and the second string represents the AI message.

  chat_history: List[Tuple[str, str]]

• Process:

  • The function initializes an empty list called buffer.
  • It iterates over each tuple in the chat_history list.
  • For each tuple, it appends a HumanMessage object for the human message and an AIMessage object for the AI message to the buffer list.

• Return Type:

  • The function returns the buffer list, which contains the formatted HumanMessage and AIMessage objects.

  List

This function ensures that the chat history is properly formatted for use within the LangChain framework, allowing for accurate and contextually relevant responses from the language model ^[1][2][3].

To continue talking to Dosu, mention @dosu.

[ni-todo-spot]

Thanks but I could not get it working.
Can u provide a new & complete example from scratch using sqlite chat history?

[dosubot[bot]]

Certainly! Here is a complete example from scratch that uses SQLite to store chat history, including intermediate steps, as part of the AgentExecutor's invoke call.

import sqlite3
from typing import List, Tuple

from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain.agents.format_scratchpad import format_to_openai_function_messages
from langchain.agents.output_parsers import OpenAIFunctionsAgentOutputParser
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_community.utilities.tavily_search import TavilySearchAPIWrapper
from langchain_core.messages import AIMessage, HumanMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_openai import ChatOpenAI

# Create the tool
search = TavilySearchAPIWrapper()
description = """"A search engine optimized for comprehensive, accurate, \
and trusted results. Useful for when you need to answer questions \
about current events or about recent information. \
Input should be a search query. \
If the user is asking about something that you don't know about, \
you should probably use this tool to see if that can provide any information."""
tavily_tool = TavilySearchResults(api_wrapper=search, description=description)

tools = [tavily_tool]

llm = ChatOpenAI(temperature=0, model="gpt-4o-mini")

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a helpful assistant. You may not need to use tools for every query - the user may just want to chat!"),
        MessagesPlaceholder(variable_name="chat_history"),
        ("user", "{input}"),
        MessagesPlaceholder(variable_name="agent_scratchpad"),
    ]
)

llm_with_tools = llm.bind(functions=tools)

class AgentInput(BaseModel):
    input: str
    chat_history: List[Tuple[str, str]] = Field(
        ..., extra={"widget": {"type": "chat", "input": "input", "output": "output"}}
    )

def _format_chat_history(chat_history: List[Tuple[str, str]]):
    buffer = []
    for human, ai in chat_history:
        buffer.append(HumanMessage(content=human))
        buffer.append(AIMessage(content=ai))
    return buffer

# SQLite setup
conn = sqlite3.connect('chat_history.db')
c = conn.cursor()
c.execute('''
    CREATE TABLE IF NOT EXISTS chat_history (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        human TEXT,
        ai TEXT
    )
''')
conn.commit()

def save_chat_history_to_db(chat_history: List[Tuple[str, str]]):
    with sqlite3.connect('chat_history.db') as conn:
        c = conn.cursor()
        c.executemany('INSERT INTO chat_history (human, ai) VALUES (?, ?)', chat_history)
        conn.commit()

agent = create_tool_calling_agent(llm_with_tools, tools, prompt)

agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True).with_types(
    input_type=AgentInput
)

# Example usage
input_data = {
    "input": "What is the latest news on AI?",
    "chat_history": [("Hello", "Hi there! How can I assist you today?")]
}

# Save chat history to SQLite before invoking the agent
save_chat_history_to_db(input_data["chat_history"])

# Invoke the agent
result = agent_executor.invoke(input_data)
print(result)

This example sets up an SQLite database to store chat history, formats the chat history for use with the LangChain framework, and ensures that intermediate steps are saved as part of the AgentExecutor's invoke call. The save_chat_history_to_db function is used to save the chat history to the SQLite database before invoking the agent ^[1][2][3].

To continue talking to Dosu, mention @dosu.

[ni-todo-spot]

well that did not work for me. Got any other ideas?

[dosubot[bot]]

You can use the state_modifier parameter to save intermediate steps in the message history. Here is an example of how to do this:

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

def save_intermediate_steps(state: AgentState):
    # Modify the state to include intermediate steps
    return state["messages"] + [("system", "Intermediate step saved.")]

app = create_react_agent(model, tools, state_modifier=save_intermediate_steps)

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

Additionally, you can use the RunnableWithMessageHistory to automatically keep track of messages. This approach allows you to maintain a conversation history and include intermediate steps in the message history. Here is an example:

from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

store = {}

def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

agent_with_chat_history = RunnableWithMessageHistory(
    agent_executor,
    get_session_history,
    input_messages_key="input",
    history_messages_key="chat_history",
)

agent_with_chat_history.invoke(
    {"input": "hi! I'm bob"},
    config={"configurable": {"session_id": "<foo>"}},
)

agent_with_chat_history.invoke(
    {"input": "what's my name?"},
    config={"configurable": {"session_id": "<foo>"}},
)

For more advanced tracking and inspection of intermediate steps, you can use LangSmith. LangSmith allows you to log traces of your application's execution, which can be very helpful for debugging and understanding the behavior of complex chains or agents. Here is how you can set it up:

import os
from getpass import getpass

os.environ["LANGCHAIN_TRACING_V2"] = "true"
if not os.environ.get("LANGCHAIN_API_KEY"):
    os.environ["LANGCHAIN_API_KEY"] = getpass("Enter your LangChain API key: ")

By using LangSmith, you can inspect the internal workings of your chains or agents, making it easier to debug and optimize your application ^[1][2][3].

To continue talking to Dosu, mention @dosu.