AgentScope 深度解析：2026年多智能体系统开发实战指南

一、AgentScope 框架概述

AgentScope 是阿里巴巴达摩院 2024 年开源的多智能体开发框架，GitHub 星标 12K+，核心优势：

官方资源：

• GitHub: https://github.com/modelscope/agentscope
• 文档: https://agentscope.readthedocs.io
• 论文: AgentScope: A Flexible Platform for Multi-Agent Systems

二、环境搭建与快速开始

2.1 安装与初始化

# 基础安装
pip install agentscope

# 完整安装（含 WebUI、服务等所有功能）
pip install "agentscope[all]"

# 验证安装
python -c "import agentscope; print(agentscope.__version__)"
# 输出：0.1.0

2.2 最小可运行示例

# quick_start.py – 5 行代码跑通第一个 Agent
import agentscope
from agentscope.agents import DialogAgent

# 1. 初始化框架（配置 API Key）
agentscope.init(
model_configs={
"config_name": "doubao-pro",
"model_type": "openai",
"model_name": "doubao-pro-32k",
"api_key": "your-api-key",
"client_args": {
"base_url": "https://ark.cn-beijing.volces.com/api/v3"
}
}
)

# 2. 创建对话 Agent
assistant = DialogAgent(
name="助手",
model_config_name="doubao-pro",
system_prompt="你是一位专业的技术顾问，回答简洁准确。"
)

# 3. 与 Agent 对话
response = assistant("请解释什么是多智能体系统？")
print(response.content)

运行输出：

多智能体系统（Multi-Agent System, MAS）是由多个自主智能体组成
的计算系统，这些智能体通过相互协作完成单个智能体难以完成的复杂
任务。每个智能体具有自主性、社会性、反应性和主动性等特征…

三、核心概念深度解析

3.1 Agent 基类与常用 Agent

from agentscope.agents import (
AgentBase, # 所有 Agent 基类
DialogAgent, # 基础对话 Agent
UserAgent, # 用户输入 Agent
ReActAgent, # ReAct 推理 Agent
ExecutiveAgent, # 工具调用 Agent
CriticAgent, # 批判/反思 Agent
OptimAgent, # 代码优化 Agent
)

# ================= 自定义 Agent 开发 =================
class CodeReviewAgent(AgentBase):
"""自定义代码审查 Agent"""

def __init__(
self,
name: str,
model_config_name: str,
**kwargs
):
super().__init__(name=name, **kwargs)

# 初始化 LLM 客户端
self.model = agentscope.init_model(model_config_name)

# 记忆系统（默认 10 轮上下文）
self.memory = agentscope.init_memory(
memory_type="list",
max_memory=10
)

# 系统提示词
self.system_prompt = """
你是资深代码审查专家，擅长发现：
1. 潜在 Bug 和逻辑错误
2. 性能优化点
3. 代码规范问题
4. 安全漏洞
5. 架构设计缺陷

输出格式：
🐛 问题描述
💡 修改建议
✅ 优化后的代码

只审查代码本身，不要泛泛而谈。
"""

def reply(self, x: dict = None) –> dict:
"""核心回复逻辑"""
# 1. 将用户输入加入记忆
if x is not None:
self.memory.add(x)

# 2. 构建提示词
prompt = self.model.format(
self.system_prompt,
self.memory.get_memory()
)

# 3. 调用 LLM
response = self.model(prompt)

# 4. 将回复加入记忆并返回
msg = self.model_to_msg(response)
self.memory.add(msg)

return msg

# 使用自定义 Agent
reviewer = CodeReviewAgent(
name="CodeReviewer",
model_config_name="doubao-pro"
)

result = reviewer({
"content": """
def calculate_sum(numbers):
total = 0
for n in numbers:
total += n
return total
"""
})
print(result.content)

3.2 Memory 记忆系统详解

from agentscope.memory import (
ListMemory, # 列表记忆（默认）
VectorMemory, # 向量记忆（RAG 检索）
SummaryMemory, # 摘要记忆
MistralMemory, # Mistral 增强记忆
)

# ================= 向量记忆 RAG 示例 =================
import faiss
from sentence_transformers import SentenceTransformer

memory = VectorMemory(
embedding_model=SentenceTransformer('all-MiniLM-L6-v2'),
index=faiss.IndexFlatL2(384),
max_memory=100,
retrieve_size=5 # 每次检索最相关的 5 条记忆
)

# 添加文档到记忆
documents = [
"AgentScope 是阿里巴巴开源的多智能体框架",
"支持 OpenAI、Claude、Qwen、Doubao 等多种模型",
"基于 Pipeline 和 MsgHub 实现工作流编排",
"内置 WebUI 可视化监控面板",
"支持 ReAct、Reflection 等高级 Agent 模式"
]

for doc in documents:
memory.add({"content": doc, "role": "system"})

# 检索相关记忆
query = "AgentScope 支持哪些模型？"
retrieved = memory.retrieve(query)
print("检索到相关记忆:", len(retrieved))
for m in retrieved:
print(f" – {m['content']}")

3.3 Service 服务层（工具调用）

