深入浅出 AgentScope 2.0：打造你的 AI 智能体军团（下篇）

六、多 Agent 协作

6.1 为什么需要多 Agent？

单个 Agent 虽然强大，但在复杂场景下往往力不从心。想象一下：

• 软件开发：需要产品经理、架构师、程序员、测试工程师协作
• 投资分析：需要行业研究员、财务分析师、风险评估师协作
• 内容创作：需要策划、撰稿、编辑、审核协作

这就是多 Agent 协作的价值所在——让多个专业化的 Agent 组成团队，共同完成复杂任务。

6.2 多 Agent 协作模式

AgentScope 支持多种多 Agent 协作模式：

模式一：顺序执行（Sequential）

Agent 按顺序依次执行，前一个 Agent 的输出作为后一个 Agent 的输入。

用户输入 → Agent A → Agent B → Agent C → 最终输出

示例：文章创作流水线

import asyncio
from agentscope.agent import Agent
from agentscope.message import UserMsg, AssistantMsg

async def sequential_workflow():
# 创建三个专业 Agent
planner = Agent(
name="策划",
system_prompt="你是一个内容策划专家，负责规划文章大纲和结构。",
model=model,
)

writer = Agent(
name="撰稿",
system_prompt="你是一个专业撰稿人，根据大纲撰写文章内容。",
model=model,
)

editor = Agent(
name="编辑",
system_prompt="你是一个资深编辑，负责审核和优化文章。",
model=model,
)

# 顺序执行
topic = "人工智能在教育领域的应用"

# Step 1: 策划生成大纲
outline = await planner.reply(
UserMsg("用户", f"请为「{topic}」这篇文章规划大纲")
)
print(f"大纲：\\n{outline.content}\\n")

# Step 2: 撰写根据大纲写文章
article = await writer.reply(
UserMsg("用户", f"根据以下大纲撰写文章：\\n{outline.content}")
)
print(f"初稿：\\n{article.content}\\n")

# Step 3: 编辑审核优化
final = await editor.reply(
UserMsg("用户", f"请审核并优化以下文章：\\n{article.content}")
)
print(f"终稿：\\n{final.content}")

asyncio.run(sequential_workflow())

模式二：并行执行（Parallel）

多个 Agent 同时执行，最后汇总结果。

┌─────────┐
│ Agent A │
└─────────┘
│
用户输入 ───────────┼──────────┬──────────→ 汇总结果
│ │
└─────────┐│
│ Agent B ││
└─────────┘│
│
┌─────────┐│
│ Agent C ││
└─────────┘│

示例：多角度分析

import asyncio
from agentscope.agent import Agent
from agentscope.message import UserMsg

async def parallel_analysis():
# 创建三个分析师 Agent
technical_analyst = Agent(
name="技术分析师",
system_prompt="你从技术角度分析问题。",
model=model,
)

business_analyst = Agent(
name="商业分析师",
system_prompt="你从商业角度分析问题。",
model=model,
)

risk_analyst = Agent(
name="风险分析师",
system_prompt="你从风险角度分析问题。",
model=model,
)

topic = "企业是否应该采用大语言模型技术？"

# 并行执行
results = await asyncio.gather(
technical_analyst.reply(UserMsg("用户", topic)),
business_analyst.reply(UserMsg("用户", topic)),
risk_analyst.reply(UserMsg("用户", topic)),
)

# 汇总结果
print("=== 技术分析 ===")
print(results[0].content)
print("\\n=== 商业分析 ===")
print(results[1].content)
print("\\n=== 风险分析 ===")
print(results[2].content)

asyncio.run(parallel_analysis())

模式三：层级协作（Hierarchical）

一个「管理者」Agent 协调多个「执行者」Agent。

┌──────────────┐
│ Manager Agent│
│ (管理者) │
└──────────────┘
│
┌───────────────┼───────────────┐
│ │ │
▼ ▼ ▼
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ Worker A │ │ Worker B │ │ Worker C │
│ (执行者) │ │ (执行者) │ │ (执行者) │
└─────────────┘ └─────────────┘ └─────────────┘

