微软AutoGen:多智能体AI开发新利器
AutoGen 框架核心功能与实战示例 框架概述 微软 AutoGen 是一个多智能体协作开发框架,支持: Python 和 .NET 双生态 智能体间自主通信与任务协调 工具调用与代码执行扩展 可视化开发工具 AutoGen Studio 核心实战场景 1. 代码执行工具 executor = LocalCommandLineCodeExecutor(work_dir="coding") too
·
https://github.com/microsoft/autogen
这个仓库是微软的 AutoGen 框架的官方代码库,主要用于构建多智能体 AI 应用程序。以下是对该仓库的详细介绍:
1. 核心定位
AutoGen 是一个支持多智能体协作的框架,允许智能体自主行动或与人类协同工作,适用于构建复杂的 AI 应用。其核心特点包括:
- 支持多智能体间的通信与协调
- 兼容多种 AI 模型(如 OpenAI 的 GPT 系列)
- 提供工具调用、代码执行等扩展能力
- 支持无代码 GUI 工具(AutoGen Studio)
2. 主要组件与功能
- 多智能体协作:提供 AgentChat 模块,支持智能体通过共享上下文或定向工作流(如 GraphFlow)协同完成任务。
- 模型集成:通过
autogen-ext
扩展支持 OpenAI、Azure 等多种模型服务。 - 工具与工作台:例如 MCP Workbench 支持网页内容获取等功能,方便智能体调用外部工具。
- AutoGen Studio:可视化工具,支持通过拖拽或 JSON 配置创建智能体团队,提供交互式测试环境(Playground)和社区组件库(Gallery)。
3. 支持的语言与生态
- Python:提供
autogen-core
、autogen-agentchat
等 PyPI 包,支持 Python 3.10+。 - .NET:提供 NuGet 包(如
Microsoft.AutoGen.Contracts
),支持 .NET 生态的开发。 - 社区扩展:包含第三方项目(如
autogen-oaiapi
、autogen-contextplus
),扩展了框架的功能。
4. 仓库结构
- 根目录包含文档(
README.md
、FAQ.md
)、配置文件(.gitignore
、codecov.yml
)等。 python/
:Python 相关代码,包括核心库、扩展(autogen-ext
)、示例(samples/
)和 AutoGen Studio。dotnet/
:.NET 相关代码,包含 SDK、示例(如dev-team
演示 GitHub 协作智能体)。docs/
:文档和用户指南,涵盖安装、快速启动、多智能体设计模式等。protos/
:定义了智能体通信的协议文件(如agent_worker.proto
)。
5. 安装与使用
- Python 核心库安装:
pip install -U "autogen-agentchat" "autogen-ext[openai]"
- AutoGen Studio(无代码界面)安装:
pip install -U "autogenstudio"
- 示例代码涵盖基础交互(如 “Hello World”)、网页浏览、多智能体编排等场景。
AutoGen Python 代码核心实战示例
涵盖多智能体协作、代码执行、工具调用等关键场景,均来自官方示例并附加解析:
一、代码执行工具(PythonCodeExecutionTool)
场景:智能体调用工具执行 Python 代码并返回结果
核心代码(来自 autogen_ext/tools/code_execution/_code_execution.py
示例):
import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_ext.code_executors.local import LocalCommandLineCodeExecutor
from autogen_ext.tools.code_execution import PythonCodeExecutionTool
async def main() -> None:
# 1. 初始化代码执行器(本地命令行环境)
executor = LocalCommandLineCodeExecutor(work_dir="coding") # 代码在 "coding" 目录执行
# 2. 创建代码执行工具
tool = PythonCodeExecutionTool(executor=executor)
# 3. 初始化智能体(使用 GPT-4o 模型)
model_client = OpenAIChatCompletionClient(model="gpt-4o")
agent = AssistantAgent(
name="code_assistant",
model_client=model_client,
tools=[tool], # 绑定工具
reflect_on_tool_use=True # 自动反思工具调用结果
)
# 4. 执行任务:计算斐波那契数列第 10 项
task = "计算斐波那契数列的第 10 项,用 Python 代码实现"
result = await agent.run(task=task)
print("执行结果:", result)
asyncio.run(main())
关键解析:
- 代码执行器:
LocalCommandLineCodeExecutor
负责在本地环境运行代码,支持隔离目录(work_dir
)避免冲突。 - 工具绑定:智能体通过
tools
参数关联PythonCodeExecutionTool
,可自动判断何时需要执行代码。 - 反思机制:
reflect_on_tool_use=True
让智能体在工具调用后检查结果,若出错会自动修正代码重试。
二、多智能体协作(国际象棋游戏)
场景:AI 智能体与人类(或随机策略)对战国际象棋
核心代码(来自 samples/agentchat_chess_game/main.py
):
import asyncio
import chess
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.ui import Console
from autogen_core.models import ChatCompletionClient
def create_ai_player(model_client: ChatCompletionClient) -> AssistantAgent:
"""创建 AI 棋手智能体"""
return AssistantAgent(
name="ai_player",
model_client=model_client,
system_message="你是国际象棋大师,用 UCI 格式返回走法(如 e2e4)",
model_context=BufferedChatCompletionContext(buffer_size=10), # 限制上下文为最近 10 条消息
)
async def get_ai_move(board: chess.Board, player: AssistantAgent) -> str:
"""获取 AI 的下一步走法"""
# 生成提示:包含当前棋盘状态、合法走法
legal_moves = ", ".join([move.uci() for move in board.legal_moves])
prompt = f"棋盘状态: {board.fen()}\n合法走法: {legal_moves}\n请用 UCI 格式返回你的走法"
# 流式获取智能体响应
result = await Console(player.run_stream(task=prompt))
# 从响应中提取走法(格式:<move>e2e4</move>)
response = result.messages[-1].content
return response.split("<move>")[1].split("</move>")[0]
async def main():
# 初始化棋盘和模型客户端
board = chess.Board()
model_client = ChatCompletionClient.load_component("model_config.yaml") # 从配置文件加载模型
ai_player = create_ai_player(model_client)
# 游戏主循环
while not board.is_game_over():
# AI 走棋
ai_move = await get_ai_move(board, ai_player)
board.push(chess.Move.from_uci(ai_move))
print(f"AI 走法: {ai_move}")
# 人类走棋(或随机走法)
user_move = input("你的走法 (UCI 格式): ")
board.push(chess.Move.from_uci(user_move))
print(f"游戏结束!结果: {board.result()}")
asyncio.run(main())
关键解析:
- 智能体角色定义:通过
system_message
约束 AI 棋手的行为(仅返回 UCI 格式走法)。 - 上下文管理:
BufferedChatCompletionContext
限制上下文长度,避免模型输入超限。 - 流式交互:
run_stream
配合Console
实现实时输出,提升交互体验。
三、任务中心记忆(Task-Centric Memory)
场景:智能体通过记忆模块学习并优化任务执行
核心代码(来自 samples/task_centric_memory/eval_retrieval.py
):
import asyncio
from autogen_core.utils import load_yaml_file
from autogen_ext.models.openai import OpenAIChatCompletionClient
from your_module import MemoryController, PageLogger, eval_retrieval # 假设的模块导入
async def run_example(config_filepath: str) -> None:
# 1. 加载配置(模型参数、记忆设置等)
config = load_yaml_file(config_filepath)
# 2. 初始化组件
logger = PageLogger(config["PageLogger"]) # 日志记录器
client = OpenAIChatCompletionClient(model="gpt-4o", api_key=config["client"]["api_key"])
memory_controller = MemoryController(
reset=True, # 重置记忆(首次运行)
client=client,
config=config["MemoryController"],
logger=logger,
)
# 3. 执行记忆检索任务(例如:从历史对话中提取关键信息)
results = await eval_retrieval(
memory_controller=memory_controller,
client=client,
logger=logger,
test_config=config["test"] # 测试用例配置
)
print("记忆检索结果:", results)
if __name__ == "__main__":
asyncio.run(run_example("config.yaml"))
关键解析:
- 记忆控制器:
MemoryController
负责存储和检索任务相关记忆(如用户偏好、历史决策)。 - 配置化设计:通过 YAML 文件分离模型参数、测试用例,便于调试和复用。
- 多场景支持:类似逻辑可用于
eval_self_teaching.py
(自我学习)、eval_teachability.py
(从用户指导学习)等场景。
四、工具调用(GraphRAG 搜索)
场景:智能体调用搜索工具回答文档相关问题
核心代码(来自 samples/agentchat_graphrag/app.py
):
import asyncio
import os
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_ext.tools.graphrag import GlobalSearchTool, LocalSearchTool
async def main() -> None:
# 检查 API 密钥
api_key = os.environ.get("OPENAI_API_KEY")
if not api_key:
print("请设置 OPENAI_API_KEY 环境变量")
return
# 1. 初始化模型客户端
model_client = OpenAIChatCompletionClient(model="gpt-4o-mini", api_key=api_key)
# 2. 初始化 GraphRAG 搜索工具
global_tool = GlobalSearchTool.from_settings(root_dir="./", config_filepath="./settings.yaml") # 全局搜索
local_tool = LocalSearchTool.from_settings(root_dir="./", config_filepath="./settings.yaml") # 局部搜索
# 3. 创建智能体(工具选择器)
assistant_agent = AssistantAgent(
name="search_assistant",
tools=[global_tool, local_tool], # 绑定两个搜索工具
model_client=model_client,
system_message=(
"根据用户问题选择工具:\n"
"- 细节问题(如特定实体)用 local_search\n"
"- 全局问题(如整体主题)用 global_search"
)
)
# 4. 执行查询(例如:询问《福尔摩斯》中车站站长对 Becher 博士的评价)
query = "What does the station-master say about Dr. Becher?"
print(f"查询: {query}")
await Console(assistant_agent.run_stream(task=query))
asyncio.run(main())
关键解析:
- 工具分类:
GlobalSearchTool
用于全局理解文档,LocalSearchTool
用于提取特定细节。 - 工具选择逻辑:通过
system_message
明确智能体的工具调用规则,避免无效调用。 - 文档处理:工具会自动处理
input/
目录下的文档(如sherlock_book.txt
),无需手动加载。
运行示例的通用步骤
- 安装依赖:
pip install -U "autogen-agentchat" "autogen-ext[openai]" "python-dotenv"
- 配置 API 密钥:
export OPENAI_API_KEY="your-api-key"
- 运行代码:
python your_script.py
这些示例覆盖了 AutoGen 最核心的功能,可根据实际需求扩展(如添加更多智能体、集成自定义工具等)。
更多推荐
所有评论(0)