Autogen API Overview

# Standard Library Imports.
import os
import io

# Third-Party Libraries.
import requests
import PIL.Image
import IPython.display

# Autogen Core & Utilities.
import autogen_core
from autogen_core.models import UserMessage
import autogen_core.tools as tools  # FunctionTool.

# Autogen Extensions / Models.
# For OpenAIChatCompletionClient.
import autogen_ext.models.openai as openai

# Autogen AgentChat Modules.
import autogen_agentchat
import autogen_agentchat.teams as teams

# For AssistantAgent.
import autogen_agentchat.agents as agent

# For Message and Event types.
import autogen_agentchat.messages as msg

# For TaskResult.
import autogen_agentchat.base as base
import autogen_agentchat.ui as ui
import autogen_agentchat.conditions as conditions

async def web_search(query: str) -> str:
    """
    Find information on the web.
    """
    return "AutoGen is a programming framework for building multi-agent applications."


assert os.environ.get("OPENAI_API_KEY", None) is not None

# Create an agent that uses the OpenAI GPT-4o model.
openai_model_client = openai.OpenAIChatCompletionClient(
    model="gpt-4.1-nano",
    api_key=os.environ["OPENAI_API_KEY"],
)
autogen_agentchat = agent.AssistantAgent(
    name="assistant",
    model_client=openai_model_client,
    tools=[web_search],
    system_message="Use tools to solve tasks.",
)

result = await openai_model_client.create(
    [UserMessage(content="What is the capital of France?", source="user")]
)
print(result)

# Create a multi-modal message with random image and text.
pil_image = PIL.Image.open(
    io.BytesIO(requests.get("https://picsum.photos/300/200").content)
)
img = autogen_core.Image(pil_image)
multi_modal_message = msg.MultiModalMessage(
    content=["Can you describe the content of this image?", img], source="user"
)
img

async def assistant_run_stream() -> None:
    # Option 1: read each message from the stream (as shown in the previous example).
    # async for message in agent.run_stream(task="Find information on AutoGen"):
    #     print(message)

    # Option 2: use Console to print all messages as they appear.
    await ui.Console(
        autogen_agentchat.run_stream(task="Find information on AutoGen"),
        output_stats=True,  # Enable stats printing.
    )


# Use asyncio.run(assistant_run_stream()) when running in a script.
await assistant_run_stream()

# Define a tool using a Python function.
async def web_search_func(query: str) -> str:
    """
    Find information on the web.
    """
    return "AutoGen is a programming framework for building multi-agent applications."


# This step is automatically performed inside the AssistantAgent if the tool is a Python function.
web_search_function_tool = tools.FunctionTool(
    web_search_func, description="Find information on the web"
)
# The schema is provided to the model during AssistantAgent's on_messages call.
web_search_function_tool.schema

model_client_no_parallel_tool_call = openai.OpenAIChatCompletionClient(
    model="gpt-4o",
    parallel_tool_calls=False,  # type: ignore
    api_key=os.environ["OPENAI_API_KEY"],
)
agent_no_parallel_tool_call = agent.AssistantAgent(
    name="assistant",
    model_client=model_client_no_parallel_tool_call,
    tools=[web_search],
    system_message="Use tools to solve tasks.",
)

agent_loop = agent.AssistantAgent(
    name="assistant_loop",
    model_client=model_client_no_parallel_tool_call,
    tools=[web_search],
    system_message="Use tools to solve tasks.",
    max_tool_iterations=10,  # At most 10 iterations of tool calls before stopping the loop.
)

streaming_assistant = agent.AssistantAgent(
    name="assistant",
    model_client=openai_model_client,
    system_message="You are a helpful assistant.",
    model_client_stream=True,  # Enable streaming tokens.
)

# Use an async function and asyncio.run() in a script.
async for message in streaming_assistant.run_stream(
    task="Name two cities in South America"
):  # type: ignore
    print(message)

