Manage inter-agent communication

Introduction

When agents in a multi-agent graph operate in isolation, they repeat work: one agent may rediscover facts another already found, produce outputs that conflict with a sibling's conclusions, or lack the context it needs to specialize its response. Without a shared communication layer, coordination collapses into redundant, expensive LLM calls with no traceability across the graph. In this lesson you will design a shared-state schema that uses append-only reducers and a live knowledge-base dictionary, then implement a context-aware worker node that reads other agents' findings before generating its own response and writes its discoveries back — so every agent in the graph can both contribute to and benefit from a running shared context.

Key Terminology

• Append-only reducer — a LangGraph state field declared as Annotated[List, operator.add] so that when multiple nodes return a list update, LangGraph merges (appends) all incoming lists rather than overwriting the field with the last writer's value.
• EnhancedSupervisorState — the six-group TypedDict schema defined in this lesson that bundles request tracking, routing history, agent results, a shared knowledge base, subtask management, and execution metrics into a single graph state object.
• Shared knowledge base — the shared_context: dict field in EnhancedSupervisorState that every agent both reads before generating its response and writes to after completing its work, acting as the live coordination layer across the graph.
• Context-aware worker node — a worker function (context_aware_worker_node) that assembles summaries from all other agents in shared_context into its LLM prompt before generating output, then writes its own summary, key_findings, and timestamp back under its name.
• AgentMessage — a typed envelope TypedDict with from_agent, to_agent, message_type (one of "request", "response", "context", "error"), content, and timestamp fields that gives inter-agent communication a traceable, structured shape.
• Spread-preserve write pattern — the {**shared, config.name: {...}} technique used when updating shared_context, which copies all existing agent keys before adding the new one so no prior agent's findings are silently deleted.

Concepts

Why Reducers Are the Foundation of Safe Parallel State

In LangGraph, every node returns a plain dict whose values are merged back into the graph state. For scalar fields, this merge is a simple overwrite — whichever node runs last wins. For list fields shared by many agents, overwrite is catastrophic: if agent_results carries no reducer and two nodes each return {"agent_results": [...]}, one result is silently dropped. The Annotated[List, operator.add] reducer tells LangGraph to concatenate incoming lists instead, so results from every node accumulate correctly regardless of execution order. This is why every collaborative list in EnhancedSupervisorState — routing_history, agent_results, consulted_agents, completed_subtasks, errors — carries the operator.add annotation. Without it, parallel branches would race to overwrite each other (see Code Walkthrough).

The Shared Knowledge Base as a Coordination Layer

Direct agent-to-agent calls require each pair of agents to know about each other, which does not scale and is hard to trace. The shared_context dictionary solves this with a different model: agents communicate through a shared store rather than through each other. An agent writes its output to a known key (its own name), and any other agent that runs later can read that key. Because shared_context is a plain dict with no operator.add reducer, the node that runs last would ordinarily overwrite the field. The spread-preserve write pattern ({**shared, config.name: {...}}) sidesteps this: each worker copies the entire existing dict before adding its own entry, turning an ordinary dict field into an effectively additive store without needing a custom reducer. The AgentMessage TypedDict gives the messages themselves a traceable envelope — sender, receiver, type, and timestamp — so the store's contents remain auditable.

Read-Before-Write: How Context Propagates in One Pass

The coordination model only works if agents actually read what others have written before forming their own response. context_aware_worker_node enforces this by pulling every entry in shared_context that does not belong to the current agent, formatting those summaries into a prose block, and injecting that block into the LLM prompt before any generation happens. This means the first agent to run sees no prior context; the second sees the first's summary; the third sees both — building up shared understanding across a single traversal of the graph without any additional LLM calls or re-routing. After generating its response, the worker writes its own summary, key_findings, and timestamp back under its config.name key so the accumulation continues. The verification step — instantiating two workers with different names, running them sequentially against the same state, and checking that both names appear as top-level keys in shared_context — confirms this read-then-write contract is correctly wired (see Code Walkthrough).

Code Walkthrough

Now that you've seen why Reducers Are the Foundation of Safe Parallel State, the Shared Knowledge Base as a Coordination Layer, and Read-Before-Write: How Context Propagates in One Pass, this walkthrough turns them into working code.

The state schema below extends a basic supervisor graph with six field groups: request tracking, routing history, agent results, a shared knowledge base, subtask management, and execution metrics. Using Annotated[List, operator.add] as the reducer ensures lists from parallel nodes are merged rather than overwritten, which is the key structural choice that makes append-only communication safe.