from agentscope.service import (
ServiceTool, # 服务工具基类
ServiceResponse, # 服务响应
ServiceExecStatus, # 执行状态
)
from agentscope.service import (
execute_python_code, # 执行 Python 代码
web_search, # 网络搜索
read_text_file, # 读取文件
write_text_file, # 写入文件
retrieval, # 向量检索
sql_query, # SQL 查询
)

# ================= 自定义服务开发 =================
@ServiceTool
def calculate_complexity(code: str) –> ServiceResponse:
"""计算代码圈复杂度

Args:
code: 源代码字符串

Returns:
圈复杂度分析结果
"""
try:
# 使用 radon 计算圈复杂度
import radon.complexity as complexity

results = complexity.cc_visit(code)

analysis = []
for result in results:
analysis.append({
"name": result.name,
"complexity": result.complexity,
"lineno": result.lineno,
"rank": result.rank
})

return ServiceResponse(
status=ServiceExecStatus.SUCCESS,
content=analysis
)

except Exception as e:
return ServiceResponse(
status=ServiceExecStatus.ERROR,
content=f"计算失败: {str(e)}"
)

# 在 Agent 中使用服务
from agentscope.agents import ExecutiveAgent

developer = ExecutiveAgent(
name="Developer",
model_config_name="doubao-pro",
tools=[calculate_complexity, execute_python_code]
)

response = developer("分析下面这段代码的复杂度并优化：\\n" + code_snippet)

四、Pipeline 工作流编排

4.1 SequentialPipeline 顺序执行

from agentscope.pipelines import SequentialPipeline

# ================= 写作工作流 =================
# 策划 → 写作 → 润色 → 审查

planner = DialogAgent(
name="策划",
model_config_name="doubao-pro",
system_prompt="你是内容策划，负责生成文章大纲。"
)

writer = DialogAgent(
name="写作",
model_config_name="doubao-pro",
system_prompt="你是技术作家，根据大纲撰写完整文章。"
)

editor = DialogAgent(
name="润色",
model_config_name="doubao-pro",
system_prompt="你是编辑，负责润色文章提升可读性。"
)

reviewer = DialogAgent(
name="审查",
model_config_name="doubao-pro",
system_prompt="你是审稿人，负责检查技术准确性。"
)

# 创建顺序 Pipeline
pipeline = SequentialPipeline([planner, writer, editor, reviewer])

# 执行工作流
result = pipeline("写一篇关于 AgentScope 的技术文章")
print("最终文章：\\n", result[–1].content)

4.2 IfElsePipeline 分支逻辑

from agentscope.pipelines import IfElsePipeline

def need_optimization(msg: dict) –> bool:
"""判断是否需要优化"""
content = msg.content.lower()
return any(
keyword in content
for keyword in ["慢", "性能", "优化", "效率"]
)

# 分支 Pipeline
optimization_pipeline = IfElsePipeline(
condition=need_optimization,
if_body=[code_optimizer, performance_tester],
else_body=[code_reviewer]
)

# 使用
result = optimization_pipeline(user_input)

4.3 LoopPipeline 循环迭代

from agentscope.pipelines import LoopPipeline

# 循环优化直到代码通过所有测试
refinement_loop = LoopPipeline(
pipeline=[code_generator, unit_tester],
max_iterations=5, # 最多迭代 5 次
stop_condition=lambda x: "所有测试通过" in x[–1].content
)

result = refinement_loop("实现一个 LRU 缓存")

4.4 MsgHub 消息中枢

from agentscope.msghub import MsgHub

# ================= 辩论场景：正方 vs 反方 =================
pro_side = DialogAgent(
name="正方",
model_config_name="doubao-pro",
system_prompt="你是正方辩手，支持 AI 取代程序员观点。"
)

con_side = DialogAgent(
name="反方",
model_config_name="doubao-pro",
system_prompt="你是反方辩手，反对 AI 取代程序员观点。"
)

judge = DialogAgent(
name="评委",
model_config_name="doubao-pro",
system_prompt="你是辩论评委，总结辩论并给出观点。"
)

# 创建消息中枢，所有参与者都能收到消息
with MsgHub([pro_side, con_side, judge]) as hub:
# 正方发言
hub.broadcast(pro_side("我认为 AI 将在 5 年内取代大部分程序员…"))

# 反方回应（能看到正方发言）
hub.broadcast(con_side("我方不同意，因为…"))

# 第二轮辩论
hub.broadcast(pro_side("针对反方观点，我想补充…"))
hub.broadcast(con_side("正方的论证忽略了…"))

# 评委总结
final_summary = judge("请总结本次辩论并给出你的判断")
print(final_summary.content)

五、实战：代码评审多智能体系统

# code_review_system.py – 完整的代码评审系统
import agentscope
from agentscope.agents import DialogAgent
from agentscope.pipelines import SequentialPipeline
from agentscope.msghub import MsgHub
from agentscope.service import execute_python_code

# 1. 初始化
agentscope.init(model_configs="model_configs.json")