# 1. Setup the Model Client (using existing openai_model_client from previous step).
# 2. Define Agent A: The Vision Specialist.
vision_agent = agent.AssistantAgent(
    name="Vision_Specialist",
    model_client=openai_model_client,
    system_message="Describe the image provided in detail.",
)
# 3. Define Agent B: The Reviewer.
reviewer_agent = agent.AssistantAgent(
    name="Reviewer",
    model_client=openai_model_client,
    system_message="Based on the description provided, suggest one creative use for this item.",
)
# 4. Create the Team (this enables true Agent-to-Agent communication).
agent_team = teams.RoundRobinGroupChat(
    [vision_agent, reviewer_agent], max_turns=2
)
# 5. Prepare the MultiModalMessage from your local file.
pil_image = PIL.Image.open("artifacts/camera.png")
img = autogen_core.Image(pil_image)
IPython.display.display(pil_image)
input_message = msg.MultiModalMessage(
    content=["Look at this camera feed and discuss it.", img], source="User"
)
# 6. Run the Team.
# The team will pass the MultiModalMessage from the User to Agent A,
# then Agent A's response to Agent B automatically.
async for message in agent_team.run_stream(task=input_message):
    # TaskResult objects don't have 'source', so we check if 'source' exists.
    if hasattr(message, "source"):
        print(f"\n--- Message from {message.source} ---")
        print(message.content)
    else:
        # This handles the final TaskResult object.
        print("\n--- Task Complete ---")
        # Optional: print(message.stop_reason)

async def internal_events():
    # 1. Setup the agent using the 'agent' alias.
    # model_client_stream=True is required to see the internal 'Thinking' events.
    solo_agent = agent.AssistantAgent(
        name="Solo_Assistant",
        model_client=openai_model_client,
        system_message="You are a helpful assistant.",
        model_client_stream=True,
    )
    print("--- Execution Started ---\n")
    # 2. Process the stream using the 'msg' and 'base' aliases.
    async for event in solo_agent.run_stream(
        task="What is the capital of France?"
    ):
        # Internal "Live" Streaming (individual tokens as they arrive).
        if isinstance(event, msg.ModelClientStreamingChunkEvent):
            if event.content:
                print(event.content, end="", flush=True)
        # Final Messages (completed response blocks).
        # We check specific types to avoid the generic ChatMessage TypeError.
        elif isinstance(
            event, (msg.TextMessage, msg.MultiModalMessage, msg.StopMessage)
        ):
            # Filter out the 'user' prompt to keep the output clean.
            if event.source != "user":
                print(f"\n\n[FINAL MESSAGE from {event.source}]:")
                print(event.content)
        # Final Summary using the 'base' alias.
        elif isinstance(event, base.TaskResult):
            print(f"\n\n--- Task Finished | Reason: {event.stop_reason} ---")


# Run the block.
await internal_events()

# Analyst system message (no TERMINATE here).
analyst = agent.AssistantAgent(
    name="Analyst",
    model_client=openai_model_client,
    system_message="""
You are a Senior Quantitative Analyst.
Answer clearly and concisely.
""",
)

# Reviewer system message.
reviewer = agent.AssistantAgent(
    name="Reviewer",
    model_client=openai_model_client,
    system_message="""
You are a peer reviewer.
Comment on the Analyst's output briefly.
""",
)

# Team termination condition (you can instead use a timeout or max_turns).
termination_condition = conditions.TextMentionTermination("TERMINATE")
team = teams.SelectorGroupChat(
    [analyst, reviewer],
    model_client=openai_model_client,
    termination_condition=termination_condition,
    max_turns=6,
)

# Task prompt.
task_prompt = "Summarize the hypothetical company 'ExampleCorp' in 2 sentences. Provide one key insight."
# Run.
async for message in team.run_stream(task=task_prompt):
    if hasattr(message, "source") and isinstance(message.content, str):
        print(f"{message.source}: {message.content.strip()}\n")
    elif hasattr(message, "stop_reason"):
        print(f"Mission Complete: {message.stop_reason}\n")

# Create the agents.
model_client = openai.OpenAIChatCompletionClient(
    model="gpt-4o-mini",
    api_key=os.environ["OPENAI_API_KEY"],
)
assistant = agent.AssistantAgent("assistant", model_client=model_client)
# Create the team setting a maximum number of turns to 1.
task = "Write a 4-line poem about the ocean."
while True:
    # Run the conversation and stream to the console.
    stream = team.run_stream(task=task)
    # Use asyncio.run(...) when running in a script.
    await ui.Console(stream)
    # Get the user response.
    task = input("Enter your feedback (type 'exit' to leave): ")
    if task.lower().strip() == "exit":
        break
