Using Recall with AutoGen

By Arc Labs Research8 min read

The Problem

In multi-agent systems like AutoGen, agents operate in isolation. Agent A solves a problem, Agent B re-solves the same problem independently, Agent C repeats the work again. There's no shared memory, no learning between agents, no coordination. Each agent rediscovers the same solutions independently, wasting time and tokens.

Solution: Shared Memory with Recall

Recall acts as a shared memory layer for all agents in the team. When Agent A solves something, it stores the solution. When Agent B faces a similar task, it retrieves Agent A's solution, adapts it, and avoids redundant work. Agents learn from each other.

Install

# Python Recall SDK
pip install recall-sdk

# AutoGen
pip install pyautogen

Integration Steps

Step 1: Initialize Shared Recall for the Team

Create a single Recall instance shared across all agents. This is the team memory.

from recall_sdk import Recall

# Team memory: shared by all agents
team_memory = Recall(
    name="autogen-team",
    metadata={
        "system": "multi-agent",
        "agents": ["researcher", "analyst", "coder"],
        "team_id": "marketing_squad"
    }
)

Step 2: Wrap Agents to Query Recall Before Acting

Modify each agent's context to include memories from Recall. Before an agent acts, it retrieves relevant past solutions and context.

from autogen import ConversableAgent

def agent_with_recall(agent, recall_instance, agent_role):
    """Wrapper that injects Recall context into agent decisions."""
    
    original_generate_reply = agent.generate_reply
    
    def generate_reply_with_recall(messages, sender, **kwargs):
        # Extract current task from messages
        current_task = messages[-1]["content"] if messages else ""
        
        # 1. Retrieve team memories relevant to this task
        team_insights = recall_instance.retrieve(
            query=current_task,
            filters={"team_id": "marketing_squad"},
            top_k=5
        )
        
        # 2. Inject team context into the agent's system prompt
        context_str = "\n".join([m.text for m in team_insights])
        if context_str:
            messages = messages.copy()
            messages[0]["content"] = (
                f"You are {agent_role}. Here's what the team has learned:\n{context_str}\n\n"
                + messages[0]["content"]
            )
        
        # 3. Call original agent logic
        return original_generate_reply(messages, sender, **kwargs)
    
    agent.generate_reply = generate_reply_with_recall
    return agent

# Create agents
researcher = ConversableAgent(
    name="researcher",
    system_message="You research market trends and competitor data.",
    llm_config={"model": "gpt-4o", "temperature": 0.7}
)

analyst = ConversableAgent(
    name="analyst",
    system_message="You analyze data and generate insights.",
    llm_config={"model": "gpt-4o", "temperature": 0.5}
)

# Wrap with Recall
researcher = agent_with_recall(researcher, team_memory, "researcher")
analyst = agent_with_recall(analyst, team_memory, "analyst")

Step 3: Store Decisions and Outcomes After Agent Actions

After each agent completes a task, store what it learned. This builds the team memory.

def store_agent_outcome(memory_instance, agent_name, task, outcome, metadata=None):
    """Store an agent's decision and result in shared memory."""
    memory_instance.store(
        text=f"Agent {agent_name} solved: {task}\n\nOutcome: {outcome}",
        memory_type="agent-decision",
        metadata={
            "agent": agent_name,
            "task": task,
            "team_id": "marketing_squad",
            **(metadata or {})
        }
    )

# After researcher completes
store_agent_outcome(
    team_memory,
    agent_name="researcher",
    task="Analyze competitor pricing strategies",
    outcome="Found 3 competitors using value-based pricing. Key insight: price premiums tied to customer support tier.",
    metadata={"confidence": 0.9, "data_source": "web_scrape"}
)

# After analyst completes
store_agent_outcome(
    team_memory,
    agent_name="analyst",
    task="Segment customers by value",
    outcome="Identified 4 high-value segments. Enterprise segment has 3x retention vs SMB.",
    metadata={"model": "kmeans", "clusters": 4}
)

Step 4: Enable Agent-to-Agent Context Passing

When agents hand off work, pass the Recall context explicitly. Agent B receives not just the message, but the learned context that Agent A built.

from autogen import GroupChat, GroupChatManager

# Initialize group chat with all agents
group_chat = GroupChat(
    agents=[researcher, analyst],
    messages=[],
    max_round=10,
    speaker_selection_method="auto"
)

chat_manager = GroupChatManager(
    groupchat=group_chat,
    llm_config={"model": "gpt-4o"}
)

# Trigger conversation
chat_manager.initiate_chat(
    recipient=researcher,
    message="Analyze our market positioning. Use team insights from past research.",
    summary_method="reflection_with_llm"
)

# After conversation, capture team learning
for message in group_chat.messages:
    if "outcome" in message["content"].lower() or "insight" in message["content"].lower():
        store_agent_outcome(
            team_memory,
            agent_name=message["name"],
            task="group_discussion",
            outcome=message["content"],
            metadata={"conversation_id": group_chat.messages[0].get("id")}
        )