示例：项目管理团队

import asyncio
from agentscope.agent import Agent
from agentscope.message import UserMsg
from agentscope.tool import Toolkit, FunctionTool

# 定义任务分配工具
def assign_task(worker: str, task: str) -> str:
"""分配任务给指定的工作者"""
return f"已将任务「{task}」分配给 {worker}"

# 创建工作者 Agent
researcher = Agent(
name="研究员",
system_prompt="你负责收集和整理信息。",
model=model,
)

developer = Agent(
name="开发者",
system_prompt="你负责编写代码和实现功能。",
model=model,
toolkit=Toolkit(tools=[
FunctionTool(assign_task)
])
)

tester = Agent(
name="测试员",
system_prompt="你负责测试和质量保证。",
model=model,
)

# 创建管理者 Agent
manager = Agent(
name="项目经理",
system_prompt="""你是一个项目经理，负责协调团队成员完成任务。
你可以分配任务给以下成员：
– 研究员：负责信息收集
– 开发者：负责代码实现
– 测试员：负责测试验证
""",
model=model,
)

async def hierarchical_workflow():
task = "开发一个简单的待办事项应用"

result = await manager.reply(
UserMsg("用户", f"请组织团队完成以下任务：{task}")
)
print(result.content)

asyncio.run(hierarchical_workflow())

6.3 消息中心（MsgHub）

AgentScope 提供了消息中心（MsgHub） 来实现灵活的多 Agent 通信。

MsgHub 的作用

MsgHub 是一个消息广播机制，允许多个 Agent 订阅和发布消息：

┌─────────────────────────────────────────────────────────────┐
│ MsgHub │
├─────────────────────────────────────────────────────────────┤
│ │
│ Agent A ──────┐ │
│ │ │
│ Agent B ──────┼──────→ 消息队列 ──────→ 广播给所有订阅者 │
│ │ │
│ Agent C ──────┘ │
│ │
└─────────────────────────────────────────────────────────────┘

使用 MsgHub

import asyncio
from agentscope.agent import Agent
from agentscope.message import UserMsg
from agentscope.hub import MsgHub

async def msghub_example():
# 创建 MsgHub
hub = MsgHub()

# 创建 Agent 并订阅 MsgHub
agent_a = Agent(
name="Agent A",
system_prompt="你是 Agent A，负责分析问题。",
model=model,
)
hub.subscribe(agent_a)

agent_b = Agent(
name="Agent B",
system_prompt="你是 Agent B，负责提供建议。",
model=model,
)
hub.subscribe(agent_b)

# 发布消息
await hub.publish(UserMsg("用户", "请讨论一下 AI 的未来发展趋势"))

# Agent A 处理消息
result_a = await agent_a.reply(None) # 从 MsgHub 获取消息
await hub.publish(result_a)

# Agent B 看到消息并回复
result_b = await agent_b.reply(None)
await hub.publish(result_b)

asyncio.run(msghub_example())

七、工具开发详解

7.1 工具的本质

在 AgentScope 中，工具本质上是一个可以被 Agent 调用的函数。Agent 通过阅读工具的描述（docstring）来理解工具的用途，然后决定何时调用。

7.2 创建自定义工具

方式一：使用 FunctionTool

最简单的方式是使用 FunctionTool 包装普通 Python 函数：

from agentscope.tool import FunctionTool

def search_database(query: str, limit: int = 10) -> str:
"""
在数据库中搜索信息。

参数:
query: 搜索关键词
limit: 返回结果的最大数量，默认为 10

返回:
搜索结果的字符串表示
"""
# 模拟数据库搜索
results = [
f"结果 {i+1}: 包含 '{query}' 的记录"
for i in range(limit)
]
return "\\n".join(results)

# 创建工具
search_tool = FunctionTool(search_database)

