前语
在处理复杂任务时，如何让多个智能体高效协作？微软推出的 Magentic-One 开源项目给出了答案。这是一个建立在 AutoGen 平台之上的全新多智能体系统，其目标是让任务分解和智能体协作变得更加智能、高效。就像一位经验丰富的“项目经理”，Magentic-One 通过精准的任务分配和动态调整，让复杂问题的解决过程井井有条，从而大幅提升开发效率。
一、什么是Magentic-One
Magentic-One是基于AutoGen开发的多智能体架构，一共由5层级组成：
1）Orchestrator，主要负责任务分解、规划、指导其他智能体执行子任务、跟踪整体进度并在需要时采取纠正措施；
2）WebSurfer，通过AI大模型能操作、解析网页浏览器内容，执行特定的任务；
3）FileSurfer，基于大模型，可读取多种本地文件并执行预定任务，包括整合、重写等；
4）Coder，同样基于大模型，主要用于编写、分析信息和创建代码；
5）ComputerTerminal，主程序控制台，可执行 Coder 的程序并安装新编程库。
这5大层级智能体会协同分工完成不同的工作，从而加速对任务的自动解析和执行。
二、智能分工协作——核心功能解析
2.1 中枢指挥官：Orchestrator（协调器）
Magentic-One 的核心能力源于其强大的 Orchestrator（协调器）。可以把它想象成一位全能的团队领队，负责管理从规划到执行的每个环节。Orchestrator 的工作主要围绕以下几个方面展开：
任务规划（Task Ledger）：每当接收到一个任务，Orchestrator 首先会制定总体计划，并将目标和相关信息清晰记录在任务账本中，为后续执行提供指导。 进度跟踪（Progress Ledger）：在任务执行过程中，Orchestrator 会实时监控每个子任务的完成情况。如果某项任务没有按计划完成，它会迅速调整策略，将任务重新分配给合适的智能体。 子任务执行与反馈：不同的智能体完成各自的子任务后，将结果反馈给 Orchestrator，后者更新任务进展并执行下一个步骤。 动态调整：如果多次尝试任务仍然未能推进，Orchestrator 会动态调整策略，重新规划任务。
2.2 双循环运行机制
在复杂任务管理中，如何确保每个环节高效推进？Magentic-One 提出了“双循环运行机制”，通过外部与内部两种循环方式实现任务的动态更新与精准跟踪。
外部循环（实心箭头，浅色背景）：负责管理任务账本（包含事实、猜测和计划），确保任务的总体目标明确，并根据情况进行动态调整。
内部循环（虚线箭头，深色背景）：负责管理进度账本（包含当前进度、分配给智能体的任务），跟踪每个子任务的执行情况，如果任务停滞，会通过内循环重新规划任务执行。
三、Magentic-One 的程序流程示例
当 Magentic-One 接到一个自动化任务时，首先由 Orchestrator 将这个复杂的任务分解为多个子任务。例如，分析一个网站的数据，并根据数据编写一份报告，同时参考本地的一些文档作为辅助，Orchestrator 会明确各个子任务的目标和大致流程。
Orchestrator 可以将任务分别委派给 FileSurfer 智能体来读取和处理文件，WebSurfer 智能体来操作 Web 浏览器，或者将任务委派给 Coder 或 ComputerTerminal 智能体来编写或执行代码。
上图展示了一个典型的程序流程：
任务定义（Task Definition）：Orchestrator 接收任务,明确目标是提取网站数据，分析相关信息并根据结果生成报告。随后，它将任务分解为若干子任务，并制定执行顺序。 步骤 1：FileSurfer 智能体访问图像，提取 Python 代码。 步骤 2：Coder 智能体分析 Python 代码。 步骤 3：ComputerTerminal 智能体执行 Python 代码，生成 C++ 代码的 URL。 步骤 4：WebSurfer 智能体访问 URL 并提取 C++ 代码。 步骤 5：另一个 Coder 智能体分析 C++ 代码。 步骤 6：ComputerTerminal 智能体执行 C++ 代码、计算并返回最终结果，完成任务。
在每一步任务完成后，相关智能体会将结果反馈给 Orchestrator，由它整合各部分数据，更新进度账本并执行后续步骤，直到整个任务成功完成。
四、Magentic-One的应用
GitHub 中的 Magentic-One 示例涵盖了从整体框架到单个智能体的多种用法，总共是五个例子：
example.py：用户输入指定任务，系统尝试完成并动态调整，体现了人与智能体的协作能力。 example_coder.py：该代码示例展示了用户与 Coder 和 Executor 代理的交互过程，通过 RoundRobinOrchestrator 协调，用户依次输入指令、生成代码片段并在 Docker 容器中执行。。 example_file_surfer.py：该代码示例展示了用户与 File Surfer 代理的交互过程，通过 RoundRobinOrchestrator 协调，用户与代理依次输入指令或执行操作以导航文件系统。 example_userproxy.py：该代码示例展示了用户与 Coder Agent的交互过程，通过 RoundRobinOrchestrator 协调，用户与Coder Agent依次输入指令或生成代码片段，但代码片段不在此示例中执行。 example_websurfer.py：该代码示例展示了用户与 Web Surfer Agent的交互过程，通过 RoundRobinOrchestrator 协调，用户与Agent依次输入指令或操作，以嵌入的隐身浏览器导航网络。
下面列出的是第一个例子example.py的代码，供大家参考。
"""此示例演示 Magentic-One 执行用户给定的任务并返回最终答案。"""
import argparse
import asyncio
import logging
import os
from autogen_core.application import SingleThreadedAgentRuntime
from autogen_core.application.logging import EVENT_LOGGER_NAME
from autogen_core.base import AgentId, AgentProxy
from autogen_core.components.code_executor import CodeBlock
from autogen_ext.code_executors import DockerCommandLineCodeExecutor
from autogen_magentic_one.agents.coder import Coder, Executor
from autogen_magentic_one.agents.file_surfer import FileSurfer
from autogen_magentic_one.agents.multimodal_web_surfer import MultimodalWebSurfer
from autogen_magentic_one.agents.orchestrator import LedgerOrchestrator
from autogen_magentic_one.agents.user_proxy import UserProxy
from autogen_magentic_one.messages import RequestReplyMessage
from autogen_magentic_one.utils import LogHandler, create_completion_client_from_env
# 注意：运行之前，请使用 'playwright install --with-deps chromium' 安装必要的浏览器依赖项
# 定义一个用于确认代码执行的函数，提示用户确认是否继续执行代码
async def confirm_code(code: CodeBlock) -> bool:
response = await asyncio.to_thread(
input,
f"Executor 即将执行代码（语言: {code.language}）：\n{code.code}\n\n是否要继续执行？（yes/no）：",
)
return response.lower() == "yes"
async def main(logs_dir: str, hil_mode: bool, save_screenshots: bool) -> None:
# 创建用于Agent执行的运行时
runtime = SingleThreadedAgentRuntime()
# 创建一个用于通信的完成客户端（使用 GPT-4o 模型）
client = create_completion_client_from_env(model="gpt-4o")
# 使用 DockerCommandLineCodeExecutor 创建代码执行器
async with DockerCommandLineCodeExecutor(work_dir=logs_dir) as code_executor:
# 在运行时中注册 Coder Agent，用于代码生成
await Coder.register(runtime, "Coder", lambda: Coder(model_client=client))
coder = AgentProxy(AgentId("Coder", "default"), runtime)
# 注册 Executor Agent，用于执行代码
await Executor.register(
runtime,
"Executor",
lambda: Executor("执行代码的Agent", executor=code_executor, confirm_execution=confirm_code),
)
executor = AgentProxy(AgentId("Executor", "default"), runtime)
# 注册其他Agent：WebSurfer 和 FileSurfer
await MultimodalWebSurfer.register(runtime, "WebSurfer", MultimodalWebSurfer)
web_surfer = AgentProxy(AgentId("WebSurfer", "default"), runtime)
await FileSurfer.register(runtime, "file_surfer", lambda: FileSurfer(model_client=client))
file_surfer = AgentProxy(AgentId("file_surfer", "default"), runtime)
await UserProxy.register(
runtime,
"UserProxy",
lambda: UserProxy(description="与您互动的当前用户。"),
)
user_proxy = AgentProxy(AgentId("UserProxy", "default"), runtime)
# 将所有Agent加入到Agent列表中
agent_list = [web_surfer, coder, executor, file_surfer]
if hil_mode:
agent_list.append(user_proxy)
# 注册协调器（LedgerOrchestrator）来管理所有Agent之间的交互
await LedgerOrchestrator.register(
runtime,
"Orchestrator",
lambda: LedgerOrchestrator(
agents=agent_list,
model_client=client,
max_rounds=30, # 最多执行 30 轮交互
max_time=25 * 60, # 最大执行时间为 25 分钟
return_final_answer=True,
),
)
# 启动运行时
runtime.start()
# 初始化网页浏览Agent
actual_surfer = await runtime.try_get_underlying_agent_instance(web_surfer.id, type=MultimodalWebSurfer)
await actual_surfer.init(
model_client=client,
downloads_folder=logs_dir,
start_page="https://www.bing.com", # 浏览器将导航到的初始网页
browser_channel="chromium", # 指定使用哪个浏览器通道
headless=True, # 是否使用无头模式
debug_dir=logs_dir,
to_save_screenshots=save_screenshots, # 是否保存浏览器截图
)
# 发送初始消息以开始与用户Agent交互
await runtime.send_message(RequestReplyMessage(), user_proxy.id)
# 当运行时空闲时停止
await runtime.stop_when_idle()
if __name__ == "__main__":
# 解析命令行参数
parser = argparse.ArgumentParser(description="运行 Magentic-One 示例并设置日志目录。")
parser.add_argument("--logs_dir", type=str, required=True, help="存储日志文件和下载内容的目录")
parser.add_argument("--hil_mode", action="store_true", default=False, help="以人类参与模式运行")
parser.add_argument(
"--save_screenshots", action="store_true", default=False, help="将额外的浏览器截图保存到文件中"
)
args = parser.parse_args()
# 确保日志目录存在
if not os.path.exists(args.logs_dir):
os.makedirs(args.logs_dir)
# 配置应用程序的日志记录
logger = logging.getLogger(EVENT_LOGGER_NAME)
logger.setLevel(logging.INFO)
log_handler = LogHandler(filename=os.path.join(args.logs_dir, "log.jsonl"))
logger.handlers = [log_handler]
# 使用 asyncio 运行主协程
asyncio.run(main(args.logs_dir, args.hil_mode, args.save_screenshots))
代码执行逻辑：
参数解析：从命令行读取日志目录和配置参数。
创建运行时：创建一个单线程运行时来执行智能体。
注册智能体：包括代码生成（Coder）、代码执行（Executor）、网页浏览（WebSurfer）、文件浏览（FileSurfer）、用户代理（UserProxy）。
创建协调器：注册 LedgerOrchestrator，用于管理各个智能体之间的交互，保证任务流顺畅。
启动运行时：开始与网页浏览器的交互，并执行用户任务。
代码执行确认：如果涉及代码执行，系统会提示用户确认。
最终交互与结果返回：通过智能体的协作，完成任务并返回结果。
实际上，微软的 Magentic-One 作为一个做智能体协作框架，在很多领域都大有可为。比如:
数据领域：智能爬取、实时分析、趋势预测… 客服领域：7x24小时无间断、跨平台服务… 软件开发：代码自动生成、文档自动整理… 电商推荐：千人千面、精准触达…
无论是企业级应用还是个人开发，Magentic-One 都能带来显著的效率提升。当然，任何新生事物的发展都需要一个过程。Magentic-One 目前在常见场景优化、提示词个性化等方面还有进一步完善的空间。但其开源属性和模块化设计，为开发者参与优化、满足特定需求提供了极大便利。
我注意到，DeepWisdom 也推出了基于MetaGPT开发的多智能体框架 AFlow。
可以预见，在基础大模型逐渐同质化的大趋势下，下一步的核心竞争点将聚焦于如何基于这些大模型开发更多面向企业应用的智能体（Agent）。
这里要注意的是在国内开发和执行数据/网站爬虫业务的时候，要先去查一下相关的法律法规，或者咨询一下律师，有不少爬虫玩家已经进去了或者即将被起诉，国家对数据安全是越来越重视了。
总结
作为一项全新技术，Magentic-One 在自动化领域展现出了巨大潜力，为我们高效完成复杂工作带来了新思路。尤其在当下人工智能飞速发展的大背景下，类似的多智能体框架必将日益受到关注。Github中的大模型默认是GPT4的，可以自行改成其他的大模型，网上有很多教程，可以自行搜索。
如果你也在为繁重的工作任务所困扰，不妨关注一下 Magentic-One。微软作为AI领域的重要玩家，它的工具一般都是值得去学习的，根据以往的经验，微软的开发工具生命力还是比较强的。
GitHub仓库：https://github.com/microsoft/autogen/tree/main/python/packages/autogen-magentic-one
技术论文：https://www.microsoft.com/en-us/research/publication/magentic-one-a-generalist-multi-agent-system-for-solving-complex-tasks/
如果你觉得今天的分享有帮助，记得点赞、收藏并转发，下次找起来更方便哦！