await model_client.close()

# Create an OpenAI model client.
model_client = openai.OpenAIChatCompletionClient(
    model="gpt-4o",
    api_key=os.environ["OPENAI_API_KEY"],
)

# Create a lazy assistant agent that always hands off to the user.
lazy_agent = agent.AssistantAgent(
    "lazy_assistant",
    model_client=model_client,
    handoffs=[base.Handoff(target="user", message="Transfer to user.")],
    system_message="If you cannot complete the task, transfer to user. Otherwise, when finished, respond with 'TERMINATE'.",
)

# Define a termination condition that checks for handoff messages.
handoff_termination = conditions.HandoffTermination(target="user")
# Define a termination condition that checks for a specific text mention.
text_termination = conditions.TextMentionTermination("TERMINATE")

# Create a single-agent team with the lazy assistant and both termination conditions.
lazy_agent_team = teams.RoundRobinGroupChat(
    [lazy_agent], termination_condition=handoff_termination | text_termination
)

# Run the team and stream to the console.
task = "What is the weather in New York?"
await ui.Console(lazy_agent_team.run_stream(task=task), output_stats=True)

await ui.Console(
    lazy_agent_team.run_stream(task="The weather in New York is sunny.")
)

model_client = openai.OpenAIChatCompletionClient(
    model="gpt-4o",
    temperature=1,
    api_key=os.environ["OPENAI_API_KEY"],
)

# Create the primary agent.
primary_agent = agent.AssistantAgent(
    "primary",
    model_client=model_client,
    system_message="You are a helpful AI assistant.",
)

# Create the critic agent.
critic_agent = agent.AssistantAgent(
    "critic",
    model_client=model_client,
    system_message="Provide constructive feedback for every message. Respond with 'APPROVE' to when your feedbacks are addressed.",
)

max_msg_termination = conditions.MaxMessageTermination(max_messages=3)
round_robin_team = teams.RoundRobinGroupChat(
    [primary_agent, critic_agent], termination_condition=max_msg_termination
)

# Use asyncio.run(...) if you are running this script as a standalone script.
await ui.Console(
    round_robin_team.run_stream(
        task="Write a unique, Haiku about the weather in Paris"
    )
)

await ui.Console(round_robin_team.run_stream())

model_client = openai.OpenAIChatCompletionClient(
    model="gpt-4o-2024-08-06",
    api_key=os.environ["OPENAI_API_KEY"],
)

assistant_agent = agent.AssistantAgent(
    name="assistant_agent",
    system_message="You are a helpful assistant",
    model_client=model_client,
)

# Use asyncio.run(...) when running in a script.
response = await assistant_agent.on_messages(
    [
        msg.TextMessage(
            content="Write a 3 line poem on lake tangayika", source="user"
        )
    ],
    autogen_core.CancellationToken(),
)
print(response.chat_message)
await model_client.close()

agent_state = await assistant_agent.save_state()
print(agent_state)

model_client = openai.OpenAIChatCompletionClient(
    model="gpt-4o-2024-08-06",
    api_key=os.environ["OPENAI_API_KEY"],
)

# Define a team.
assistant_agent = agent.AssistantAgent(
    name="assistant_agent",
    system_message="You are a helpful assistant",
    model_client=model_client,
)
agent_team = teams.RoundRobinGroupChat(
    [assistant_agent],
    termination_condition=conditions.MaxMessageTermination(max_messages=2),
)

# Run the team and stream messages to the console.
stream = agent_team.run_stream(
    task="Write a beautiful poem 3-line about lake tangayika"
)

# Use asyncio.run(...) when running in a script.
await ui.Console(stream)

# Save the state of the agent team.
team_state = await agent_team.save_state()

await agent_team.reset()
stream = agent_team.run_stream(
    task="What was the last line of the poem you wrote?"
)
await ui.Console(stream)

print(team_state)

# Load team state.
await agent_team.load_state(team_state)
stream = agent_team.run_stream(
    task="What was the last line of the poem you wrote?"
)
await ui.Console(stream)