LangChain Tutorial (Python): running agents in parallel for beginners

This tutorial shows you how to run multiple LangChain agents in parallel from Python and collect their outputs in one place. You need this when one agent is too slow for the job, or when you want separate agents to handle different subtasks like research, summarization, and validation at the same time.

What You'll Need

• •Python 3.10+
• •An OpenAI API key
• •langchain
• •langchain-openai
• •langgraph
• •python-dotenv for local env loading
• •Basic familiarity with LangChain chat models and tools

Install the packages:

pip install langchain langchain-openai langgraph python-dotenv

Set your API key in your shell or a .env file:

export OPENAI_API_KEY="your-key-here"

Step-by-Step

1. •Start with a small agent that can answer a single task. We’ll use two independent agents with different prompts so we can run them side by side later.

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

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

research_prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a concise research assistant."),
    ("human", "Give me three bullet points about: {topic}")
])

summary_prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a concise summarizer."),
    ("human", "Summarize this topic in two sentences: {topic}")
])

research_chain = research_prompt | llm
summary_chain = summary_prompt | llm

2. •Wrap each chain in an agent-like function. In production, these functions often represent separate specialists, even if they are not full tool-using agents yet.

from typing import Dict

def run_research(topic: str) -> Dict[str, str]:
    result = research_chain.invoke({"topic": topic})
    return {"research": result.content}

def run_summary(topic: str) -> Dict[str, str]:
    result = summary_chain.invoke({"topic": topic})
    return {"summary": result.content}

3. •Run both tasks in parallel using ThreadPoolExecutor. This is the simplest beginner-friendly pattern for concurrent LLM calls because each request is independent.

from concurrent.futures import ThreadPoolExecutor, as_completed

def run_parallel_agents(topic: str) -> Dict[str, str]:
    tasks = {
        "research": lambda: run_research(topic),
        "summary": lambda: run_summary(topic),
    }

    results = {}
    with ThreadPoolExecutor(max_workers=2) as executor:
        future_map = {executor.submit(fn): name for name, fn in tasks.items()}

        for future in as_completed(future_map):
            results.update(future.result())

    return results

4. •Add a simple orchestrator so you can call everything from one entry point. This makes it easier to plug into a FastAPI route, CLI script, or background worker later.

def main():
    topic = "banking fraud detection"
    results = run_parallel_agents(topic)

    print("\n=== Research ===")
    print(results["research"])

    print("\n=== Summary ===")
    print(results["summary"])

if __name__ == "__main__":
    main()

5. •If you want LangGraph to manage parallel branches explicitly, use it once you outgrow raw threads. The graph below runs two nodes from the same input and merges their outputs into one state object.

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class AgentState(TypedDict):
    topic: str
    research: str
    summary: str

def research_node(state: AgentState):
    return {"research": run_research(state["topic"])["research"]}

def summary_node(state: AgentState):
    return {"summary": run_summary(state["topic"])["summary"]}

graph = StateGraph(AgentState)
graph.add_node("research_node", research_node)
graph.add_node("summary_node", summary_node)
graph.add_edge(START, "research_node")
graph.add_edge(START, "summary_node")
graph.add_edge("research_node", END)
graph.add_edge("summary_node", END)

app = graph.compile()

Testing It

Run the script and confirm that both outputs appear for the same topic. You should see the research bullets and the summary printed separately, which proves both calls completed successfully.

If one output is missing, check your API key first and then verify that both functions return dictionaries with matching keys. If you use LangGraph, make sure your state type includes every field you want to merge back into the final result.

For a more realistic test, replace the sample topic with something longer like "claims automation in insurance" and compare response times against sequential execution. Parallel execution should finish faster than calling each chain one after the other.

Next Steps

• •Add real tools to each agent using create_tool_calling_agent
• •Replace ThreadPoolExecutor with LangGraph routing and conditional edges
• •Add retries and timeouts for production API reliability

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.