测试人快速构建自己的多智能体AI应用

在人工智能飞速发展的今天，智能体（Agent）技术正成为连接大模型与现实应用的重要桥梁。作为微软推出的多智能体开发框架，AutoGen正在重新定义我们构建AI应用的方式。无论你是AI初学者还是资深开发者，掌握AutoGen都将为你的技术栈增添强大助力。
AutoGen：多智能体开发的新范式
AutoGen是一个专门用于创建多智能体AI应用的先进框架，其核心理念是让多个智能体能够自主运行或与人类协作完成复杂任务。
核心思想：”通过对话解决问题”。它允许你定义多种不同角色和能力的智能体（例如，一个擅长写代码的“程序员”，一个负责执行代码的“助手”，一个提供专业知识的“专家”），并设定它们之间的交互规则。
关键组件：
AssistantAgent：通常是基于大语言模型（如GPT-4）的“大脑”，负责思考、规划和生成回复。
UserProxyAgent：一个特殊的智能体,通过输入函数代表人类用户，主要用于在团队运行期间提供人类反馈，比如执行代码、调用函数等。
RoundRobinGroupChat：最简单的团队类型,参与者按照轮询顺序依次发言。每个智能体在轮到它时会向所有其他参与者广播消息
SelectorGroupChat：在每条消息后，使用 ChatCompletion 模型选择下一位发言人的团队。
Chainlit：为LLM应用量身定制的UI框架
Chainlit是一个专为大语言模型应用设计的开源Python框架，让你用极少的代码快速构建功能丰富的Web应用。
核心特性：
实时流式输出：文字逐个”流”出来的丝滑体验
丰富的元素支持：文本、、PDF、代码块、思维链等
会话记忆：天然支持多轮对话
极简开发：几十行代码即可构建完整应用
AutoGen + Chainlit 协同优势
AutoGen：负责多智能体的核心逻辑，处理智能体间的对话协作和任务执行
Chainlit：提供现代化的Web界面，实现实时消息流、可视化对话和交互式体验
这两者的结合，让开发者既能享受AutoGen强大的多智能体能力，又能通过Chainlit提供优秀的用户体验。
环境配置
pip install autogen-agentchat autogen-ext[openai] chainlit
实例1：基础聊天界面
import chainlit as cl
from typing import List, cast
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.messages import ModelClientStreamingChunkEvent, TextMessage
from autogen_core import CancellationToken

模型客户端配置

from model_client import deepseek_client

@cl.set_starters
asyncdefset_starts() -> List[cl.Starter]:
“””设置聊天启动选项”””
return [
cl.Starter(
label=’Python 脚本’,
message=’请写一个Python脚本，实现合并两个有序列表’,
),
cl.Starter(
label=’SQL 脚本’,
message=’请写一个SQL脚本，查询所有用户的姓名和年龄’,
)
]

@cl.on_chat_start
asyncdefstart() -> None:
“””聊天开始时初始化智能体”””
agent = AssistantAgent(
name=’Python助手’,
model_client=deepseek_client,
system_message=’你是一个Python脚本助手，用于根据用户查询返回相关的Python脚本。’,
model_client_stream=True
)
cl.user_session.set(‘agent’, agent)

@cl.on_message
asyncdefchat(message: cl.Message) -> None:
“””处理用户消息”””
agent = cast(AssistantAgent, cl.user_session.get(“agent”))
response = cl.Message(content=””)

流式处理智能体响应

asyncfor msg in agent.on_messages_stream(
messages=[TextMessage(content=message.content, source=”user”)],
cancellation_token=CancellationToken(),
):
if isinstance(msg, ModelClientStreamingChunkEvent):
await response.stream_token(msg.content)
elif isinstance(msg, Response):
await response.send()
执行效果
实例2：多智能体协同系统

import chainlit as cl
from typing import List, cast
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import TextMentionTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.messages import ModelClientStreamingChunkEvent, TextMessage
from autogen_core import CancellationToken

