ag2

Important

Looking for ConversableAgent, GroupChat, or import autogen? That's now AG2 Classic.

As of AG2 v1.0, the protocol-driven framework is the top-level package, imported as ag2. The classic framework has moved to its own repository — ag2ai/ag2-classic, documented at classic.docs.ag2.ai. It is still maintained and installable; nothing you have built stops working.

This repository (pip install ag2) no longer ships the autogen import name or the classic agent classes.

AG2: Open-Source AgentOS for AI Agents

Table of contents

AG2 Classic (the autogen.* namespace)

Are you using AG2 Classic?

import autogen  # the autogen.* namespace
from autogen import ConversableAgent, GroupChat  # classic agent classes
from autogen import AssistantAgent, UserProxyAgent

pip install ag2-classic

Note

AG2 v1.0 (pip install ag2) is not a drop-in upgrade from Classic. The agent model, orchestration, and imports all changed. See the group chat migration guide.

Getting started

Installation

pip install ag2[openai]

pip install 'ag2[openai]'

Setup your API keys

export OPENAI_API_KEY="<your-api-key>"     # or ANTHROPIC_API_KEY, GEMINI_API_KEY, ...

Run your first agent

import asyncio

from ag2 import Agent
from ag2.config import OpenAIConfig

agent = Agent(
    "assistant",
    prompt="You are a helpful assistant.",
    config=OpenAIConfig(model="gpt-4o-mini"),
)


async def main() -> None:
    reply = await agent.ask("Summarize the main differences between Python lists and tuples.")
    print(reply.body)


asyncio.run(main())

Example applications

Introduction of different agent concepts

Agents

import asyncio

from ag2 import Agent
from ag2.config import OpenAIConfig

reviewer = Agent(
    "reviewer",
    prompt=(
        "You are a code reviewer. Analyze the provided code and suggest improvements. "
        "Do not generate code, only suggest improvements."
    ),
    config=OpenAIConfig(model="gpt-4o-mini"),
)


async def main() -> None:
    reply = await reviewer.ask("def fib(n): return n if n < 2 else fib(n-1) + fib(n-2)")
    print(reply.body)

    # Continue the same conversation — prior turns stay in scope.
    follow_up = await reply.ask("Which of those matters most for large n?")
    print(follow_up.body)


asyncio.run(main())

Tools

import asyncio
from datetime import datetime

from ag2 import Agent, tool
from ag2.config import OpenAIConfig


@tool
async def get_weekday(date_string: str) -> str:
    """Get the day of the week for a given date, formatted as YYYY-MM-DD."""
    return datetime.strptime(date_string, "%Y-%m-%d").strftime("%A")


date_agent = Agent(
    "date_agent",
    prompt="You find the day of the week for a given date.",
    config=OpenAIConfig(model="gpt-4o-mini"),
    tools=[get_weekday],
)


async def main() -> None:
    reply = await date_agent.ask("I was born on 1995-03-25, what day was it?")
    print(reply.body)


asyncio.run(main())

Human in the Loop

import asyncio

from ag2 import Agent, Context, tool
from ag2.config import OpenAIConfig
from ag2.events import HumanInputRequest, HumanMessage


@tool
async def publish_lesson_plan(context: Context, plan: str) -> str:
    """Publish a lesson plan, once a human educator has approved it."""
    answer = await context.input(f"Approve this lesson plan?\n\n{plan}")
    if answer.strip().lower().startswith("y"):
        return "Published."
    return f"Rejected by the educator: {answer}"


def hitl_hook(event: HumanInputRequest) -> HumanMessage:
    # Block here for real input — a CLI prompt, a web UI, a queue.
    return HumanMessage(content=input(f"{event.content}\n> "))


teacher = Agent(
    "teacher",
    prompt=(
        "Draft a short lesson plan, then call the publish_lesson_plan tool to have it published. "
        "Approval is collected by the tool — never ask for approval in plain text."
    ),
    config=OpenAIConfig(model="gpt-4o-mini"),
    tools=[publish_lesson_plan],
    hitl_hook=hitl_hook,
)


async def main() -> None:
    reply = await teacher.ask("Let's introduce our kids to the solar system.")
    print(reply.body)


asyncio.run(main())

Orchestrating Multiple Agents

import asyncio

from ag2 import Agent
from ag2.config import OpenAIConfig
from ag2.knowledge import MemoryKnowledgeStore
from ag2.network import EV_TEXT, Hub

MAX_MESSAGES = 4