# 2. 创建角色 Agent
static_analyzer = DialogAgent(
name="静态分析专家",
model_config_name="doubao-pro",
system_prompt="""
你是静态代码分析专家，擅长：
– 发现语法错误和类型问题
– 识别代码异味和坏味道
– 检查代码规范一致性
输出格式：分点列出，附行号
"""
)

performance_auditor = DialogAgent(
name="性能审计师",
model_config_name="doubao-pro",
system_prompt="""
你是性能优化专家，擅长：
– 识别时间复杂度问题
– 发现内存泄漏风险
– 给出算法优化建议
输出格式：影响范围 + 优化方案 + 预期收益
"""
)

security_reviewer = DialogAgent(
name="安全审查员",
model_config_name="doubao-pro",
system_prompt="""
你是应用安全专家，擅长：
– SQL 注入、XSS 等漏洞识别
– 认证授权缺陷
– 敏感数据泄露风险
输出格式：风险等级 + 漏洞描述 + 修复方案
"""
)

architecture_advisor = DialogAgent(
name="架构师",
model_config_name="doubao-pro",
system_prompt="""
你是系统架构师，负责：
– 评估整体设计合理性
– 识别架构缺陷
– 给出重构建议
输出格式：架构评估 + 问题分析 + 改进方案
"""
)

chief_architect = DialogAgent(
name="总架构师",
model_config_name="doubao-pro",
system_prompt="""
你是总架构师，整合所有审查意见，生成最终报告。
要求：
1. 按严重程度排序
2. 给出优先级（P0/P1/P2）
3. 生成可执行的修复计划
"""
)

# 3. 运行代码评审工作流
def run_code_review(code: str) –> dict:
"""执行完整代码评审"""

print("🔍 开始代码评审…\\n")

# 并行执行各专家评审（MsgHub 广播模式）
with MsgHub([
static_analyzer,
performance_auditor,
security_reviewer,
architecture_advisor,
chief_architect
]) as hub:

# 广播待评审代码
review_request = {
"content": f"请评审以下代码：\\n```python\\n{code}\\n```"
}

hub.broadcast(static_analyzer(review_request))
hub.broadcast(performance_auditor(review_request))
hub.broadcast(security_reviewer(review_request))
hub.broadcast(architecture_advisor(review_request))

# 总架构师汇总（能看到所有专家意见）
final_report = chief_architect("生成最终评审报告")

return {
"static_analysis": static_analyzer.memory.get_memory()[–1],
"performance_audit": performance_auditor.memory.get_memory()[–1],
"security_review": security_reviewer.memory.get_memory()[–1],
"architecture_advice": architecture_advisor.memory.get_memory()[–1],
"final_report": final_report
}

# 4. 使用示例
if __name__ == "__main__":
code_to_review = """
def process_user_data(users):
result = []
for user in users:
sql = f"SELECT * FROM data WHERE id = {user.id}"
cursor.execute(sql)
data = cursor.fetchall()
result.append(data)
return result
"""

report = run_code_review(code_to_review)

print("\\n" + "="*60)
print("📋 最终评审报告")
print("="*60)
print(report["final_report"].content)

六、WebUI 可视化监控

# start_webui.py – 启动可视化监控
import agentscope

agentscope.init(
model_configs="model_configs.json",
studio_url="http://localhost:8080" # 启用 WebUI
)

# 启动 AgentScope Studio
agentscope.studio.launch(
host="0.0.0.0",
port=8080,
debug=True
)

# 浏览器访问：http://localhost:8080
# 功能包括：
# – Agent 实时状态监控
# – 消息流可视化
# – 性能指标图表
# – 日志搜索与过滤
# – 系统资源使用情况

七、性能优化与最佳实践

7.1 并发优化

from concurrent.futures import ThreadPoolExecutor
import functools

# 并行执行多个 Agent
def parallel_execute(agents, task: str, max_workers: int = 4):
"""并行执行多个 Agent"""
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = [
executor.submit(agent, task)
for agent in agents
]
results = [f.result() for f in futures]
return results

# 使用
agents = [agent1, agent2, agent3, agent4]
results = parallel_execute(agents, "评审这段代码", max_workers=4)

7.2 成本控制策略

# Token 消耗监控
from agentscope.callback import TokenCounterCallback

# 注册 Token 统计回调
token_counter = TokenCounterCallback()
agentscope.register_callback(token_counter)

# 执行任务后查看统计
print(f"输入 Token: {token_counter.prompt_tokens}")
print(f"输出 Token: {token_counter.completion_tokens}")
print(f"总成本: ${token_counter.total_cost:.4f}")

# 成本控制配置
agentscope.init(
model_configs="model_configs.json",
budget_config={
"max_budget_per_month": 50.0, # 每月 50 美元
"alert_threshold": 0.8, # 用到 80% 告警
"auto_stop": True # 超预算自动停止
}
)

7.3 最佳实践清单

八、框架对比总结

结论：AgentScope 是目前国内多智能体框架的最佳选择，尤其适合需要支持豆包、通义千问等国产大模型的项目。其 Pipeline + MsgHub 设计让复杂工作流编排变得简单直观，内置 WebUI 更是调试和监控的利器。