# Multi-Agent Collaboration and Context Sharing/Isolation Architecture in Python

> Explore the ai-agent-book Python architecture for multi-agent collaboration. Discover how Swarm, Context, and ContextRule enable secure, configurable memory sharing and isolation.

- Repository: [Bojie Li/ai-agent-book](https://github.com/bojieli/ai-agent-book)
- Tags: architecture
- Published: 2026-07-30

---

**The `ai-agent-book` framework implements a three-pillar architecture—Swarm for topology management, Context for state isolation or sharing, and ContextRule with PromptProcessor for token-budget optimization—enabling secure multi-agent collaboration with configurable memory boundaries.**

The `bojieli/ai-agent-book` repository provides a production-ready Python framework for building distributed AI systems where multiple agents must coordinate while maintaining strict control over context boundaries. This architecture, primarily implemented in **Chapter 8**, solves the fundamental challenge of multi-agent collaboration and context sharing/isolation by treating `Context` as a first-class object that can be globally shared or locally isolated per task.

## The Three Pillars of Multi-Agent Architecture

The framework organizes multi-agent capabilities around three core abstractions that handle topology, state, and resource constraints.

### Swarm: Topology and Execution Engine

The `Swarm` class in [`chapter8/gaia-experience/AWorld/aworld/core/agent/swarm.py`](https://github.com/bojieli/ai-agent-book/blob/main/chapter8/gaia-experience/AWorld/aworld/core/agent/swarm.py) serves as the execution engine that defines how agents connect and communicate. It supports three distinct topologies via the `GraphBuildType` enumeration:

```python
class GraphBuildType(Enum):
    WORKFLOW = "workflow"
    HANDOFF = "handoff"
    TEAM = "team"

```

During initialization, the `Swarm` constructor receives a topology (list of agents or sub-graphs) and a `root_agent` that receives external inputs. The critical orchestration happens in `reset()`, which performs four key operations:

1. Instantiates a **global** `Context` (or re-uses a supplied one).
2. Builds an `AgentGraph` via the selected builder (`BUILD_CLS`).
3. Registers all agents with the `AgentFactory` using `Swarm.register_agent`.
4. Propagates the global context to every agent via `agent.context = context`.

This design creates a **shared-state** environment by default, where all agents read and write to the same `Context` instance. However, isolation can be achieved by passing a fresh `Context` object to `reset()` for specific sub-tasks.

### Context: State Management and Isolation Boundaries

The `Context` class in [`chapter8/gaia-experience/AWorld/aworld/core/context/base.py`](https://github.com/bojieli/ai-agent-book/blob/main/chapter8/gaia-experience/AWorld/aworld/core/context/base.py) acts as the single source of truth for both configuration and runtime state. It separates concerns into immutable and mutable stores:

```python
self.context_info = ContextState()      # Mutable runtime state

self.agent_info = ConfigDict()          # Immutable configuration

self.trajectories = OrderedDict()       # Execution history

```

**Immutable fields** (stored in `agent_info`) include `agent_id`, `system_prompt`, `tool_names`, and `context_rule`. **Mutable fields** (stored in `context_info`) include `messages`, `step` counters, `trajectories`, and `token_usage`.

The lifecycle of a `Context` matches its owning agent: created on agent initialization, updated during execution, and destroyed with the agent. This lifecycle management enables precise **isolation vs. sharing** control. When `Swarm.reset()` receives a fresh `Context()` instance, that sub-swarm operates in complete isolation—its token usage, messages, and intermediate state never leak to parent or sibling agents. Conversely, passing the same `Context` across multiple swarms enables rich collaboration where agents share memory, joint planning history, and cumulative token budgets.

### ContextRule and PromptProcessor: Token Budget Optimization

To prevent context window overflow during multi-agent collaboration, the framework implements automated compression and truncation via `ContextRuleConfig` and `PromptProcessor`.

The `ContextRuleConfig` class in [`chapter8/gaia-experience/AWorld/aworld/config/conf.py`](https://github.com/bojieli/ai-agent-book/blob/main/chapter8/gaia-experience/AWorld/aworld/config/conf.py) (lines 147‑155) bundles optimization settings:

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

```

- `optimization_config` controls the maximum token-budget ratio via `max_token_budget_ratio`.
- `llm_compression_config` toggles compression, selects the algorithm (`compress_type`), and sets the trigger threshold (`trigger_compress_token_length`).

The `PromptProcessor` in [`chapter8/gaia-experience/AWorld/aworld/core/context/processor/prompt_processor.py`](https://github.com/bojieli/ai-agent-book/blob/main/chapter8/gaia-experience/AWorld/aworld/core/context/processor/prompt_processor.py) instantiates per-agent with the agent's `ContextRuleConfig` and `ModelConfig`. It builds three processing pipelines:

1. **TruncateCompressor** – Quick token-budget trimming.
2. **ChunkUtils** – Optional chunking for long histories.
3. **LLMCompressor** or **LLMLinguaCompressor** – Advanced compression algorithms.

The processor exposes helper methods like `is_out_of_context()`, `should_compress_conversation()`, and `should_compress_tool_result()` that agents call before LLM requests. If the token count exceeds `get_max_tokens()`, the processor automatically applies the configured compression strategy, ensuring every LLM call respects the model's context window.

## Collaboration Patterns: Workflow, Handoff, and Team Topologies

The framework supports three distinct collaboration patterns, all leveraging the same `Context` sharing mechanism:

| Pattern | Implementation | Use Case |
|---------|---------------|----------|
| **Workflow** | DAG built by `BUILD_CLS[GraphBuildType.WORKFLOW]` with guaranteed single start node and no cycles | Linear pipelines (e.g., retrieve → plan → execute) |
| **Handoff** | Enforced pairwise connections requiring `(src, dst)` tuples | Agent-to-agent delegation where the first finishes before the second starts |
| **Team** | Root agent as coordinator with concurrent agents sharing the same `Context` | Crowd-sourced reasoning, voting, or parallel tool calls |

In all three patterns, agents **share** information naturally when using the same `Context` instance. By providing a **different `Context`** to a sub-swarm, you achieve **isolation** for sensitive sub-tasks.

## Implementing Context Sharing vs. Isolation

### Shared Context Example (Team Topology)

The following example demonstrates two agents collaborating via a shared `Context`:

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

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

        context.context_info.set("plan", "download file A")
        return [{"role": "assistant", "content": "Plan ready"}]

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

        plan = context.context_info.get("plan")
        return [{"role": "assistant", "content": f"Executed {plan}"}]

# Create shared Context

shared_ctx = Context()

# Build Team topology

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

# Initialize with shared context

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

```

Both agents read and write to `shared_ctx`, enabling real-time collaboration through shared memory.

### Isolated Sub-Task Example

To prevent context pollution during sensitive operations, instantiate a fresh `Context`:

```python

# Parent swarm with existing context

parent_swarm = Swarm(...)

# Create isolated context for sub-task

sub_ctx = Context()
sub_swarm = Swarm(
    topology=[SubAgentA(), SubAgentB()],
    root_agent=SubAgentA(),
    build_type=GraphBuildType.WORKFLOW,
)

# Run isolated execution

sub_swarm.reset(content="Sub-task request", context=sub_ctx)

while not sub_swarm._finished:
    cur = sub_swarm.cur_agent
    resp = cur.async_policy(messages=[], context=cur.context)
    # Process without affecting parent context

```

After execution, `parent_swarm`'s context remains unchanged, ensuring **context isolation** for the sub-task.

## Configuring Context Compression and Token Limits

Control memory usage across collaborative agents using `ContextRuleConfig`:

```python
from aworld.config.conf import ContextRuleConfig, LlmCompressionConfig, OptimizationConfig, ModelConfig

# Configure LLM-based compression

compression_cfg = LlmCompressionConfig(
    enabled=True,
    compress_type='llm',
    trigger_compress_token_length=10000,
    compress_model=ModelConfig(
        llm_model_name="gpt-4o-mini",
        llm_provider="openai",
        max_model_len=128000,
    ),
)

# Assemble context rule

context_rule = ContextRuleConfig(
    optimization_config=OptimizationConfig(
        enabled=True, 
        max_token_budget_ratio=0.5
    ),
    llm_compression_config=compression_cfg,
)

```

When attached to an agent via `AgentConfig`, the `PromptProcessor` automatically applies these rules, compressing or truncating conversation history when token counts exceed the configured budget ratio.

## Summary

- **Swarm** ([`swarm.py`](https://github.com/bojieli/ai-agent-book/blob/main/swarm.py)) manages multi-agent topology through Workflow, Handoff, and Team patterns, propagating context to all registered agents during `reset()`.
- **Context** ([`base.py`](https://github.com/bojieli/ai-agent-book/blob/main/base.py)) separates immutable configuration from mutable runtime state, enabling both shared-memory collaboration and strict isolation via instance control.
- **ContextRuleConfig** ([`conf.py`](https://github.com/bojieli/ai-agent-book/blob/main/conf.py)) and **PromptProcessor** ([`prompt_processor.py`](https://github.com/bojieli/ai-agent-book/blob/main/prompt_processor.py)) enforce token budgets through configurable compression strategies (truncation, LLM-based, or LLMLingua).
- Isolation is achieved by passing fresh `Context` instances to sub-swarms, while sharing occurs naturally when multiple agents reference the same `Context` object.
- All three collaboration patterns (Workflow, Handoff, Team) support both context sharing and isolation depending on the `Context` instance provided to `Swarm.reset()`.

## Frequently Asked Questions

### How does the Swarm class manage agent topology?

The `Swarm` class uses the `GraphBuildType` enumeration to determine construction mode—WORKFLOW for DAGs, HANDOFF for pairwise delegation, or TEAM for coordinated concurrency. During `reset()`, it instantiates the appropriate builder from `BUILD_CLS`, registers agents via `AgentFactory`, and binds the execution context to each agent, creating the communication graph that drives the multi-agent execution loop.

### What is the difference between shared and isolated context in multi-agent systems?

**Shared context** occurs when multiple agents receive the same `Context` instance during `Swarm.reset()`, allowing them to read each other's messages, token usage, and custom state stored in `context_info`. **Isolated context** occurs when a sub-swarm receives a fresh `Context()` instance, creating a sandbox where the sub-task's intermediate state, trajectories, and token consumption remain invisible to parent or sibling agents.

### How does the PromptProcessor handle token budget limits?

The `PromptProcessor` evaluates token counts against the `max_token_budget_ratio` defined in `ContextRuleConfig`. When `should_compress_conversation()` or `is_out_of_context()` returns true, it executes `decide_compression_strategy()` to select between truncation (fast), LLM-based compression (semantic), or LLMLingua compression (algorithmic), ensuring the message list fits within the model's `max_model_len` before the LLM call executes.

### Which collaboration pattern should I use for sequential agent delegation?

Use the **HANDOFF** topology (`GraphBuildType.HANDOFF`) for sequential delegation where Agent A must complete before Agent B begins. This pattern enforces pairwise `(src, dst)` connections and guarantees execution order, making it ideal for approval workflows or staged processing pipelines where downstream agents depend on finalized upstream outputs.