六、多 Agent 协作6.1 为什么需要多 Agent单个 Agent 虽然强大但在复杂场景下往往力不从心。想象一下软件开发需要产品经理、架构师、程序员、测试工程师协作投资分析需要行业研究员、财务分析师、风险评估师协作内容创作需要策划、撰稿、编辑、审核协作这就是多 Agent 协作的价值所在——让多个专业化的 Agent 组成团队共同完成复杂任务。6.2 多 Agent 协作模式AgentScope 支持多种多 Agent 协作模式模式一顺序执行SequentialAgent 按顺序依次执行前一个 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你是一个内容策划专家负责规划文章大纲和结构。, modelmodel, ) writer Agent( name撰稿, system_prompt你是一个专业撰稿人根据大纲撰写文章内容。, modelmodel, ) editor Agent( name编辑, system_prompt你是一个资深编辑负责审核和优化文章。, modelmodel, ) # 顺序执行 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你从技术角度分析问题。, modelmodel, ) business_analyst Agent( name商业分析师, system_prompt你从商业角度分析问题。, modelmodel, ) risk_analyst Agent( name风险分析师, system_prompt你从风险角度分析问题。, modelmodel, ) 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你负责收集和整理信息。, modelmodel, ) developer Agent( name开发者, system_prompt你负责编写代码和实现功能。, modelmodel, toolkitToolkit(tools[ FunctionTool(assign_task) ]) ) tester Agent( name测试员, system_prompt你负责测试和质量保证。, modelmodel, ) # 创建管理者 Agent manager Agent( name项目经理, system_prompt你是一个项目经理负责协调团队成员完成任务。 你可以分配任务给以下成员 - 研究员负责信息收集 - 开发者负责代码实现 - 测试员负责测试验证 , modelmodel, ) async def hierarchical_workflow(): task 开发一个简单的待办事项应用 result await manager.reply( UserMsg(用户, f请组织团队完成以下任务{task}) ) print(result.content) asyncio.run(hierarchical_workflow())6.3 消息中心MsgHubAgentScope 提供了消息中心MsgHub来实现灵活的多 Agent 通信。MsgHub 的作用MsgHub 是一个消息广播机制允许多个 Agent 订阅和发布消息┌─────────────────────────────────────────────────────────────┐ │ MsgHub │ ├─────────────────────────────────────────────────────────────┤ │ │ │ Agent A ──────┐ │ │ │ │ │ Agent B ──────┼──────→ 消息队列 ──────→ 广播给所有订阅者 │ │ │ │ │ Agent C ──────┘ │ │ │ └─────────────────────────────────────────────────────────────┘使用 MsgHubimport 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( nameAgent A, system_prompt你是 Agent A负责分析问题。, modelmodel, ) hub.subscribe(agent_a) agent_b Agent( nameAgent B, system_prompt你是 Agent B负责提供建议。, modelmodel, ) 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结果 {i1}: 包含 {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( contentresult, 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_levelPermissionLevel.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( typeimage, sourceBase64Source( media_typeimage/png, database64.b64encode(image_data).decode() ) ) ] )7.4 MCP 工具集成AgentScope 支持MCPModel Context Protocol可以轻松集成外部 MCP 服务器提供的工具from agentscope.tool import MCPTool # 连接到 MCP 服务器 mcp_tool MCPTool( server_urlhttp://localhost:8080/mcp, tool_namedatabase_query, ) toolkit Toolkit(tools[mcp_tool])八、权限控制与安全8.1 为什么需要权限控制当 Agent 可以执行 Shell 命令、读写文件、调用 API 时安全问题变得至关重要数据泄露Agent 可能读取敏感文件系统破坏Agent 可能执行危险的 Shell 命令资源滥用Agent 可能调用昂贵的 API8.2 AgentScope 的权限系统AgentScope 提供了细粒度的权限控制机制权限级别级别说明LOW低风险操作如读取公开文件MEDIUM中等风险操作如写入文件HIGH高风险操作如删除文件、执行命令权限行为行为说明ALLOW允许执行无需确认DENY拒绝执行ASK询问用户是否允许配置权限from agentscope.agent import Agent from agentscope.permission import PermissionBehavior, PermissionLevel from agentscope.tool import Toolkit, Bash, Read, Write # 配置权限行为 permission_behavior PermissionBehavior( # 低权限操作自动允许 lowPermissionBehavior.ALLOW, # 中权限操作询问用户 mediumPermissionBehavior.ASK, # 高权限操作拒绝 highPermissionBehavior.DENY, ) agent Agent( name助手, system_prompt你是一个有帮助的助手。, modelmodel, toolkitToolkit(tools[ Bash(permission_levelPermissionLevel.HIGH), # 高权限 Read(permission_levelPermissionLevel.LOW), # 低权限 Write(permission_levelPermissionLevel.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你是一个有帮助的助手。, modelmodel, toolkitToolkit(tools[ Bash(permission_levelPermissionLevel.HIGH), ]), ) async for event in agent.reply_stream( UserMsg(用户, 删除 test.txt 文件) ): match event.type: case EventType.REQUIRE_USER_CONFIRM: # Agent 请求用户确认 print(fAgent 想要执行: {event.tool_name}) print(f参数: {event.arguments}) # 获取用户输入 confirm input(是否允许(y/n): ) if confirm.lower() y: # 用户确认返回确认结果 yield UserConfirmResultEvent( confirmTrue, tool_call_idevent.tool_call_id, ) else: # 用户拒绝 yield UserConfirmResultEvent( confirmFalse, tool_call_idevent.tool_call_id, ) case EventType.TEXT_BLOCK_DELTA: print(event.delta, end)九、生产部署Agent Service9.1 什么是 Agent ServiceAgentScope 提供了一个生产就绪的 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 devWeb UI 默认运行在http://localhost:5173。9.3 自定义 Agent 服务你可以自定义 Agent 服务添加自己的 Agentfrom 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 助手。, modelOpenAIChatModel( modelgpt-4o, api_keyyour-api-key, ), ) # 创建 Agent 服务配置 config AgentConfig( agent_factorycreate_agent, max_sessions_per_user10, # 每个用户最多 10 个会话 session_timeout3600, # 会话超时时间秒 ) # 注册服务 service AgentService(config) app.include_router(service.router)9.4 API 接口Agent Service 提供以下 API接口方法说明/sessionsPOST创建新会话/sessions/{id}GET获取会话信息/sessions/{id}/chatPOST发送消息/sessions/{id}/chat/streamPOST流式发送消息示例调用 APIimport requests # 创建会话 response requests.post( http://localhost:8000/sessions, json{user_id: user_123} ) session_id response.json()[session_id] # 发送消息 response requests.post( fhttp://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_results3) 1. Python 官方教程\n 摘要: ...\n 链接: https://... pass10.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 fAgent 出错了: {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_ratio0.7, # 70% 时触发压缩 reserve_ratio0.3, # 保留最近 30% 的对话 )优化三使用批量工具调用from agentscope.tool import ToolGroup # 将相关工具组合 file_tools ToolGroup( namefile_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..., modelmodel, context_configContextConfig( trigger_ratio0.8, # 80% 时触发压缩 reserve_ratio0.2, # 保留最近 20% ), ) # 定期压缩上下文 await agent.compress_context()11.4 问题工具执行超时症状工具执行时间过长导致请求超时。解决方案import asyncio async def slow_operation() - str: 可能很慢的操作 try: # 设置超时 result await asyncio.wait_for( do_slow_operation(), timeout30.0 # 30 秒超时 ) return result except asyncio.TimeoutError: return 操作超时请稍后重试十二、总结与展望12.1 核心要点回顾通过这篇文章我们学习了什么是 AI Agent能够感知环境、做出决策、执行动作的自主实体AgentScope 是什么阿里巴巴开源的生产就绪的 Agent 框架核心概念Agent、Tool、Message、Event、Model、State、Middleware如何使用安装、创建 Agent、添加工具、处理事件流多 Agent 协作顺序、并行、层级模式以及 MsgHub工具开发FunctionTool、ToolBase、MCP 集成权限控制权限级别、人机协作生产部署Agent Service、Kubernetes 部署12.2 AgentScope vs 其他框架特性AgentScopeLangChainCrewAI学习曲线低中中生产就绪✅⚠️⚠️多租户支持✅❌❌内置 Web UI✅❌❌流式输出✅✅⚠️权限控制✅❌❌MCP 支持✅⚠️❌12.3 参考资源GitHub 仓库https://github.com/agentscope-ai/agentscope官方文档https://docs.agentscope.io/论文AgentScope 1.0: A Developer-Centric Framework for Building Agentic ApplicationsAgentScope: 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, DashScopeCredentialB. 创建 Agent 模板agent Agent( name助手, system_prompt你是一个有帮助的 AI 助手。, modelOpenAIChatModel( credentialOpenAICredential(api_keyyour-key), modelgpt-4o, ), toolkitToolkit(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 智能体军团
深入浅出 AgentScope 2.0：打造你的 AI 智能体军团（下篇）
发布时间：2026/5/28 15:19:52
六、多 Agent 协作6.1 为什么需要多 Agent单个 Agent 虽然强大但在复杂场景下往往力不从心。想象一下软件开发需要产品经理、架构师、程序员、测试工程师协作投资分析需要行业研究员、财务分析师、风险评估师协作内容创作需要策划、撰稿、编辑、审核协作这就是多 Agent 协作的价值所在——让多个专业化的 Agent 组成团队共同完成复杂任务。6.2 多 Agent 协作模式AgentScope 支持多种多 Agent 协作模式模式一顺序执行SequentialAgent 按顺序依次执行前一个 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你是一个内容策划专家负责规划文章大纲和结构。, modelmodel, ) writer Agent( name撰稿, system_prompt你是一个专业撰稿人根据大纲撰写文章内容。, modelmodel, ) editor Agent( name编辑, system_prompt你是一个资深编辑负责审核和优化文章。, modelmodel, ) # 顺序执行 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你从技术角度分析问题。, modelmodel, ) business_analyst Agent( name商业分析师, system_prompt你从商业角度分析问题。, modelmodel, ) risk_analyst Agent( name风险分析师, system_prompt你从风险角度分析问题。, modelmodel, ) 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你负责收集和整理信息。, modelmodel, ) developer Agent( name开发者, system_prompt你负责编写代码和实现功能。, modelmodel, toolkitToolkit(tools[ FunctionTool(assign_task) ]) ) tester Agent( name测试员, system_prompt你负责测试和质量保证。, modelmodel, ) # 创建管理者 Agent manager Agent( name项目经理, system_prompt你是一个项目经理负责协调团队成员完成任务。 你可以分配任务给以下成员 - 研究员负责信息收集 - 开发者负责代码实现 - 测试员负责测试验证 , modelmodel, ) async def hierarchical_workflow(): task 开发一个简单的待办事项应用 result await manager.reply( UserMsg(用户, f请组织团队完成以下任务{task}) ) print(result.content) asyncio.run(hierarchical_workflow())6.3 消息中心MsgHubAgentScope 提供了消息中心MsgHub来实现灵活的多 Agent 通信。MsgHub 的作用MsgHub 是一个消息广播机制允许多个 Agent 订阅和发布消息┌─────────────────────────────────────────────────────────────┐ │ MsgHub │ ├─────────────────────────────────────────────────────────────┤ │ │ │ Agent A ──────┐ │ │ │ │ │ Agent B ──────┼──────→ 消息队列 ──────→ 广播给所有订阅者 │ │ │ │ │ Agent C ──────┘ │ │ │ └─────────────────────────────────────────────────────────────┘使用 MsgHubimport 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( nameAgent A, system_prompt你是 Agent A负责分析问题。, modelmodel, ) hub.subscribe(agent_a) agent_b Agent( nameAgent B, system_prompt你是 Agent B负责提供建议。, modelmodel, ) 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结果 {i1}: 包含 {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( contentresult, 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_levelPermissionLevel.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( typeimage, sourceBase64Source( media_typeimage/png, database64.b64encode(image_data).decode() ) ) ] )7.4 MCP 工具集成AgentScope 支持MCPModel Context Protocol可以轻松集成外部 MCP 服务器提供的工具from agentscope.tool import MCPTool # 连接到 MCP 服务器 mcp_tool MCPTool( server_urlhttp://localhost:8080/mcp, tool_namedatabase_query, ) toolkit Toolkit(tools[mcp_tool])八、权限控制与安全8.1 为什么需要权限控制当 Agent 可以执行 Shell 命令、读写文件、调用 API 时安全问题变得至关重要数据泄露Agent 可能读取敏感文件系统破坏Agent 可能执行危险的 Shell 命令资源滥用Agent 可能调用昂贵的 API8.2 AgentScope 的权限系统AgentScope 提供了细粒度的权限控制机制权限级别级别说明LOW低风险操作如读取公开文件MEDIUM中等风险操作如写入文件HIGH高风险操作如删除文件、执行命令权限行为行为说明ALLOW允许执行无需确认DENY拒绝执行ASK询问用户是否允许配置权限from agentscope.agent import Agent from agentscope.permission import PermissionBehavior, PermissionLevel from agentscope.tool import Toolkit, Bash, Read, Write # 配置权限行为 permission_behavior PermissionBehavior( # 低权限操作自动允许 lowPermissionBehavior.ALLOW, # 中权限操作询问用户 mediumPermissionBehavior.ASK, # 高权限操作拒绝 highPermissionBehavior.DENY, ) agent Agent( name助手, system_prompt你是一个有帮助的助手。, modelmodel, toolkitToolkit(tools[ Bash(permission_levelPermissionLevel.HIGH), # 高权限 Read(permission_levelPermissionLevel.LOW), # 低权限 Write(permission_levelPermissionLevel.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你是一个有帮助的助手。, modelmodel, toolkitToolkit(tools[ Bash(permission_levelPermissionLevel.HIGH), ]), ) async for event in agent.reply_stream( UserMsg(用户, 删除 test.txt 文件) ): match event.type: case EventType.REQUIRE_USER_CONFIRM: # Agent 请求用户确认 print(fAgent 想要执行: {event.tool_name}) print(f参数: {event.arguments}) # 获取用户输入 confirm input(是否允许(y/n): ) if confirm.lower() y: # 用户确认返回确认结果 yield UserConfirmResultEvent( confirmTrue, tool_call_idevent.tool_call_id, ) else: # 用户拒绝 yield UserConfirmResultEvent( confirmFalse, tool_call_idevent.tool_call_id, ) case EventType.TEXT_BLOCK_DELTA: print(event.delta, end)九、生产部署Agent Service9.1 什么是 Agent ServiceAgentScope 提供了一个生产就绪的 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 devWeb UI 默认运行在http://localhost:5173。9.3 自定义 Agent 服务你可以自定义 Agent 服务添加自己的 Agentfrom 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 助手。, modelOpenAIChatModel( modelgpt-4o, api_keyyour-api-key, ), ) # 创建 Agent 服务配置 config AgentConfig( agent_factorycreate_agent, max_sessions_per_user10, # 每个用户最多 10 个会话 session_timeout3600, # 会话超时时间秒 ) # 注册服务 service AgentService(config) app.include_router(service.router)9.4 API 接口Agent Service 提供以下 API接口方法说明/sessionsPOST创建新会话/sessions/{id}GET获取会话信息/sessions/{id}/chatPOST发送消息/sessions/{id}/chat/streamPOST流式发送消息示例调用 APIimport requests # 创建会话 response requests.post( http://localhost:8000/sessions, json{user_id: user_123} ) session_id response.json()[session_id] # 发送消息 response requests.post( fhttp://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_results3) 1. Python 官方教程\n 摘要: ...\n 链接: https://... pass10.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 fAgent 出错了: {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_ratio0.7, # 70% 时触发压缩 reserve_ratio0.3, # 保留最近 30% 的对话 )优化三使用批量工具调用from agentscope.tool import ToolGroup # 将相关工具组合 file_tools ToolGroup( namefile_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..., modelmodel, context_configContextConfig( trigger_ratio0.8, # 80% 时触发压缩 reserve_ratio0.2, # 保留最近 20% ), ) # 定期压缩上下文 await agent.compress_context()11.4 问题工具执行超时症状工具执行时间过长导致请求超时。解决方案import asyncio async def slow_operation() - str: 可能很慢的操作 try: # 设置超时 result await asyncio.wait_for( do_slow_operation(), timeout30.0 # 30 秒超时 ) return result except asyncio.TimeoutError: return 操作超时请稍后重试十二、总结与展望12.1 核心要点回顾通过这篇文章我们学习了什么是 AI Agent能够感知环境、做出决策、执行动作的自主实体AgentScope 是什么阿里巴巴开源的生产就绪的 Agent 框架核心概念Agent、Tool、Message、Event、Model、State、Middleware如何使用安装、创建 Agent、添加工具、处理事件流多 Agent 协作顺序、并行、层级模式以及 MsgHub工具开发FunctionTool、ToolBase、MCP 集成权限控制权限级别、人机协作生产部署Agent Service、Kubernetes 部署12.2 AgentScope vs 其他框架特性AgentScopeLangChainCrewAI学习曲线低中中生产就绪✅⚠️⚠️多租户支持✅❌❌内置 Web UI✅❌❌流式输出✅✅⚠️权限控制✅❌❌MCP 支持✅⚠️❌12.3 参考资源GitHub 仓库https://github.com/agentscope-ai/agentscope官方文档https://docs.agentscope.io/论文AgentScope 1.0: A Developer-Centric Framework for Building Agentic ApplicationsAgentScope: 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, DashScopeCredentialB. 创建 Agent 模板agent Agent( name助手, system_prompt你是一个有帮助的 AI 助手。, modelOpenAIChatModel( credentialOpenAICredential(api_keyyour-key), modelgpt-4o, ), toolkitToolkit(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 智能体军团
