Parallel execution with shared memory

Multiple specialized agents analyze the same task in parallel, each from their unique perspective. Agents share findings through a common archival memory, enabling parallel processing with automatic knowledge synthesis.

When to use this pattern

The parallel execution pattern works well when you need to:

• Analyze the same content from multiple perspectives (security review, performance analysis, accessibility audit)
• Get diverse insights on a single problem (market analysis from regional experts, code review from different specialists)
• Process large datasets with different analytical approaches (sentiment analysis, topic extraction, entity recognition)
• Build comprehensive understanding through multiple specialized viewpoints

Example use case: A code review system runs three agents in parallel on the same codebase. A security analyst checks for vulnerabilities, a performance analyst identifies bottlenecks, and an accessibility analyst reviews UI compliance. Each agent inserts findings into a shared archive with descriptive tags. The system then queries the archive to produce a comprehensive review report synthesizing all perspectives.

How it works

The pattern uses two key components:

1. Shared archival memory: A common archive where all agents insert and search findings
2. Tag-based organization: Each agent tags findings with agent-id, topic, and analysis-phase for easy filtering and coordination

Agents work independently and asynchronously:

• Each agent processes the same task from their specialized perspective
• Findings are inserted into the shared archive with semantic search and tags
• Agents can search the archive to see what others have discovered
• Results are synthesized by querying the archive across all agent contributions

Key advantages:

• No coordination overhead: Agents work independently without message passing
• Scalable knowledge base: Archives handle 30,000+ memories without performance degradation
• Semantic deduplication: Search shows related findings, preventing duplicate work
• Agent-immutable design: Agents can only insert and search, preventing conflicts

Implementation

Step 1: Create shared archive

First, create the archive where all agents will contribute findings:

from letta_client import Letta
import os

client = Letta(api_key=os.getenv("LETTA_API_KEY"))

# Shared archive for all analysis findings
analysis_archive = client.archives.create(
    name="code_analysis_findings",
    description="Shared findings from parallel code analysis"
)

Step 2: Create specialized agents

Create agents with distinct perspectives, each equipped with archival memory tools:

# Security analyst
security_agent = client.agents.create(
    name="security_analyst",
    model="anthropic/claude-sonnet-4-5-20250929",
    memory_blocks=[{
        "label": "persona",
        "value": "I analyze code for security vulnerabilities, authentication issues, and data exposure risks."
    }],
    tools=["archival_memory_insert", "archival_memory_search"]
)

# Performance analyst
performance_agent = client.agents.create(
    name="performance_analyst",
    model="anthropic/claude-sonnet-4-5-20250929",
    memory_blocks=[{
        "label": "persona",
        "value": "I analyze code for performance bottlenecks, inefficient algorithms, and scalability issues."
    }],
    tools=["archival_memory_insert", "archival_memory_search"]
)

# Accessibility analyst
accessibility_agent = client.agents.create(
    name="accessibility_analyst",
    model="anthropic/claude-sonnet-4-5-20250929",
    memory_blocks=[{
        "label": "persona",
        "value": "I analyze code for accessibility compliance, ARIA labels, and screen reader compatibility."
    }],
    tools=["archival_memory_insert", "archival_memory_search"]
)

Critical requirement: Agents must have archival tools explicitly attached. Without "archival_memory_insert" and "archival_memory_search" in the tools list, agents cannot interact with the archive.

Step 3: Attach shared archive to all agents

Attach the archive to each agent so they all share the same memory:

client.agents.archives.attach(
    agent_id=security_agent.id,
    archive_id=analysis_archive.id
)

client.agents.archives.attach(
    agent_id=performance_agent.id,
    archive_id=analysis_archive.id
)

client.agents.archives.attach(
    agent_id=accessibility_agent.id,
    archive_id=analysis_archive.id
)

Step 4: Run agents in parallel

Send the same task to all agents concurrently using Python’s ThreadPoolExecutor:

from concurrent.futures import ThreadPoolExecutor, as_completed

code_to_analyze = """
def process_user_data(user_input):
    query = f"SELECT * FROM users WHERE name = '{user_input}'"
    results = db.execute(query)
    return results
"""

