How to Implement Multi-Agent Collaboration with Context Sharing and Isolation in Python

Use the Swarm, Context, and ContextRule classes from the ai-agent-book repository to build multi-agent systems with configurable sharing or isolation of state, messages, and token budgets.

This guide walks through the production-ready architecture from Chapter 8 of the ai-agent-book repository (by bojieli). The system supports workflow, handoff, and team collaboration patterns while giving you precise control over what agents share and what remains isolated.

Core Architecture: Three Pillars

The implementation rests on three interconnected components:

Component	Source File	Responsibility
Swarm	aworld/core/agent/swarm.py	Orchestrates multi-agent topology and execution
Context	aworld/core/context/base.py	Holds configuration and mutable runtime state
ContextRule + PromptProcessor	aworld/config/conf.py + aworld/core/context/processor/prompt_processor.py	Enforces token limits via compression and truncation

Understanding how these interact is essential for implementing multi-agent collaboration with context sharing and isolation correctly.

The Swarm: Defining Multi-Agent Topology

The Swarm class in swarm.py manages agent relationships through three graph construction modes:

class GraphBuildType(Enum):
    WORKFLOW = "workflow"    # Linear DAG, single start node, no cycles

    HANDOFF = "handoff"      # Pairwise (src, dst) agent delegation

    TEAM = "team"            # Coordinator with concurrent parallel agents

Key Swarm Mechanics

When you call swarm.reset(task, context), three critical operations occur (lines referenced from swarm.py):

1. Context propagation — The supplied Context is attached to every agent via agent.context = context
2. Agent registration — All agents register with Swarm.register_agent
3. Tool sharing — Global tools in self.tools extend to every agent's tool_names

This design means context sharing is the default. Agents automatically see the same messages, token usage, and custom state when they share a Context instance.

The Context Object: Sharing vs. Isolation

The Context class in base.py (lines 110-112) separates immutable configuration from mutable state:

self.context_info = ContextState()      # Mutable: messages, step, trajectories, token_usage

self.agent_info = ConfigDict()          # Immutable: agent_id, system_prompt, tool_names, context_rule

self.trajectories = OrderedDict()       # Execution history per task

Achieving Context Sharing

Pass the same Context instance to multiple swarms or agents:

from aworld.core.agent.swarm import Swarm, GraphBuildType
from aworld.core.context.base import Context
from aworld.core.agent.base import BaseAgent

class PlannerAgent(BaseAgent):
    def async_policy(self, messages, context: Context):
        # Write plan to shared context

        context.context_info.set("plan", "retrieve documents → analyze → summarize")
        return [{"role": "assistant", "content": "Plan created"}]

class ExecutorAgent(BaseAgent):
    def async_policy(self, messages, context: Context):
        # Read plan from shared context

        plan = context.context_info.get("plan")
        # Execute and update shared state

        context.context_info.set("executed_step", "documents retrieved")
        return [{"role": "assistant", "content": f"Executed: {plan}"}]

# Shared context enables collaboration

shared_ctx = Context()
shared_ctx.context_info.set("user_query", "Summarize latest AI research")

# TEAM topology: root coordinates, others execute in parallel

team_swarm = Swarm(
    topology=[PlannerAgent(), ExecutorAgent()],
    root_agent=PlannerAgent(),
    max_steps=5,
    build_type=GraphBuildType.TEAM,
)

team_swarm.reset(content="User query", context=shared_ctx)

Both agents read from and write to shared_ctx.context_info, enabling joint planning and shared memory.

Achieving Context Isolation

Create a fresh Context for sub-tasks that must not pollute parent state:

# Parent execution has its own context

parent_ctx = Context()
parent_swarm = Swarm(..., build_type=GraphBuildType.WORKFLOW)
parent_swarm.reset(content="Main task", context=parent_ctx)

# Inside an agent: launch isolated sub-task

def async_policy(self, messages, context: Context):
    # Fresh context = complete isolation

    isolated_ctx = Context()
    isolated_ctx.context_info.set("subtask_only_data", "sensitive intermediate result")
    
    sub_swarm = Swarm(
        topology=[ResearchAgent(), VerifyAgent()],
        root_agent=ResearchAgent(),
        max_steps=3,
        build_type=GraphBuildType.HANDOFF,  # Strict pairwise handoff

    )
    sub_swarm.reset(content="Verify claim", context=isolated_ctx)
    
    # Execute sub-task...

    # Parent context remains untouched: no token usage, no messages leaked

    return result

Isolation guarantees: Token counts, message history, and custom keys in isolated_ctx never merge into parent_ctx.

ContextRule and PromptProcessor: Token Budget Management

Long-running multi-agent systems hit context window limits. The ContextRuleConfig in conf.py (lines 147-155) configures automatic compression:

class ContextRuleConfig(BaseConfig):
    optimization_config: OptimizationConfig = OptimizationConfig()
    llm_compression_config: LlmCompressionConfig = LlmCompressionConfig()

Configuring Compression Strategy

from aworld.config.conf import (
    ContextRuleConfig,
    OptimizationConfig,
    LlmCompressionConfig,
    ModelConfig,
)
from aworld.config.conf import AgentConfig

# 1. Define when and how to compress