Code snippetpython
1from typing import TypedDict, Annotated, List, Optional, Literal
2import operator
3from datetime import datetime
4
5class EnhancedSupervisorState(TypedDict):
6    # Request tracking
7    user_request: str
8    request_id: str
9    request_timestamp: str
10
11    # Routing state
12    current_agent: Optional[str]
13    routing_complete: bool
14    routing_history: Annotated[List[dict], operator.add]
15
16    # Agent results
17    agent_results: Annotated[List[dict], operator.add]
18    consulted_agents: Annotated[List[str], operator.add]
19
20    # Shared knowledge base all agents can read and update
21    shared_context: dict
22
23    # Task breakdown
24    subtasks: List[dict]
25    completed_subtasks: Annotated[List[str], operator.add]
26
27    # Final output and metadata
28    final_response: Optional[str]
29    total_tokens_used: int
30    execution_time_ms: int
31    errors: Annotated[List[str], operator.add]
32
33class AgentMessage(TypedDict):
34    from_agent: str
35    to_agent: str
36    message_type: Literal["request", "response", "context", "error"]
37    content: dict
38    timestamp: str

The worker node below reads shared_context before building its prompt, so it incorporates every prior agent's summary, then writes its own findings back under its name so downstream agents can do the same.

Code snippetpython
1async def context_aware_worker_node(
2    config: WorkerAgentConfig,
3    state: EnhancedSupervisorState
4) -> dict:
5    shared = state.get("shared_context", {})
6    relevant_context = "\n".join(
7        f"From {name}: {ctx.get('summary', 'No summary')}"
8        for name, ctx in shared.items()
9        if name != config.name
10    ) or "No context from other agents"
11
12    prompt = f"""{config.system_prompt}
13
14Shared Context from Other Agents:
15{relevant_context}
16
17Your Task:
18{state.get("task_for_agent", state["user_request"])}
19
20Respond as JSON with a 'summary' for other agents and 'details' for full output."""
21
22    async with httpx.AsyncClient(timeout=90.0) as client:
23        response = await client.post(
24            f"{GEMINI_PROXY_URL}/v1/chat/completions",
25            json={
26                "model": "gemini-2.0-flash",
27                "messages": [{"role": "user", "content": prompt}],
28                "temperature": config.temperature,
29                "response_format": {"type": "json_object"}
30            },
31            headers={"Authorization": "Bearer student-token"}
32        )
33        output = json.loads(response.json()["choices"][0]["message"]["content"])
34
35    new_shared_context = {
36        **shared,
37        config.name: {
38            "summary": output.get("summary", "Work completed"),
39            "key_findings": output.get("key_findings", []),
40            "timestamp": datetime.now().isoformat()
41        }
42    }
43
44    return {
45        "agent_results": [{"agent": config.name, "output": output}],
46        "consulted_agents": [config.name],
47        "shared_context": new_shared_context
48    }

To verify the wiring, instantiate two worker nodes with different config.name values, invoke them sequentially against the same EnhancedSupervisorState, and confirm that state["shared_context"] contains both names as top-level keys — if it does, inter-agent context passing is working correctly.

Do's and Don'ts

Having walked through managing inter-agent communication above, the following Do's and Don'ts distill it into practice.

Do's

1. ✓Do use Annotated[List, operator.add] for every append-only field in EnhancedSupervisorState — this reducer merges lists from parallel nodes rather than overwriting them, which is what makes agent_results, routing_history, and consulted_agents safe to write from concurrent worker nodes without race conditions.
2. ✓Do read shared_context at the top of context_aware_worker_node before constructing the prompt — building relevant_context from prior agents' summaries before the LLM call is what prevents redundant rediscovery and lets each agent specialize its response based on what sibling agents already found.
3. ✓Do write each worker's findings back into shared_context under its own config.name key, including both a "summary" and "key_findings" field — the summary field is what downstream agents pull into their prompts, so omitting it causes the relevant_context loop to fall back to "No summary" and breaks the inter-agent context chain.

Don'ts

1. ✗Don't use a plain List field without operator.add for any state field written by multiple nodes — LangGraph's default reducer overwrites the list with the last writer's value, so a sibling node's agent_results entries will be silently dropped when two worker nodes run in parallel.
2. ✗Don't include the current agent's own name when reading shared_context — the if name != config.name guard in the context-building loop exists specifically to prevent an agent from hallucinating on its own previous output; removing it causes self-referential prompt contamination that compounds across retries.
3. ✗Don't return a brand-new shared_context dict without spreading the existing state first ({**shared, config.name: {...}}) — overwriting the dict wholesale instead of merging it clears every other agent's findings, collapsing the shared knowledge base to a single agent's output regardless of how many nodes have already written into it.