LangChain Tutorial (Python): running agents in parallel for intermediate developers

This tutorial shows how to run multiple LangChain agents in parallel from Python, then merge their outputs into one result. You need this when a single agent is too slow, or when you want specialized agents to handle separate subtasks like research, risk review, and summarization at the same time.

What You'll Need

• •Python 3.10+
• •langchain
• •langchain-openai
• •An OpenAI API key in OPENAI_API_KEY
• •Basic familiarity with LangChain chat models and tools
• •A terminal and a virtual environment

Install the packages:

pip install langchain langchain-openai

Step-by-Step

1. •Start by defining a shared model and a few focused tools. For parallel agents, keep each agent narrow so you can combine their outputs predictably.

import os
from typing import TypedDict

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY", "")

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

@tool
def get_market_summary(topic: str) -> str:
    """Return a short market-style summary for a topic."""
    return f"{topic}: steady demand, moderate competition, and clear enterprise use cases."

@tool
def get_risk_notes(topic: str) -> str:
    """Return a short risk assessment for a topic."""
    return f"{topic}: watch data privacy, model drift, and approval workflows."

tools = [get_market_summary, get_risk_notes]

2. •Build two agents with different jobs. One agent will focus on business value, the other on risk, and both can run independently on the same input.

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

business_prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a business analyst. Use tools when helpful."),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

risk_prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a risk analyst. Use tools when helpful."),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

business_agent = create_tool_calling_agent(llm, tools, business_prompt)
risk_agent = create_tool_calling_agent(llm, tools, risk_prompt)

business_executor = AgentExecutor(agent=business_agent, tools=tools, verbose=False)
risk_executor = AgentExecutor(agent=risk_agent, tools=tools, verbose=False)

3. •Run both agents concurrently with asyncio.gather. This is the core pattern: one request fans out into multiple specialized agent calls.

import asyncio

class ParallelResult(TypedDict):
    business: str
    risk: str

async def run_parallel_agents(topic: str) -> ParallelResult:
    query = f"Analyze {topic}. Use available tools if useful."
    
    business_task = business_executor.ainvoke({"input": query})
    risk_task = risk_executor.ainvoke({"input": query})

    business_result, risk_result = await asyncio.gather(business_task, risk_task)

    return {
        "business": business_result["output"],
        "risk": risk_result["output"],
    }

4. •Add a final synthesis step that combines both outputs into one response. In production, this is where you turn parallel agent work into something usable by another service or UI.

synth_prompt = ChatPromptTemplate.from_messages([
    ("system", "You are an executive assistant. Merge the inputs into one concise answer."),
    ("human", "Business analysis:\n{business}\n\nRisk analysis:\n{risk}\n\nWrite a final recommendation."),
])

synth_chain = synth_prompt | llm

async def main():
    result = await run_parallel_agents("AI underwriting automation")
    final = await synth_chain.ainvoke(result)
    print("BUSINESS:\n", result["business"])
    print("\nRISK:\n", result["risk"])
    print("\nFINAL:\n", final.content)

if __name__ == "__main__":
    asyncio.run(main())

5. •If you want to scale this pattern beyond two agents, wrap the tasks in a list and gather them dynamically. This keeps the code clean when you add more specialist agents later.

async def run_many(topic: str):
    query = f"Analyze {topic}. Use available tools if useful."
    
    tasks = {
        "business": business_executor.ainvoke({"input": query}),
        "risk": risk_executor.ainvoke({"input": query}),
    }

    results = await asyncio.gather(*tasks.values())
    return {name: result["output"] for name, result in zip(tasks.keys(), results)}

Testing It

Run the script with a real topic like AI underwriting automation and confirm that both agent outputs appear before the final synthesis. You should see total runtime closer to the slower of the two calls rather than the sum of both calls.

If one agent hangs or fails, check whether your environment variable is set correctly and whether your model name is valid for your account. Also verify that each executor has access to the same tool list and that your prompt includes the agent_scratchpad placeholder.

A good sanity test is to temporarily add print() statements before each ainvoke() call so you can confirm both tasks start before either finishes. If you need stronger guarantees in production, add timeouts around each task and handle exceptions per agent instead of letting one failure kill the whole batch.

Next Steps

• •Add per-agent timeouts and retry policies so one slow dependency does not block the whole workflow.
• •Replace static tools with real internal services like policy lookup, customer profile retrieval, or claims APIs.
• •Learn LangGraph next if you want stateful orchestration instead of ad hoc parallel execution

Keep learning

• •The complete AI Agents Roadmap — my full 8-step breakdown
• •Free: The AI Agent Starter Kit — PDF checklist + starter code
• •Work with me — I build AI for banks and insurance companies

By Cyprian Aarons, AI Consultant at Topiax.