Docs

Documentation

AutoGen + ModelRiver

Multi-agent conversations with automatic provider failover. Route every AutoGen agent through ModelRiver for resilience and observability.

Overview

AutoGen is Microsoft's framework for building multi-agent conversational systems. Agents talk to each other, use tools, and solve complex problems through structured dialogue. By routing through ModelRiver, every agent message gets failover, cost tracking, and structured output guarantees.

What you get:

  • Every agent conversation turn routes through ModelRiver
  • Automatic failover during multi-turn conversations
  • Cost tracking for each agent in the topology
  • Provider switching from the console without code changes

Quick start

Install dependencies

Bash
pip install autogen-agentchat autogen-ext[openai]

Connect AutoGen to ModelRiver

PYTHON
1from autogen_ext.models.openai import OpenAIChatCompletionClient
2 
3model_client = OpenAIChatCompletionClient(
4 model="my-chat-workflow",
5 base_url="https://api.modelriver.com/v1",
6 api_key="mr_live_YOUR_API_KEY",
7)

Two-agent conversation

PYTHON
1import asyncio
2from autogen_agentchat.agents import AssistantAgent
3from autogen_agentchat.conditions import TextMentionTermination
4from autogen_agentchat.teams import RoundRobinGroupChat
5 
6model_client = OpenAIChatCompletionClient(
7 model="my-chat-workflow",
8 base_url="https://api.modelriver.com/v1",
9 api_key="mr_live_YOUR_API_KEY",
10)
11 
12# Define agents
13coder = AssistantAgent(
14 name="Coder",
15 system_message="You write clean, efficient Python code. When done, say TERMINATE.",
16 model_client=model_client,
17)
18 
19reviewer = AssistantAgent(
20 name="Reviewer",
21 system_message="You review code for bugs, security, and performance. When satisfied, say TERMINATE.",
22 model_client=model_client,
23)
24 
25# Create a team
26termination = TextMentionTermination("TERMINATE")
27team = RoundRobinGroupChat(
28 participants=[coder, reviewer],
29 termination_condition=termination,
30 max_turns=6,
31)
32 
33async def main():
34 result = await team.run(task="Write a Python function to validate email addresses using regex.")
35 for msg in result.messages:
36 print(f"[{msg.source}]: {msg.content[:200]}")
37 
38asyncio.run(main())

Per-agent model routing

Give each agent a different ModelRiver workflow:

PYTHON
1# Fast model for the coder (lots of back-and-forth)
2coder_client = OpenAIChatCompletionClient(
3 model="fast-coder", # GPT-4o-mini workflow
4 base_url="https://api.modelriver.com/v1",
5 api_key="mr_live_YOUR_API_KEY",
6)
7 
8# Powerful model for the reviewer (reasoning matters)
9reviewer_client = OpenAIChatCompletionClient(
10 model="strict-reviewer", # GPT-4o / Claude 3.5 workflow
11 base_url="https://api.modelriver.com/v1",
12 api_key="mr_live_YOUR_API_KEY",
13)
14 
15coder = AssistantAgent(name="Coder", system_message="...", model_client=coder_client)
16reviewer = AssistantAgent(name="Reviewer", system_message="...", model_client=reviewer_client)

Agents with tools

PYTHON
1from autogen_core import FunctionCall
2 
3def get_weather(location: str) -> str:
4 """Get current weather for a location."""
5 return f"22°C and sunny in {location}"
6 
7def search_database(query: str) -> str:
8 """Search the internal knowledge base."""
9 return f"Found 3 results for: {query}"
10 
11assistant = AssistantAgent(
12 name="Assistant",
13 system_message="You help users with questions. Use tools when needed.",
14 model_client=model_client,
15 tools=[get_weather, search_database],
16)

Group chat with selector

Use a selector to dynamically choose which agent speaks next:

PYTHON
1from autogen_agentchat.teams import SelectorGroupChat
2 
3researcher = AssistantAgent(name="Researcher", system_message="...", model_client=model_client)
4analyst = AssistantAgent(name="Analyst", system_message="...", model_client=model_client)
5writer = AssistantAgent(name="Writer", system_message="...", model_client=model_client)
6 
7team = SelectorGroupChat(
8 participants=[researcher, analyst, writer],
9 model_client=model_client, # Selector also routes through ModelRiver
10 termination_condition=termination,
11 max_turns=10,
12)
13 
14result = await team.run(task="Analyse the AI market and write a report")

Streaming conversations

PYTHON
1async def main():
2 stream = team.run_stream(task="Write a Python web scraper")
3 async for message in stream:
4 if hasattr(message, "content"):
5 print(f"[{message.source}]: {message.content[:200]}")
6 
7asyncio.run(main())

Best practices

  1. Use different workflows per agent: Match model capability to agent role
  2. Set max_turns: Prevent runaway conversations that generate excessive costs
  3. Monitor turn costs: Check Request Logs for per-turn token usage
  4. Configure fallbacks: Multi-turn conversations are especially vulnerable to provider outages
  5. Use structured outputs: Define output schemas for agents that need to produce data

Next steps