From raw OpenAI SDK to AutoGen: adding multi-agent support

Moving from the raw OpenAI SDK to AutoGen for multi-agent workflows isn’t a drop-in replacement — it’s an architectural shift. The openai sdk to autogen migration requires rethinking how you structure conversations, manage state, and delegate work across specialized agents. This guide walks through the transition with runnable code at each step so you can verify progress before moving on.

Step 1: Understand the architectural shift

The raw OpenAI SDK gives you a single completion endpoint. You send messages, get a response, and manage history yourself. AutoGen introduces agents that maintain their own system prompts, tool registries, and conversation state. Instead of one request-response loop, you orchestrate a group chat where agents speak in turns, call tools, and hand off to each other.

Key mental model changes:

• System prompts become agent definitions — each agent has a persistent personality and capability set
• Tool calling moves from ad-hoc to registered — functions are declared on agents, not passed per-request
• Conversation history is distributed — each agent sees a filtered view of the group chat
• Termination conditions replace max_tokens — you decide when the task is done, not the model

Verify your understanding: write down the three agents you’d need for a code-review workflow (e.g., Reviewer, Tester, Summarizer). If you can’t name them, pause and design the agent topology first.

Step 2: Set up AutoGen dependencies

Install the core package and the OpenAI model client. AutoGen 0.4+ uses a provider-agnostic client interface, so you’ll wire your existing OpenAI-compatible endpoint through OpenAIChatCompletionClient.

pip install "autogen-agentchat>=0.4" "autogen-ext[openai]>=0.4"

If you’re routing through a gateway like n4n.ai that serves 240+ models behind one OpenAI-compatible endpoint, point the client at your gateway URL and API key:

# config.py
import os
from autogen_ext.models.openai import OpenAIChatCompletionClient

def get_model_client(model: str = "gpt-4o-mini") -> OpenAIChatCompletionClient:
    return OpenAIChatCompletionClient(
        model=model,
        api_key=os.getenv("OPENAI_API_KEY"),
        base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"),
        # Optional: pass through provider-specific headers
        default_headers={"X-Request-Source": "autogen-migration"},
    )

Verify: run python -c "from config import get_model_client; print(get_model_client().model)" and confirm it prints your model name.

Step 3: Define agents and their roles

Replace your monolithic system prompt with discrete AssistantAgent instances. Each agent gets a name, system message, model client, and optional tool list.

# agents.py
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.base import ChatAgent
from config import get_model_client

def create_reviewer() -> AssistantAgent:
    return AssistantAgent(
        name="Reviewer",
        system_message=(
            "You are a senior code reviewer. Focus on correctness, security, "
            "and maintainability. Be specific — cite file paths and line numbers. "
            "When you find issues, categorize them: [BLOCKER], [MAJOR], [MINOR], [NIT]."
        ),
        model_client=get_model_client("gpt-4o"),
        description="Reviews code changes for quality and correctness",
    )

def create_tester() -> AssistantAgent:
    return AssistantAgent(
        name="Tester",
        system_message=(
            "You write and run tests. Given a code change, propose unit tests "
            "that cover the new logic. Use pytest. If tests fail, analyze the "
            "failure and suggest fixes. Never invent test results — run them."
        ),
        model_client=get_model_client("gpt-4o-mini"),
        description="Writes and executes tests for code changes",
    )

def create_summarizer() -> AssistantAgent:
    return AssistantAgent(
        name="Summarizer",
        system_message=(
            "You synthesize multi-agent discussions into a concise PR review summary. "
            "Include: overall verdict (APPROVE/REQUEST_CHANGES), top 3 issues, "
            "and action items for the author. Keep it under 200 words."
        ),
        model_client=get_model_client("gpt-4o-mini"),
        description="Produces final review summary",
    )

Verify: create a quick smoke test:

# test_agents.py
import asyncio
from agents import create_reviewer, create_tester, create_summarizer
from autogen_agentchat.messages import TextMessage

async def smoke_test():
    reviewer = create_reviewer()
    msg = TextMessage(content="Review this: def add(a, b): return a - b", source="user")
    response = await reviewer.on_messages([msg], cancellation_token=None)
    print(f"Reviewer: {response.chat_message.content[:200]}...")

asyncio.run(smoke_test())

Run it. You should see a structured review with categories. If the agent responds conversationally instead of categorizing, tighten the system message.

Step 4: Configure conversation patterns

AutoGen provides RoundRobinGroupChat for sequential turns and SelectorGroupChat for dynamic speaker selection. For code review, a fixed order works: Reviewer → Tester → Summarizer.

# workflow.py
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import MaxMessageTermination, TextMentionTermination
from agents import create_reviewer, create_tester, create_summarizer

def build_review_team() -> RoundRobinGroupChat:
    reviewer = create_reviewer()
    tester = create_tester()
    summarizer = create_summarizer()

    # Terminate after summarizer speaks OR after 10 messages total
    termination = TextMentionTermination("SUMMARY_COMPLETE") | MaxMessageTermination(10)

    return RoundRobinGroupChat(
        participants=[reviewer, tester, summarizer],
        termination_condition=termination,
    )

