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

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 serves as the execution engine that defines how agents connect and communicate. It supports three distinct topologies via the GraphBuildType enumeration:

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 acts as the single source of truth for both configuration and runtime state. It separates concerns into immutable and mutable stores:

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 (lines 147‑155) bundles optimization settings:

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

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:

# 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:

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) manages multi-agent topology through Workflow, Handoff, and Team patterns, propagating context to all registered agents during reset().
• Context (base.py) separates immutable configuration from mutable runtime state, enabling both shared-memory collaboration and strict isolation via instance control.
• ContextRuleConfig (conf.py) and PromptProcessor (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.

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