Step 5: Measure Coordination Improvement

Track how often agents reuse team memory vs. re-solve. When reuse increases, coordination improves.

def query_with_tracking(recall_instance, query, filters, agent_name):
    """Retrieve with metrics."""
    results = recall_instance.retrieve(
        query=query,
        filters=filters,
        top_k=5
    )
    
    # Log reuse
    if results:
        print(f"[{agent_name}] Reused {len(results)} team insights (hit)")
    else:
        print(f"[{agent_name}] No prior solutions found (miss) — will solve from scratch")
    
    return results

# Track over time
researcher_insights = query_with_tracking(
    team_memory,
    query="competitor pricing",
    filters={"team_id": "marketing_squad"},
    agent_name="analyst"
)

Code Comparison

Without Recall: Agents Work in Isolation

from autogen import ConversableAgent, GroupChat, GroupChatManager

# Create agents with no shared memory
researcher = ConversableAgent(
    name="researcher",
    system_message="You research market trends.",
    llm_config={"model": "gpt-4o"}
)

analyst = ConversableAgent(
    name="analyst",
    system_message="You analyze data.",
    llm_config={"model": "gpt-4o"}
)

# Group chat
group_chat = GroupChat(
    agents=[researcher, analyst],
    messages=[],
    max_round=10,
    speaker_selection_method="auto"
)

chat_manager = GroupChatManager(groupchat=group_chat, llm_config={"model": "gpt-4o"})

# Trigger conversation
chat_manager.initiate_chat(
    recipient=researcher,
    message="Analyze our market positioning and competitive landscape."
)

# Problem: If analyst asks researcher "What did you find about competitors?"
# Researcher re-searches and re-analyzes. No shared memory.
# Both agents burn tokens on redundant tasks.

With Recall: Shared Team Memory

from recall_sdk import Recall
from autogen import ConversableAgent, GroupChat, GroupChatManager

# 1. Create shared team memory
team_memory = Recall(
    name="marketing-team",
    metadata={"system": "multi-agent", "team": "marketing"}
)

# 2. Create agents
researcher = ConversableAgent(
    name="researcher",
    system_message="You research market trends. Check team memory first.",
    llm_config={"model": "gpt-4o"}
)

analyst = ConversableAgent(
    name="analyst",
    system_message="You analyze data. Use team insights from memory.",
    llm_config={"model": "gpt-4o"}
)

# 3. Wrap agents to use Recall
def agent_with_recall(agent, recall_instance):
    original_reply = agent.generate_reply
    
    def new_reply(messages, sender, **kwargs):
        # Retrieve team memory
        current_task = messages[-1]["content"] if messages else ""
        team_insights = recall_instance.retrieve(
            query=current_task,
            filters={"team": "marketing"},
            top_k=5
        )
        
        # Inject into context
        if team_insights:
            messages = messages.copy()
            context = "\n".join([m.text for m in team_insights])
            messages[0]["content"] = (
                f"Team has learned:\n{context}\n\n"
                + messages[0]["content"]
            )
        
        return original_reply(messages, sender, **kwargs)
    
    agent.generate_reply = new_reply
    return agent

researcher = agent_with_recall(researcher, team_memory)
analyst = agent_with_recall(analyst, team_memory)

# 4. Group chat
group_chat = GroupChat(
    agents=[researcher, analyst],
    messages=[],
    max_round=10,
    speaker_selection_method="auto"
)

chat_manager = GroupChatManager(groupchat=group_chat, llm_config={"model": "gpt-4o"})

# 5. Run and capture outcomes
chat_manager.initiate_chat(
    recipient=researcher,
    message="Analyze our market positioning and competitive landscape."
)

# 6. Store what was learned
for message in group_chat.messages:
    if any(word in message["content"].lower() for word in ["found", "insight", "discovered", "learned"]):
        team_memory.store(
            text=f"{message['name']}: {message['content']}",
            memory_type="agent-insight",
            metadata={
                "agent": message["name"],
                "team": "marketing"
            }
        )

# Next conversation: both agents retrieve team insights before responding.
# No redundant research. Agents build on each other's work.

Implementation Checklist

  • Install Recall SDK (pip install recall-sdk) and AutoGen
  • Create a single Recall instance shared by all agents in the team
  • Wrap agent methods to retrieve Recall context before acting
  • Store agent outcomes after task completion with team_memory.store()
  • Include agent name, task type, and team metadata in stored memories
  • Test a 2-task sequence: first task stores outcome, second task retrieves and builds on it
  • Track reuse metrics: how often agents retrieve vs. re-solve

Next Steps

Metric

50% improvement in multi-agent coordination. By sharing memory across agents, redundant work drops dramatically. Agent B reuses solutions Agent A discovered, cutting team token spend and decision latency by half. Agents learn from each other instead of rediscovering in parallel.

Related reading

Updates from the lab.

Engineering notes, research drops, occasional product updates. Roughly monthly.