The TextMentionTermination lets the summarizer end the conversation by emitting a magic token. Update the summarizer’s system message to include: “End your response with SUMMARY_COMPLETE when done.”

Verify: run a full group chat on a trivial example:

# test_workflow.py
import asyncio
from autogen_agentchat.messages import TextMessage
from workflow import build_review_team

async def test_full_flow():
    team = build_review_team()
    task = TextMessage(
        content="Review this PR: changed `add(a, b)` to return `a - b` instead of `a + b`",
        source="user",
    )
    result = await team.run(task=task)
    print(f"Messages exchanged: {len(result.messages)}")
    for msg in result.messages:
        print(f"[{msg.source}]: {msg.content[:150]}...")

asyncio.run(test_full_flow())

You should see three distinct turns ending with the summarizer’s verdict. If the chat runs to MaxMessageTermination, your termination token isn’t being emitted — check the summarizer’s output.

Step 5: Handle tool calling and function execution

In the raw SDK, you pass tools per request. In AutoGen, tools are bound to agents at creation time. Define functions as plain Python callables, then register them.

# tools.py
import subprocess
import json
from pathlib import Path
from typing import Annotated
from autogen_core.tools import FunctionTool

def run_pytest(
    test_path: Annotated[str, "Path to test file or directory"],
    extra_args: Annotated[str, "Additional pytest args"] = "-xvs",
) -> str:
    """Run pytest and return stdout/stderr."""
    result = subprocess.run(
        ["pytest", test_path, *extra_args.split()],
        capture_output=True,
        text=True,
        timeout=120,
    )
    return json.dumps({
        "returncode": result.returncode,
        "stdout": result.stdout[-5000:],  # truncate
        "stderr": result.stderr[-2000:],
    })

def read_file(
    path: Annotated[str, "File path relative to repo root"],
) -> str:
    """Read a file's contents."""
    try:
        return Path(path).read_text()
    except Exception as e:
        return f"ERROR: {e}"

# Wrap for AutoGen
run_pytest_tool = FunctionTool(run_pytest, description="Execute pytest on a test file")
read_file_tool = FunctionTool(read_file, description="Read a source file")

Attach tools to the agents that need them:

# agents.py (updated)
from autogen_agentchat.agents import AssistantAgent
from tools import run_pytest_tool, read_file_tool

def create_tester() -> AssistantAgent:
    return AssistantAgent(
        name="Tester",
        system_message=(
            "You write and run tests. Given a code change, propose unit tests "
            "that cover the new logic. Use pytest. If tests fail, analyze the "
            "failure and suggest fixes. Never invent test results — run them.\n\n"
            "Available tools: run_pytest, read_file"
        ),
        model_client=get_model_client("gpt-4o-mini"),
        tools=[run_pytest_tool, read_file_tool],
        description="Writes and executes tests for code changes",
    )

Verify: give the tester a real file to test. Create a temp module:

# /tmp/sample/math.py
def add(a: int, b: int) -> int:
    return a - b  # intentional bug

# /tmp/sample/test_math.py
import pytest
from math import add

def test_add():
    assert add(2, 3) == 5

Then test the tool binding:

# test_tools.py
import asyncio
from agents import create_tester
from autogen_agentchat.messages import TextMessage

async def test_tool_use():
    tester = create_tester()
    msg = TextMessage(
        content="Write a test for /tmp/sample/math.py and run it. The add function is buggy.",
        source="user",
    )
    response = await tester.on_messages([msg], cancellation_token=None)
    print(response.chat_message.content)

asyncio.run(test_tool_use())

You should see the agent call read_file, then run_pytest, then analyze the failure. If it skips tool calls, ensure the system message explicitly instructs tool use and the model supports function calling.

Step 6: Add memory and state management

The raw SDK leaves history management to you. AutoGen agents maintain internal state, but for long-running workflows you need external persistence — especially if the process restarts between turns.

Use ChatHistory with a custom store. Here’s a minimal JSONL backend:

# memory.py
import json
from pathlib import Path
from typing import AsyncIterator
from autogen_agentchat.base import ChatAgent
from autogen_agentchat.messages import BaseMessage, TextMessage
from autogen_core import CancellationToken

class JSONLHistory:
    def __init__(self, path: Path):
        self.path = path
        self.path.parent.mkdir(parents=True, exist_ok=True)

    async def append(self, message: BaseMessage) -> None:
        record = message.model_dump()
        with self.path.open("a") as f:
            f.write(json.dumps(record) + "\n")

    async def get_all(self) -> list[BaseMessage]:
        if not self.path.exists():
            return []
        messages = []
        with self.path.open() as f:
            for line in f:
                data = json.loads(line)
                messages.append(TextMessage(**data))  # simplify; use real deserialization
        return messages

    async def clear(self) -> None:
        if self.path.exists():
            self.path.unlink()