# 添加到工具集
toolkit = Toolkit(tools=[search_tool])

重要提示：

• 函数必须有清晰的类型注解
• 函数必须有详细的文档字符串
• 文档字符串应该说明参数的含义和返回值

方式二：继承 ToolBase

对于更复杂的工具，可以继承 ToolBase 类：

from agentscope.tool import ToolBase, ToolResponse
from typing import Any

class WeatherTool(ToolBase):
"""天气查询工具"""

name = "get_weather"
description = "获取指定城市的天气信息"

# 定义参数 schema
parameters_schema = {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称，如'北京'、'上海'"
},
"date": {
"type": "string",
"description": "日期，格式为 YYYY-MM-DD，默认为今天",
"default": "今天"
}
},
"required": ["city"]
}

async def __call__(
self,
city: str,
date: str = "今天",
**kwargs: Any
) -> ToolResponse:
"""
执行天气查询

参数:
city: 城市名称
date: 查询日期

返回:
ToolResponse 包含天气信息
"""
# 这里应该调用真实的天气 API
# 示例使用模拟数据
weather_data = {
"北京": {
"今天": "晴，18-25℃，空气质量良好",
"明天": "多云，16-23℃，有轻微雾霾"
},
"上海": {
"今天": "小雨，20-26℃，建议带伞",
"明天": "阴，19-25℃"
}
}

result = weather_data.get(city, {}).get(date, "暂无数据")

return ToolResponse(
content=result,
metadata={"city": city, "date": date}
)

# 使用自定义工具
toolkit = Toolkit(tools=[WeatherTool()])

7.3 工具的高级特性

异步工具

工具可以是异步函数，适合执行 I/O 密集型操作：

from agentscope.tool import FunctionTool
import aiohttp

async def fetch_url(url: str) -> str:
"""
异步获取网页内容。

参数:
url: 要获取的网页 URL

返回:
网页内容
"""
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return await response.text()

fetch_tool = FunctionTool(fetch_url)

工具权限控制

可以为工具设置权限级别，控制 Agent 的调用行为：

from agentscope.tool import FunctionTool
from agentscope.permission import PermissionLevel

def delete_file(path: str) -> str:
"""
删除指定文件（危险操作）。

参数:
path: 文件路径

返回:
操作结果
"""
import os
os.remove(path)
return f"文件 {path} 已删除"

delete_tool = FunctionTool(
delete_file,
permission_level=PermissionLevel.HIGH, # 高权限要求
)

工具结果处理

工具可以返回多种类型的结果：

from agentscope.tool import ToolResponse
from agentscope.message import DataBlock, Base64Source
import base64

async def generate_image(prompt: str) -> ToolResponse:
"""
根据描述生成图片。

参数:
prompt: 图片描述

返回:
生成的图片
"""
# 调用图片生成 API（示例）
image_data = await call_image_api(prompt)

# 返回图片数据
return ToolResponse(
content="图片已生成",
blocks=[
DataBlock(
type="image",
source=Base64Source(
media_type="image/png",
data=base64.b64encode(image_data).decode()
)
)
]
)

7.4 MCP 工具集成

AgentScope 支持 MCP（Model Context Protocol），可以轻松集成外部 MCP 服务器提供的工具：

from agentscope.tool import MCPTool

# 连接到 MCP 服务器
mcp_tool = MCPTool(
server_url="http://localhost:8080/mcp",
tool_name="database_query",
)

toolkit = Toolkit(tools=[mcp_tool])

八、权限控制与安全

8.1 为什么需要权限控制？

当 Agent 可以执行 Shell 命令、读写文件、调用 API 时，安全问题变得至关重要：

• 数据泄露：Agent 可能读取敏感文件
• 系统破坏：Agent 可能执行危险的 Shell 命令
• 资源滥用：Agent 可能调用昂贵的 API

8.2 AgentScope 的权限系统

AgentScope 提供了细粒度的权限控制机制：