async def wait_for_messages(hub: Hub, channel_id: str, expected: int, timeout: float = 120.0) -> None:
    """Poll the hub's write-ahead log until `expected` messages have been exchanged."""
    deadline = asyncio.get_event_loop().time() + timeout
    while asyncio.get_event_loop().time() < deadline:
        wal = await hub.read_wal(channel_id)
        if sum(1 for e in wal if e.event_type == EV_TEXT) >= expected:
            return
        await asyncio.sleep(0.05)
    raise TimeoutError(f"only reached {expected} messages before timing out")


async def main() -> None:
    config = OpenAIConfig(model="gpt-4o-mini")
    hub = await Hub.open(MemoryKnowledgeStore(), ttl_sweep_interval=0)

    planner = await hub.register(
        Agent("planner", prompt="You plan school lessons. Reply in one short sentence.", config=config),
    )
    reviewer = await hub.register(
        Agent("reviewer", prompt="You critique lesson plans. Reply in one short sentence.", config=config),
    )

    # Free-form 2-party channel: either side may speak at any time.
    channel = await planner.open(type="conversation", target="reviewer")
    await channel.send("What should a fourth-grade solar system lesson lead with?", audience=[reviewer.agent_id])

    # Both default handlers reply to every inbound message, so the conversation
    # auto-drives. `conversation` never auto-closes — we bound it and close it.
    await wait_for_messages(hub, channel.channel_id, expected=MAX_MESSAGES)
    await channel.close()

    # Replay the exchange from the hub's write-ahead log.
    names = {planner.agent_id: "planner", reviewer.agent_id: "reviewer"}
    for envelope in await hub.read_wal(channel.channel_id):
        if envelope.event_type == EV_TEXT:
            print(f"{names[envelope.sender_id]}: {envelope.event_data['text']}")

    await hub.close()


asyncio.run(main())

The agent harness: knowledge and compaction

import asyncio
from pathlib import Path

from ag2 import Agent, KnowledgeConfig, MemoryStream
from ag2.compact import CompactTrigger, SummarizeCompact
from ag2.config import OpenAIConfig
from ag2.events import CompactionCompleted, CompactionFailed
from ag2.knowledge import DiskKnowledgeStore
from ag2.policies import WorkingMemoryPolicy

config = OpenAIConfig(model="gpt-4o-mini")

# Disk-backed, so anything the agent remembers survives the process exiting.
STORE_DIR = Path("./knowledge_demo")
store = DiskKnowledgeStore(STORE_DIR)

agent = Agent(
    "tutor",
    prompt=(
        "You are a science tutor for a fourth-grade teacher. "
        "Whenever you learn a durable fact about the teacher or their class, use the knowledge "
        "tool to `read` `memory/working.md`, then `write` it back with the new fact appended as "
        "a bullet. Never drop a fact that is already there. "
        "Keep every reply to one or two sentences."
    ),
    config=config,
    knowledge=KnowledgeConfig(
        store=store,
        # Summarize the dropped history instead of forgetting it outright.
        compact=SummarizeCompact(target=4, config=config),
        compact_trigger=CompactTrigger(max_events=8),
    ),
    # Injects memory/working.md into the system prompt on every turn, so recall
    # does not depend on the model choosing to call the knowledge tool.
    assembly=[WorkingMemoryPolicy()],
)

stream = MemoryStream()


# Subscribe to both outcomes — watching only Completed would hide a failing strategy.
@stream.where(CompactionCompleted).subscribe()
async def on_compacted(event: CompactionCompleted) -> None:
    print(f"  [compacted: {event.events_before} -> {event.events_after} events via {event.strategy}]")


@stream.where(CompactionFailed).subscribe()
async def on_compaction_failed(event: CompactionFailed) -> None:
    print(f"  [compaction FAILED: {event.strategy}]")


async def main() -> None:
    # A returning session: no conversation history at all, so anything the agent
    # recalls here came off disk, via the knowledge store.
    if STORE_DIR.exists():
        reply = await agent.ask("What do you remember about me and my class?", stream=stream)
        print(f"tutor: {reply.body}")
        return

    reply = await agent.ask("I'm Dana. I teach 4th grade at Rosewood Elementary, 26 students.", stream=stream)
    print(f"tutor: {reply.body}")

    for turn in ["My class struggles most with why we have seasons.", "Suggest one hands-on demo for that."]:
        reply = await reply.ask(turn)
        print(f"tutor: {reply.body}")


asyncio.run(main())

  [compacted: 18 -> 3 events via SummarizeCompact]
tutor: I remember that you, Dana, teach 4th grade at Rosewood Elementary with 26 students,
       and your class struggles the most with understanding why we have seasons.

Advanced agentic design patterns

Code style and linting

pip install prek
prek install

prek run --all-files

Contributors Wall

License