一、AgentScope 框架概述

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

特性	说明
🎯 消息驱动	基于消息原语的通信机制，支持丰富的消息类型
🧩 模块化设计	Agent、Memory、Parser、Service 完全可配置
🔄 流程编排	Pipeline + MsgHub 实现复杂工作流编排
📊 内置监控	WebUI 可视化监控智能体运行状态
🔌 多模型支持	OpenAI、Doubao、Qwen、LLaMA 等 20+ 模型
🛡️ 高鲁棒性	容错机制、重试策略、异常处理

官方资源：

• 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 最佳实践清单

建议	说明
✅ 明确角色边界	每个 Agent 只做一件事，职责单一
✅ 用好 MsgHub	多 Agent 协作优先用 MsgHub 广播模式
✅ Memory 分层	短期记忆用 ListMemory，长期知识用 VectorMemory
✅ 异常处理	给关键 Agent 添加重试和 fallback 机制
✅ 成本监控	开启 Token 统计，设置预算告警
✅ WebUI 必开	可视化调试效率提升 5x+
❌ 避免过度工程	简单任务用 SequentialPipeline，不要过度设计
❌ 无限上下文	合理设置 max_memory，避免上下文爆炸

八、框架对比总结

特性	AgentScope	AutoGen	LangGraph	CrewAI
工作流编排	⭐⭐⭐⭐⭐ Pipeline	⭐⭐⭐⭐ GroupChat	⭐⭐⭐⭐⭐ Graph	⭐⭐⭐ Process
消息通信	⭐⭐⭐⭐⭐ MsgHub	⭐⭐⭐ 内置	⭐⭐⭐ 显式边	⭐⭐⭐ 任务委派
可视化监控	⭐⭐⭐⭐⭐ WebUI	⭐⭐ 有限	⭐⭐⭐ 需集成	⭐ Tracer
容错机制	⭐⭐⭐⭐⭐ 完整	⭐⭐⭐ 中等	⭐⭐⭐ 中等	⭐⭐ 基础
中文支持	⭐⭐⭐⭐⭐ 原生	⭐⭐⭐ 一般	⭐⭐⭐ 一般	⭐⭐⭐ 一般
国内模型	✅ 完整支持	❌ 需自行适配	❌ 需自行适配	❌ 需自行适配
学习曲线	🟢 平缓	🟡 中等	🔴 陡峭	🟡 中等

---

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