def analyze_code(agent_id, perspective, tags):
    """Helper function to send analysis task to an agent"""
    return client.agents.messages.create(
        agent_id=agent_id,
        messages=[{
            "role": "user",
            "content": f"""Analyze this code for {perspective} issues:

{code_to_analyze}

Use archival_memory_insert to store each finding with tags: {tags}"""
        }]
    )

# Define analysis tasks
tasks = [
    (security_agent.id, "security", ['security', 'agent-security', 'initial-scan']),
    (performance_agent.id, "performance", ['performance', 'agent-performance', 'initial-scan']),
    (accessibility_agent.id, "accessibility", ['accessibility', 'agent-accessibility', 'initial-scan'])
]

# Run analyses in parallel
with ThreadPoolExecutor(max_workers=3) as executor:
    futures = [executor.submit(analyze_code, agent_id, perspective, tags)
               for agent_id, perspective, tags in tasks]

    # Wait for all to complete
    for future in as_completed(futures):
        future.result()  # Raises exception if analysis failed

Step 5: Synthesize results

Use a synthesis agent to query the archive and combine findings:

# Create synthesis agent with archive access
synthesis_agent = client.agents.create(
    name="synthesis_agent",
    model="anthropic/claude-sonnet-4-5-20250929",
    memory_blocks=[{
        "label": "persona",
        "value": "I synthesize findings from multiple analysts into comprehensive reports."
    }],
    tools=["archival_memory_search"]
)

client.agents.archives.attach(
    agent_id=synthesis_agent.id,
    archive_id=analysis_archive.id
)

# Synthesis agent searches and combines all findings
synthesis = client.agents.messages.create(
    agent_id=synthesis_agent.id,
    messages=[{
        "role": "user",
        "content": "Search the archive for all findings. Create a comprehensive code review report grouped by severity and type."
    }]
)

Minimal example

Here’s a minimal end-to-end example:

from letta_client import Letta
import os

client = Letta(api_key=os.getenv("LETTA_API_KEY"))

# Create shared archive
archive = client.archives.create(name="analysis_findings")

# Create two agents with different perspectives
agent1 = client.agents.create(
    model="anthropic/claude-sonnet-4-5-20250929",
    memory_blocks=[{"label": "persona", "value": "I analyze security issues."}],
    tools=["archival_memory_insert", "archival_memory_search"]
)

agent2 = client.agents.create(
    model="anthropic/claude-sonnet-4-5-20250929",
    memory_blocks=[{"label": "persona", "value": "I analyze performance issues."}],
    tools=["archival_memory_insert", "archival_memory_search"]
)

# Attach archive to both agents
client.agents.archives.attach(agent_id=agent1.id, archive_id=archive.id)
client.agents.archives.attach(agent_id=agent2.id, archive_id=archive.id)

# Both agents analyze the same code in parallel
from concurrent.futures import ThreadPoolExecutor

code = "def foo(x): return x * x"

def analyze(agent_id, focus):
    return client.agents.messages.create(
        agent_id=agent_id,
        messages=[{
            "role": "user",
            "content": f"Analyze for {focus}: {code}. Store findings with archival_memory_insert."
        }]
    )

# Run both analyses in parallel
with ThreadPoolExecutor(max_workers=2) as executor:
    security_future = executor.submit(analyze, agent1.id, "security")
    performance_future = executor.submit(analyze, agent2.id, "performance")
    security_future.result()
    performance_future.result()

# Agent2 can search and see all findings
response = client.agents.messages.create(
    agent_id=agent2.id,
    messages=[{
        "role": "user",
        "content": "Search the archive for all findings. Summarize what was discovered."
    }]
)

# Cleanup
client.agents.delete(agent1.id)
client.agents.delete(agent2.id)
client.archives.delete(archive.id)

Best practices

Use descriptive tags: Tag findings with agent-id, topic, and phase (['security', 'agent-security', 'initial-scan']) for easy filtering and organization.

Enable semantic search: Before inserting, agents should search the archive for related findings to avoid duplicating work other agents have already discovered.

Keep agents specialized: Each agent should have a clear, distinct perspective. Overlapping responsibilities reduce the value of parallel analysis.

Design clear personas: Give each agent a specific analytical focus and domain expertise in their persona.

Monitor archive growth: Check archive size and tag distribution to ensure balanced contributions across agents.

Handle large tasks: For large codebases or datasets, break the work into smaller chunks and run parallel analysis on each chunk.