权限级别

权限行为

配置权限

from agentscope.agent import Agent
from agentscope.permission import PermissionBehavior, PermissionLevel
from agentscope.tool import Toolkit, Bash, Read, Write

# 配置权限行为
permission_behavior = PermissionBehavior(
# 低权限操作：自动允许
low=PermissionBehavior.ALLOW,
# 中权限操作：询问用户
medium=PermissionBehavior.ASK,
# 高权限操作：拒绝
high=PermissionBehavior.DENY,
)

agent = Agent(
name="助手",
system_prompt="你是一个有帮助的助手。",
model=model,
toolkit=Toolkit(tools=[
Bash(permission_level=PermissionLevel.HIGH), # 高权限
Read(permission_level=PermissionLevel.LOW), # 低权限
Write(permission_level=PermissionLevel.MEDIUM), # 中权限
]),
)

# 设置权限行为
agent.state.permission_context.set_behavior(permission_behavior)

8.3 人机协作（Human-in-the-Loop）

对于敏感操作，AgentScope 支持「人机协作」模式，让人类确认 Agent 的决策：

from agentscope.event import EventType, RequireUserConfirmEvent

async def human_in_the_loop():
agent = Agent(
name="助手",
system_prompt="你是一个有帮助的助手。",
model=model,
toolkit=Toolkit(tools=[
Bash(permission_level=PermissionLevel.HIGH),
]),
)

async for event in agent.reply_stream(
UserMsg("用户", "删除 test.txt 文件")
):
match event.type:
case EventType.REQUIRE_USER_CONFIRM:
# Agent 请求用户确认
print(f"Agent 想要执行: {event.tool_name}")
print(f"参数: {event.arguments}")

# 获取用户输入
confirm = input("是否允许？(y/n): ")

if confirm.lower() == 'y':
# 用户确认，返回确认结果
yield UserConfirmResultEvent(
confirm=True,
tool_call_id=event.tool_call_id,
)
else:
# 用户拒绝
yield UserConfirmResultEvent(
confirm=False,
tool_call_id=event.tool_call_id,
)

case EventType.TEXT_BLOCK_DELTA:
print(event.delta, end="")

九、生产部署：Agent Service

9.1 什么是 Agent Service？

AgentScope 提供了一个生产就绪的 Agent 服务框架，支持：

• 多租户：一个服务实例服务多个用户/组织
• 多会话：用户可以同时进行多个对话
• Web UI：开箱即用的聊天界面
• RESTful API：标准化的 API 接口

9.2 快速部署

Step 1：启动后端服务

# 克隆仓库
git clone https://github.com/agentscope-ai/agentscope

# 进入服务目录
cd agentscope/examples/agent_service

# 启动后端
python main.py

后端服务默认运行在 http://localhost:8000。

Step 2：启动 Web UI

# 打开新终端
cd agentscope/examples/web_ui

# 安装依赖
pnpm install

# 启动前端
pnpm dev

Web UI 默认运行在 http://localhost:5173。

9.3 自定义 Agent 服务

你可以自定义 Agent 服务，添加自己的 Agent：

from fastapi import FastAPI
from agentscope.service import AgentService, AgentConfig
from agentscope.agent import Agent
from agentscope.model import OpenAIChatModel

# 创建 FastAPI 应用
app = FastAPI()

# 创建 Agent 工厂函数
def create_agent(session_id: str, user_id: str) -> Agent:
return Agent(
name="助手",
system_prompt="你是一个有帮助的 AI 助手。",
model=OpenAIChatModel(
model="gpt-4o",
api_key="your-api-key",
),
)

# 创建 Agent 服务配置
config = AgentConfig(
agent_factory=create_agent,
max_sessions_per_user=10, # 每个用户最多 10 个会话
session_timeout=3600, # 会话超时时间（秒）
)

# 注册服务
service = AgentService(config)
app.include_router(service.router)

9.4 API 接口

