LangGraph Parallel Execution Issue: AIMessages Without Corresponding ToolMessage

[ZaslavkyDi]

Hi everyone,

I ran into an issue when trying to send two parallel questions to the agent using the same thread_id. The second question, which was sent a few seconds later, resulted in the following error:
ValueError: Found AIMessages with tool_calls that do not have a corresponding ToolMessage. Here are the first few of those tool calls.

Our team uses the prebuilt LangGraph langgraph.prebuilt.create_react_agent function to create a graph and langgraph.checkpoint.postgres.aio.AsyncPostgresSaver for checkpointing.

This issue occurs because question #2 picks up the chat history, which includes messages from question #1 that is still in progress.

Is there a way to handle or prevent parallel calls within LangChain/LangGraph?

Any insights would be greatly appreciated!

[vbarda]

Can you provide a code example that illustrates the issue? From reading the problem description and the error, it doesn't seem like the issue is related to parallel tool calling (but parallel tool calling can be disabled via create_react_agent(model.bind_tools(tools, parallel_tool_calls=False), tools), which is supported by OpenAI and Anthropic)

I am not sure what your implementation looks like, but in general, if you are sending multiple inputs on the same thread, you might want to wait until the original thread is finished before sending inputs, otherwise as you pointed out you might get a malformed message history that model providers don't accept. this conceptual doc from LangGraph platform might be helpful https://langchain-ai.github.io/langgraph/concepts/double_texting/

[ZaslavkyDi]

Thanks for the quick response, @vbarda.

I’ve checked your link, and our team is indeed experiencing the “double texting” issue.

Unfortunately, the create_react_agent(model.bind_tools(tools, parallel_tool_calls=False), tools) option does not work for us because we make an API call that creates a new agent instance for execution.
Technically, this is not a case of parallel tool calling but rather two separate agents using the same thread_id to retrieve chat history.

---

More Details

We have an agent that utilizes other agents as tools. When a user hits the API endpoint, our service generates a new instance of this agent along with its agent-tools (agents utilized as tools).

Since we use a checkpointer to store and retrieve message history based on thread_id, we run into the following issue:
• Different agent instances share the same memory, leading to unintended interactions and message duplication.

---

Code to Reproduce the Issue

The code below demonstrates the double texting issue when multiple agent instances share the same memory.

Note: You may need to run this code multiple times to reproduce the error. It’s hard to reproduce this issue with a dummy example.

import os
import random
from concurrent.futures.thread import ThreadPoolExecutor
from time import sleep
from typing import Literal
import multiprocessing

from langchain_core.tools import tool, BaseTool
from langchain_openai import AzureChatOpenAI
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver

# ----- Global Config for demonstration -----
MODEL_NAME = os.getenv("MODEL_NAME_PRIMARY")
DEPLOYMENT_NAME = os.getenv("DEPLOYMENT_NAME_PRIMARY")
API_VERSION = os.getenv("OPENAI_API_VERSION_PRIMARY")
AZURE_ENDPOINT = os.getenv("AZURE_OPENAI_ENDPOINT_PRIMARY")
API_KEY = os.getenv("AZURE_OPENAI_API_KEY_PRIMARY")

# Our async tool
@tool
def get_weather_in_nyc(city: Literal["nyc"]):
    """Use this to get NYC weather information."""
    sleep(random.uniform(1, 7))
    return "It might be cloudy in NYC"


@tool
def get_weather_in_sf(city: Literal["nyc"]):
    """Use this to get SF weather information."""
    sleep(random.uniform(1, 7))
    return "It might be sunny in SF"


def worker(city: str) -> str:
    """
    This is the function that will run *inside* each new process.
    1. It creates a new event loop (because each process is independent).
    2. It creates a new agent + tool + checkpoint saver.
    3. It calls the agent asynchronously and returns the final string result.
    """
    def run_agent_for_city(city: str) -> str:
        sleep(random.uniform(1, 3))

        # Recreate the LLM within this process
        llm = AzureChatOpenAI(
            api_version=API_VERSION,
            azure_endpoint=AZURE_ENDPOINT,
            api_key=API_KEY,
            azure_deployment=DEPLOYMENT_NAME,
            model=MODEL_NAME,
            temperature=0.0,
        )

        # Create a new checkpointer in this process
        # (Beware concurrency if many processes write at once)
        with SqliteSaver.from_conn_string("checkpoints.db") as saver:
            tools = [
                get_weather_in_nyc,
                get_weather_in_sf,
            ]

            agent = create_react_agent(
                llm,
                tools=tools,
                checkpointer=saver
            )

            # Attempt to call the agent
            # We'll catch the AssertionError from unknown city
            try:
                response = agent.invoke(
                    {
                        "messages": [
                            ("user", f"What is the weather in {city}?")
                        ]
                    },
                    config={"configurable": {"thread_id": 1}}
                )
            except AssertionError as e:
                response = f"Error: {str(e)}"

            return response["messages"][-1]

    # Run the async code *within* this worker
    return run_agent_for_city(city)


def main():
    cities = ["nyc", "sf", "nyc", "nyc", "sf"]

    # Use a Pool with number of processes = # CPU cores (or specify manually)
    with ThreadPoolExecutor(2) as pool:
        # Map each city to the worker function
        results = pool.map(worker, cities)

    # Show results
    for city, weather in zip(cities, results):
        print(f"Weather in {city}: {weather}")


if __name__ == "__main__":
    main()

[vbarda]

i would say there are a couple of solutions here:

• you can wrap your react agent in a subgraph (e.g. add it to a parent graph with a single node) and then use Send in the parent graph to run the subgraph node with multiple inputs concurrently. this should ensure that the results are be written correctly to the checkpointer. see this doc for using Send https://langchain-ai.github.io/langgraph/how-tos/map-reduce/
• if you must keep the current setup and use the same thread, you would need to handle double-texting yourself (e.g. possibly wait for one of the agent runs to finish), although this probably conflicts with the idea of running the worker on multiple inputs concurrently

[ZaslavkyDi]

I will check this out! Thanks for the answer

[ZaslavkyDi]

Initially, I thought I could reuse the _valid_chat_history(...) function within the state_modifier function to gain more control over the issue and handle it more effectively.

However, during my experiments, I discovered that _valid_chat_history is executed before the state_modifier function.

Simplified Execution Flow (I may have missed some steps):
1. Success Path: acall_model() -> _valid_chat_history() → If OK, then → acall_model() -> ainvoke() -> state_modifier()
2. Failure Path: acall_model() -> _valid_chat_history() → If FAILED, then → raise ValueError(...)

Reference: acall_model

I realized that I could achieve the desired behavior by overriding the create_react_agent and acall_model() functions, but I’m wondering if there is a way to reuse the existing implementation instead.

Just an idea for Improvement

Would it make sense to introduce a more flexible chat history validation mechanism, similar to state_modifier, or provide an option to disable it when needed?
This would allow developers to implement custom chat history validation and influence the execution flow, making the system more adaptable to different use cases.

[vbarda]

This does sound nice, will look more into this, however i don't think this is the right way to solve your original issue, see my response above