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

> Implement multi-agent collaboration in Python using Swarm, Context, and ContextRule from ai-agent-book. Master configurable state sharing and isolation for advanced AI systems.

- Repository: [Bojie Li/ai-agent-book](https://github.com/bojieli/ai-agent-book)
- Tags: how-to-guide
- Published: 2026-08-06

---

**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](https://github.com/bojieli/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`](https://github.com/bojieli/ai-agent-book/blob/main/aworld/core/agent/swarm.py) | Orchestrates multi-agent topology and execution |
| **Context** | [`aworld/core/context/base.py`](https://github.com/bojieli/ai-agent-book/blob/main/aworld/core/context/base.py) | Holds configuration and mutable runtime state |
| **ContextRule + PromptProcessor** | [`aworld/config/conf.py`](https://github.com/bojieli/ai-agent-book/blob/main/aworld/config/conf.py) + [`aworld/core/context/processor/prompt_processor.py`](https://github.com/bojieli/ai-agent-book/blob/main/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`](https://github.com/bojieli/ai-agent-book/blob/main/swarm.py) manages agent relationships through three graph construction modes:

```python
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`](https://github.com/bojieli/ai-agent-book/blob/main/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`](https://github.com/bojieli/ai-agent-book/blob/main/base.py) (lines 110-112) separates immutable configuration from mutable state:

```python
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:

```python
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:

```python

# 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`](https://github.com/bojieli/ai-agent-book/blob/main/conf.py) (lines 147-155) configures automatic compression:

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

```

### Configuring Compression Strategy

```python
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`](https://github.com/bojieli/ai-agent-book/blob/main/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:

```python

# 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

```python
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.