Agent Service 提供以下 API：

示例：调用 API

import requests

# 创建会话
response = requests.post(
"http://localhost:8000/sessions",
json={"user_id": "user_123"}
)
session_id = response.json()["session_id"]

# 发送消息
response = requests.post(
f"http://localhost:8000/sessions/{session_id}/chat",
json={"message": "你好！"}
)
print(response.json()["reply"])

9.5 Kubernetes 部署

AgentScope 支持在 Kubernetes 集群中部署：

apiVersion: apps/v1
kind: Deployment
metadata:
name: agentscope-service
spec:
replicas: 3
selector:
matchLabels:
app: agentscope
template:
metadata:
labels:
app: agentscope
spec:
containers:
– name: agentscope
image: agentscope/agent-service:latest
ports:
– containerPort: 8000
env:
– name: OPENAI_API_KEY
valueFrom:
secretKeyRef:
name: agentscope-secrets
key: openai-api-key
—
apiVersion: v1
kind: Service
metadata:
name: agentscope-service
spec:
selector:
app: agentscope
ports:
– port: 80
targetPort: 8000
type: LoadBalancer

十、最佳实践

10.1 系统提示词设计

好的系统提示词是 Agent 成功的关键：

# 好的系统提示词示例
system_prompt = """
你是一个专业的 Python 编程助手，名叫 PyBot。

## 你的职责
1. 帮助用户解决 Python 编程问题
2. 提供代码示例和最佳实践建议
3. 解释代码的工作原理

## 你的行为准则
1. 回答要简洁明了，避免冗长
2. 代码示例要完整可运行
3. 如果不确定，诚实地说"我不确定"
4. 对于复杂问题，分步骤解答

## 你的限制
1. 不要执行危险的系统命令
2. 不要访问敏感文件
3. 不要提供有害的代码
"""

10.2 工具设计原则

原则一：单一职责

每个工具应该只做一件事：

# 好的设计：职责单一
def read_file(path: str) -> str:
"""读取文件内容"""
pass

def write_file(path: str, content: str) -> str:
"""写入文件内容"""
pass

# 不好的设计：职责混乱
def file_operation(path: str, content: str = None, mode: str = "read") -> str:
"""文件操作（读或写）"""
pass

原则二：清晰的文档

def search_web(query: str, num_results: int = 5) -> str:
"""
在网络上搜索信息。

这个工具使用搜索引擎搜索与查询相关的网页信息。
返回的结果包括标题、摘要和链接。

参数:
query: 搜索关键词或问题。应该是一个完整的句子或短语，
例如"Python 如何读取 JSON 文件"。
num_results: 返回结果的数量，范围 1-10，默认为 5。

返回:
格式化的搜索结果字符串，每个结果包含：
– 标题
– 摘要
– 链接

示例:
>>> search_web("Python 教程", num_results=3)
"1. Python 官方教程\\n 摘要: …\\n 链接: https://…"
"""
pass

10.3 错误处理

Agent 应该优雅地处理错误：

from agentscope.agent import Agent
from agentscope.exception import AgentException

async def safe_agent_call(agent: Agent, message: str) -> str:
try:
result = await agent.reply(UserMsg("用户", message))
return result.content
except AgentException as e:
return f"Agent 出错了: {e}"
except Exception as e:
return f"未知错误: {e}"

10.4 性能优化

优化一：使用流式输出

# 不推荐：等待完整响应
result = await agent.reply(message)
print(result.content)

# 推荐：流式输出
async for event in agent.reply_stream(message):
if event.type == EventType.TEXT_BLOCK_DELTA:
print(event.delta, end="")

优化二：合理设置上下文压缩

from agentscope.agent import ContextConfig

context_config = ContextConfig(
trigger_ratio=0.7, # 70% 时触发压缩
reserve_ratio=0.3, # 保留最近 30% 的对话
)

优化三：使用批量工具调用

from agentscope.tool import ToolGroup

