AutoGen：微软的多 Agent 对话框架

import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import MaxMessageTermination
from autogen_ext.models.openai import OpenAIChatCompletionClient

async def main():
    model_client = OpenAIChatCompletionClient(model="gpt-4o-mini")

    # 助手 Agent
    assistant = AssistantAgent(
        name="Assistant",
        model_client=model_client,
        system_message="你是一个有帮助的助手，用中文回答。",
    )

    # 批评者 Agent
    critic = AssistantAgent(
        name="Critic",
        model_client=model_client,
        system_message="你是一个批评者，检查助手的回答是否准确和完整，指出改进点。",
    )

    # 组成团队，轮流发言
    team = RoundRobinGroupChat(
        participants=[assistant, critic],
        termination_condition=MaxMessageTermination(max_messages=4),
    )

    # 运行
    result = await team.run(task="解释什么是 RAG（检索增强生成）")

    for message in result.messages:
        print(f"\n[{message.source}]")
        print(message.content)

asyncio.run(main())

import asyncio
from autogen_agentchat.agents import AssistantAgent, CodeExecutorAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_ext.code_executors.local import LocalCommandLineCodeExecutor

async def main():
    model_client = OpenAIChatCompletionClient(model="gpt-4o")

    # 代码编写者
    coder = AssistantAgent(
        name="Coder",
        model_client=model_client,
        system_message="""你是一个 Python 程序员。
编写代码时用 ```python 代码块包裹，只写代码，不加解释。
代码解决问题后回复 TERMINATE。""",
    )

    # 代码执行者（在本地沙箱执行）
    executor = CodeExecutorAgent(
        name="Executor",
        code_executor=LocalCommandLineCodeExecutor(work_dir="./tmp"),
    )

    team = RoundRobinGroupChat(
        participants=[coder, executor],
        termination_condition=TextMentionTermination("TERMINATE"),
    )

    result = await team.run(
        task="计算前 20 个斐波那契数，并用 matplotlib 画出折线图，保存为 fib.png"
    )

    for msg in result.messages:
        print(f"\n[{msg.source}]: {msg.content[:200]}")

asyncio.run(main())

from autogen_agentchat.agents import AssistantAgent
from autogen_core.tools import FunctionTool
from autogen_ext.models.openai import OpenAIChatCompletionClient

def get_weather(city: str) -> str:
    """获取城市天气"""
    data = {"北京": "晴天 25°C", "上海": "多云 22°C"}
    return data.get(city, "暂无数据")

def search_web(query: str) -> str:
    """搜索网络"""
    return f"关于 '{query}' 的搜索结果（模拟）"

# 包装成 FunctionTool
tools = [
    FunctionTool(get_weather, description="获取城市天气"),
    FunctionTool(search_web, description="搜索网络信息"),
]

agent = AssistantAgent(
    name="ToolAgent",
    model_client=OpenAIChatCompletionClient(model="gpt-4o-mini"),
    tools=tools,
    system_message="你可以查天气和搜索信息来回答用户问题。",
)

import asyncio
async def main():
    result = await agent.run(task="北京天气怎样？同时搜索一下 AutoGen 是什么")
    print(result.messages[-1].content)

asyncio.run(main())

from autogen_agentchat.teams import SelectorGroupChat
from autogen_agentchat.conditions import MaxMessageTermination

# 多个专家 Agent
planner = AssistantAgent(
    name="Planner",
    model_client=model_client,
    system_message="你是规划专家，负责把任务拆分成子任务。",
)

researcher = AssistantAgent(
    name="Researcher",
    model_client=model_client,
    system_message="你是研究专家，负责收集信息。",
)

writer = AssistantAgent(
    name="Writer",
    model_client=model_client,
    system_message="你是写作专家，负责整合信息生成报告。",
)

# SelectorGroupChat 让 LLM 决定下一步该谁说话
team = SelectorGroupChat(
    participants=[planner, researcher, writer],
    model_client=model_client,  # 用这个模型决定发言顺序
    termination_condition=MaxMessageTermination(max_messages=10),
)

async def main():
    result = await team.run(task="写一份关于 LangGraph 的技术分析报告")
    print(result.messages[-1].content)

asyncio.run(main())

from autogen_agentchat.agents import UserProxyAgent

# UserProxyAgent 在关键步骤请求人类确认
human = UserProxyAgent(
    name="Human",
    input_func=input,  # 从终端读取输入
)

assistant = AssistantAgent(
    name="Assistant",
    model_client=model_client,
    system_message="你是助手，需要人类审批后才能执行操作。",
)

team = RoundRobinGroupChat(
    participants=[assistant, human],
    termination_condition=TextMentionTermination("APPROVED"),
)

# Agent 会生成方案，等待人类输入 "APPROVED" 确认

async def stream_example():
    async for message in team.run_stream(task="写一首诗"):
        if hasattr(message, "content"):
            print(f"[{message.source}]: ", end="")
            print(message.content)