Wire it into the team by wrapping agents with a history-aware delegate, or simpler: persist the team’s run() output after each turn. For production, consider autogen-ext’s built-in SQLiteChatHistory or Redis.

Verify: run the workflow twice with the same history file and confirm the second run sees prior context.

# test_memory.py
import asyncio
from workflow import build_review_team
from memory import JSONLHistory
from autogen_agentchat.messages import TextMessage

async def test_persistence():
    history = JSONLHistory(Path("/tmp/review_history.jsonl"))
    team = build_review_team()

    # First run
    await team.run(task=TextMessage(content="Review: def add(a,b): return a-b", source="user"))

    # Second run — should reference first review
    await team.run(task=TextMessage(content="Any follow-up on the previous review?", source="user"))

    # Check history file
    messages = await history.get_all()
    print(f"Total messages persisted: {len(messages)}")

asyncio.run(test_persistence())

Step 7: Test and verify the migration

Create a golden dataset of 5-10 representative tasks from your production workload. For each, capture the raw SDK output (your baseline) and the AutoGen team output. Compare on:

• Task completion: does the multi-agent flow reach a valid conclusion?
• Token usage: sum across all agents vs. single SDK call
• Latency: wall-clock time for the full group chat
• Quality: human eval on a 1-5 scale for correctness and usefulness

# eval.py
import asyncio
import time
from dataclasses import dataclass
from workflow import build_review_team
from autogen_agentchat.messages import TextMessage

@dataclass
class EvalCase:
    name: str
    input: str
    expected_keywords: list[str]  # must appear in final summary

CASES = [
    EvalCase(
        name="buggy_add",
        input="Review: def add(a, b): return a - b",
        expected_keywords=["BLOCKER", "incorrect", "subtraction"],
    ),
    EvalCase(
        name="missing_docstring",
        input="Review: def process(data): return data.strip()",
        expected_keywords=["MINOR", "docstring"],
    ),
    # add 3-8 more from real PRs
]

async def run_eval():
    team = build_review_team()
    results = []

    for case in CASES:
        start = time.perf_counter()
        result = await team.run(task=TextMessage(content=case.input, source="user"))
        elapsed = time.perf_counter() - start

        final_msg = result.messages[-1].content if result.messages else ""
        passed = all(kw.lower() in final_msg.lower() for kw in case.expected_keywords)
        total_tokens = sum(
            m.models_usage.total_tokens for m in result.messages if m.models_usage
        )

        results.append({
            "case": case.name,
            "passed": passed,
            "latency_s": round(elapsed, 2),
            "tokens": total_tokens,
            "summary": final_msg[:300],
        })
        print(f"{case.name}: {'PASS' if passed else 'FAIL'} | {elapsed:.1f}s | {total_tokens} tok")

    return results

asyncio.run(run_eval())

Set a threshold: all cases must pass before cutting over. If latency is unacceptable, consider:

• Using smaller models for Tester/Summarizer (already done above)
• Parallelizing independent agents with SelectorGroupChat
• Caching repeated file reads

Step 8: Production considerations

The migration isn’t done until you’ve addressed observability, cost control, and failure modes.

Observability: wrap the team run with structured logging. Emit one log line per agent turn with agent, tokens_in, tokens_out, tools_called, duration_ms. This lets you build dashboards for cost per review, bottleneck detection, and error rates.

Cost control: set a per-run token budget. AutoGen doesn’t enforce this natively — add a termination condition that checks cumulative usage:

# cost_guard.py
from autogen_agentchat.conditions import TerminationCondition
from autogen_agentchat.messages import BaseMessage

class TokenBudgetTermination(TerminationCondition):
    def __init__(self, max_tokens: int):
        self.max_tokens = max_tokens
        self.used = 0

    async def should_terminate(self, messages: list[BaseMessage]) -> bool:
        for m in messages:
            if m.models_usage:
                self.used += m.models_usage.total_tokens
        return self.used >= self.max_tokens

Add it to your team: termination = TokenBudgetTermination(50000) | TextMentionTermination("SUMMARY_COMPLETE").

Failure modes: agents can loop, hallucinate tool calls, or emit malformed JSON. Implement a supervisor agent that watches for:

• More than 3 consecutive tool errors from the same agent
• Repetitive content (cosine similarity > 0.95 between last 2 messages)
• Exceeding per-agent turn limits

The supervisor can inject a corrective message or terminate the run.

Rollout strategy: shadow mode first. Run both the raw SDK path and AutoGen path on production traffic, compare outputs, and only switch when eval passes and cost/latency are within 20% of baseline.

The openai sdk to autogen migration pays off when you need specialization (different prompts per role), tool isolation (reviewer doesn’t need pytest), and auditable reasoning (each agent’s contribution is traceable). For single-turn completions, the raw SDK remains simpler. Choose the abstraction that matches your workflow’s complexity.