llm_compression = LlmCompressionConfig(
    enabled=True,
    compress_type='llm',  # Alternative: 'llmlingua' for faster CPU-based compression

    trigger_compress_token_length=10000,
    compress_model=ModelConfig(
        llm_model_name="gpt-4o-mini",
        llm_provider="openai",
        max_model_len=128000,
    ),
)

# 2. Set overall budget constraint

optimization = OptimizationConfig(
    enabled=True,
    max_token_budget_ratio=0.5,  # Use at most 50% of model's context window

)

# 3. Assemble rule

context_rule = ContextRuleConfig(
    optimization_config=optimization,
    llm_compression_config=llm_compression,
)

# 4. Attach to agent configuration

agent_cfg = AgentConfig(context_rule=context_rule)
my_agent = BaseAgent(conf=agent_cfg)

How PromptProcessor Enforces Limits

The PromptProcessor in prompt_processor.py builds three pipelines per agent:

1. TruncateCompressor — Fast token-budget trimming when over limit
2. ChunkUtils — Optional semantic chunking for long histories
3. LLMCompressor or LLMLinguaCompressor — Algorithm selected by compress_type

Agents check compression needs before each LLM call:

# Inside agent.async_policy()

if self.context.rules.should_compress_conversation(self.context.messages):
    # PromptProcessor automatically applies truncate/compress strategy

    messages = self.prompt_processor.process(self.context.messages)

The decide_compression_strategy() method (lines 14-46) returns a CompressionDecision explaining whether and why compression occurred.

Collaboration Patterns in Practice

Pattern	Topology	Context Behavior	Best For
Workflow	GraphBuildType.WORKFLOW	Shared context, linear progression	ETL pipelines, sequential reasoning
Handoff	GraphBuildType.HANDOFF	Shared context, strict (src,dst) pairs	Agent delegation with completion guarantee
Team	GraphBuildType.TEAM	Shared context, concurrent execution	Voting, ensemble methods, parallel tool calls

All patterns use the same context propagation mechanism in Swarm.reset(). The difference lies in graph construction (BUILD_CLS selection) and execution scheduling.

Complete Implementation Example

from aworld.core.agent.base import BaseAgent
from aworld.core.agent.swarm import Swarm, GraphBuildType
from aworld.core.context.base import Context
from aworld.config.conf import AgentConfig, ContextRuleConfig, OptimizationConfig

class RouterAgent(BaseAgent):
    """Decides which specialist handles the query."""
    def async_policy(self, messages, context: Context):
        query = context.context_info.get("user_query")
        if "code" in query.lower():
            context.context_info.set("route_to", "coder")
        else:
            context.context_info.set("route_to", "researcher")
        return [{"role": "assistant", "content": "Routed"}]

class CoderAgent(BaseAgent):
    """Generates code; isolated sub-task for security review."""
    def async_policy(self, messages, context: Context):
        # Main work in shared context

        context.context_info.set("generated_code", "def hello(): pass")
        
        # Isolated security review doesn't leak code to other agents

        review_ctx = Context()
        review_ctx.context_info.set("code_to_review", context.context_info.get("generated_code"))
        
        review_swarm = Swarm(
            topology=[SecurityAgent()],
            root_agent=SecurityAgent(),
            max_steps=2,
            build_type=GraphBuildType.WORKFLOW,
        )
        review_swarm.reset(content="Review for vulnerabilities", context=review_ctx)
        # Execute review...

        
        # Merge only the verdict, not the full review context

        context.context_info.set("security_passed", True)
        return [{"role": "assistant", "content": "Code generated and reviewed"}]

# Build TEAM topology with shared planning context

shared_planning_ctx = Context()
shared_planning_ctx.context_info.set("project", "Multi-agent API")

orchestrator = Swarm(
    topology=[RouterAgent(), CoderAgent(), ResearchAgent()],
    root_agent=RouterAgent(),
    max_steps=10,
    build_type=GraphBuildType.TEAM,
)

orchestrator.reset(content="Build a Python API client", context=shared_planning_ctx)

This demonstrates selective isolation: broad collaboration via shared_planning_ctx, with sensitive sub-tasks isolated in review_ctx.

Summary

• Use Swarm with GraphBuildType to define multi-agent topology (workflow, handoff, team)
• Use Context as the unit of sharing: same instance = shared state, new instance = isolation
• Use ContextRuleConfig to enforce token budgets via automatic compression and truncation
• Call swarm.reset(task, context) to bind contexts; agents automatically receive agent.context reference
• Access state through context.context_info (mutable) and context.agent_info (immutable configuration)

Frequently Asked Questions

How do I prevent one agent from seeing another agent's internal reasoning?

Create a fresh Context for the agent that performs internal reasoning. Pass this isolated context to a sub-swarm. Only explicit values you copy back to the parent context become visible to other agents.

What happens when the shared context exceeds the model's token limit?

The PromptProcessor automatically triggers compression based on ContextRuleConfig. It first attempts truncation, then applies LLM-based or LLMLingua compression if configured. The max_token_budget_ratio parameter reserves headroom to prevent hard failures.

Can agents in a TEAM topology have different context rules?

Yes. Each agent receives its own Context reference (typically shared), but agent_info (including context_rule) is agent-specific. Configure via AgentConfig(context_rule=...) when instantiating each agent. The shared context_info holds runtime state, while per-agent rules govern how that agent compresses its view of the context.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}