模型客户端配置

from model_client import deepseek_client

@cl.on_chat_start
asyncdefon_chat_start():
“””初始化多智能体团队”””

创建助手智能体 – 负责主要任务执行

assistant_agent = AssistantAgent(
    name='assistant_agent',
    model_client=deepseek_client,
    system_message='you are a helpful assistant',
    model_client_stream=True
)

创建评审智能体 – 负责提供反馈和审核

critic_agent = AssistantAgent(
    name='critic_agent', 
    model_client=deepseek_client,
    system_message='You are a critic. Provide constructive feedback. Respond with "APPROVE" if your feedback has been addressed.',
    model_client_stream=True
)

设置终止条件 – 当评审智能体说出”APPROVE”时结束对话

termination = TextMentionTermination('APPROVE', sources=['critic_agent'])

创建轮询群聊团队 – 智能体轮流发言

group_chat = RoundRobinGroupChat(
    [assistant_agent, critic_agent],
    termination_condition=termination
)

cl.user_session.set('team', group_chat)

@cl.set_starters
asyncdefset_starts() -> List[cl.Starter]:
“””定义聊天启动器”””
return [
cl.Starter(
label=”诗歌创作”,
message=”Write a poem about the ocean.”,
),
cl.Starter(
label=”故事写作”,
message=”Write a story about a detective solving a mystery.”,
),
cl.Starter(
label=”代码编写”,
message=”Write a function that merge two list of numbers into single sorted list.”,
),
]

@cl.on_message
asyncdefchat(message: cl.Message) -> None:
“””处理团队对话”””
team = cast(RoundRobinGroupChat, cl.user_session.get(“team”))
streaming_response = None

运行团队对话流

asyncfor msg in team.run_stream(
task=[TextMessage(content=message.content, source=”user”)],
cancellation_token=CancellationToken(),
):
if isinstance(msg, ModelClientStreamingChunkEvent):

流式输出处理

if streaming_response isNone:
streaming_response = cl.Message(content=msg.source + “: “, author=msg.source)
await streaming_response.stream_token(msg.content)
elif streaming_response isnotNone:

完成流式输出

await streaming_response.send()
streaming_response = None
elif isinstance(msg, TaskResult):

任务终止处理

        final_message = "任务完成。"

if msg.stop_reason:
final_message += msg.stop_reason
await cl.Message(content=final_message).send()

代码精讲
@cl.on_chat_start：这是一个Chainlit装饰器，在每次用户打开新的聊天窗口或开始新会话时自动触发，在这里创建智能体实例并存入 cl.user_session（一个与当前用户会话绑定的临时存储空间）
@cl.on_message： 这是应用的核心消息处理函数，相当于聊天机器人的”大脑”，接收用户的 message 对象，包含消息内容和元数据，每当用户发送一条消息（无论是手动输入还是点击starter），这个函数就会自动调用
@cl.set_starters：定义聊天界面的快捷启动按钮，提升用户体验，用户可以直接点击预设问题开始对话，无需手动输入
执行效果

运行应用
在终端中运行：
chainlit run your_script.py
技术架构优势
AutoGen的分层设计
Foundation Layer (基础层): autogen-core 实现 AgentRuntime、消息类型和工具框架
Agent Framework Layer (智能体框架层): autogen-agentchat 提供 AssistantAgent、GroupChat 等高级抽象
Extensions Layer (扩展层): autogen-ext 提供模型客户端、代码执行器和 MCP 集成
Application Layer (应用层): autogenstudio 提供无代码 GUI 构建器
多智能体协作模式
两个智能体通过对话协作解决问题，特别适合：
自动代码生成：程序员智能体 + 执行助手智能体
复杂问题求解：多个专家智能体协同工作
质量保证：执行智能体 + 评审智能体的双重检查机制
这种架构既保证了框架的稳定性，又提供了极大的灵活性，让开发者可以根据需求在不同层次上进行定制开发。