# 将相关工具组合
file_tools = ToolGroup(
name="file_operations",
tools=[Read(), Write(), Edit()],
)

toolkit = Toolkit(tools=[file_tools])

十一、常见问题与解决方案

11.1 问题：Agent 不断重复调用工具

症状：Agent 在一个循环中不断调用同一个工具。

原因：工具返回的结果格式不正确，Agent 无法理解结果。

解决方案：

# 确保工具返回清晰的结果
def search(query: str) -> str:
"""
搜索信息。

返回格式化的结果，包含明确的结束标记。
"""
results = do_search(query)

# 返回格式化的结果
return f"""
搜索结果：
{format_results(results)}

[搜索完成，共找到 {len(results)} 条结果]
"""

11.2 问题：Agent 不调用工具

症状：Agent 只用文字回复，不调用提供的工具。

原因：

解决方案：

# 1. 改进工具描述
def get_weather(city: str) -> str:
"""
获取天气信息。当用户询问天气相关问题时，必须调用此工具。

参数:
city: 城市名称
"""
pass

# 2. 改进系统提示词
system_prompt = """
你是一个助手。当用户询问天气时，你必须使用 get_weather 工具，
不要编造天气信息。
"""

11.3 问题：上下文过长导致错误

症状：对话进行一段时间后，出现上下文长度超限错误。

解决方案：

from agentscope.agent import Agent, ContextConfig

agent = Agent(
name="助手",
system_prompt="…",
model=model,
context_config=ContextConfig(
trigger_ratio=0.8, # 80% 时触发压缩
reserve_ratio=0.2, # 保留最近 20%
),
)

# 定期压缩上下文
await agent.compress_context()

11.4 问题：工具执行超时

症状：工具执行时间过长，导致请求超时。

解决方案：

import asyncio

async def slow_operation() -> str:
"""可能很慢的操作"""
try:
# 设置超时
result = await asyncio.wait_for(
do_slow_operation(),
timeout=30.0 # 30 秒超时
)
return result
except asyncio.TimeoutError:
return "操作超时，请稍后重试"

十二、总结与展望

12.1 核心要点回顾

通过这篇文章，我们学习了：

12.2 AgentScope vs 其他框架

12.3 参考资源

• GitHub 仓库：https://github.com/agentscope-ai/agentscope
• 官方文档：https://docs.agentscope.io/
• 论文：
  • AgentScope 1.0: A Developer-Centric Framework for Building Agentic Applications
  • AgentScope: A Flexible yet Robust Multi-Agent Platform

附录：快速参考卡片

A. 常用导入

from agentscope.agent import Agent
from agentscope.tool import Toolkit, Bash, Read, Write, Edit, FunctionTool
from agentscope.model import OpenAIChatModel, DashScopeChatModel
from agentscope.message import UserMsg, AssistantMsg, SystemMsg
from agentscope.event import EventType
from agentscope.credential import OpenAICredential, DashScopeCredential

B. 创建 Agent 模板

agent = Agent(
name="助手",
system_prompt="你是一个有帮助的 AI 助手。",
model=OpenAIChatModel(
credential=OpenAICredential(api_key="your-key"),
model="gpt-4o",
),
toolkit=Toolkit(tools=[Bash(), Read(), Write()]),
)

C. 流式处理模板

async for event in agent.reply_stream(UserMsg("用户", "你好")):
match event.type:
case EventType.TEXT_BLOCK_DELTA:
print(event.delta, end="")
case EventType.TOOL_CALL_START:
print(f"\\n[调用工具: {event.tool_name}]")

D. 自定义工具模板

def my_tool(param: str) -> str:
"""
工具描述。

参数:
param: 参数描述

返回:
返回值描述
"""
return f"处理结果: {param}"

toolkit = Toolkit(tools=[FunctionTool(my_tool)])

全文完

希望这篇文章能帮助你理解和使用 AgentScope，打造属于你的 AI 智能体军团！