智能体开发框架实战（AgentScope）

AgentScope 是阿里巴巴推出的轻量级、高扩展性智能体开发框架核心目标是降低智能体Agent开发门槛支持单智能体快速迭代与多智能体协同调度尤其在对话交互、任务拆解、工具调用等场景中表现突出。本文将从环境搭建→单智能体开发→多智能体协同→实战案例四个维度带您快速掌握 AgentScope 的核心用法。一、前置知识AgentScope 的核心概念在实战前需先理解 AgentScope 的 3 个核心组件它们是搭建智能体系统的基础Agent智能体框架的核心执行单元负责接收输入、决策逻辑如是否调用工具、生成输出。支持自定义角色如 “数据分析师”“客服专员”与能力如工具调用、记忆管理。Tool工具Agent 可调用的外部能力载体如 API 接口天气查询、数据查询、本地函数文件读写、数据计算、第三方服务数据库连接、LLM 模型。Scope作用域用于管理多智能体的协同规则定义 Agent 间的通信方式如 “请求 - 响应”“广播”、任务分配逻辑如 “轮询”“指定转发”确保多智能体有序协作。二、实战第一步环境搭建与依赖安装AgentScope 基于 Python 开发支持 Windows/macOS/Linux 系统推荐 Python 3.8 版本。1. 安装 AgentScope通过 pip 直接安装官方最新版本截至 2024 年 5 月稳定版为 0.3.0pip install agentscope -i https://pypi.org/simple若需开发自定义工具或修改框架源码可从 GitHub 克隆源码安装git clone https://github.com/bytedance/agentscope.git cd agentscope pip install -e .[all] # 安装所有依赖含LLM对接、工具测试等2. 配置基础依赖以 LLM 对接为例AgentScope 默认支持对接主流大模型如 GPT-3.5/4、字节豆包、通义千问需在项目根目录创建config.yaml配置文件指定模型参数以对接 OpenAI GPT-3.5 为例# config.yaml model_configs: gpt-3.5-turbo: type: openai # 模型类型openai/ernie/beanbot等 api_key: your-openai-api-key # 替换为您的API Key model: gpt-3.5-turbo temperature: 0.7 # 生成文本的随机性0~1 max_tokens: 2048 # 最大生成 tokens 数三、实战第二步开发第一个单智能体天气查询助手我们先实现一个简单的单智能体 ——天气查询助手核心能力接收用户的城市输入调用 “天气查询工具”返回格式化天气结果。1. 步骤 1定义 “天气查询工具”首先自定义一个 Tool模拟调用天气 API 的逻辑实际场景可替换为真实天气 API如高德 / 百度天气接口from agentscope.tools import BaseTool, ToolReturn class WeatherQueryTool(BaseTool): # 工具名称需唯一Agent调用时通过名称匹配 name weather_query_tool # 工具描述帮助Agent判断是否需要调用该工具 description 用于查询指定城市的实时天气输入参数为城市名称如西安北京、上海 def _run(self, city: str) - ToolReturn: 工具核心执行逻辑模拟返回天气数据 :param city: 城市名称 :return: ToolReturn 对象含执行结果 # 模拟天气API返回结果实际场景替换为 requests.get 调用 mock_weather { 珠海: 晴22~30℃东北风2级湿度45%, 西安: 多云24~28℃东南风3级湿度60%, 广州: 雷阵雨26~32℃南风1级湿度75% } if city in mock_weather: result f【{city}实时天气】{mock_weather[city]} return ToolReturn(resultresult, is_successTrue) else: return ToolReturn(resultf暂不支持查询{city}的天气, is_successFalse)2. 步骤 2创建 “天气助手 Agent”基于BaseAgent自定义 Agent实现 “接收用户输入→判断是否调用工具→返回结果” 的逻辑from agentscope.agents import BaseAgent from agentscope.message import Msg from typing import List class WeatherAssistantAgent(BaseAgent): def __init__(self, name: str, tools: List[BaseTool]): super().__init__(namename, toolstools) # 加载LLM用于理解用户输入、生成自然语言回复 self.llm self._load_llm(gpt-3.5-turbo) # 对应config.yaml中的模型名 def reply(self, x: Msg) - Msg: Agent 核心交互逻辑接收消息x返回回复Msg user_input x.content # 获取用户输入如“查询北京天气” # 1. 调用LLM判断是否需要调用工具天气查询工具 tool_check_prompt f 用户输入{user_input} 任务判断是否需要调用工具。若需要返回工具名称weather_query_tool和参数city城市名若不需要直接生成回复。 输出格式{{need_tool: True/False, tool_name: xxx, tool_params: {{city: xxx}}}} llm_result self.llm.generate(tool_check_prompt) tool_decision eval(llm_result) # 解析LLM返回的决策实际场景需用JSON解析避免eval风险 # 2. 若需要调用工具执行工具并获取结果 if tool_decision[need_tool]: tool_name tool_decision[tool_name] tool_params tool_decision[tool_params] # 调用工具通过self.tools匹配工具名称 tool_return self.call_tool(tool_name, **tool_params) if tool_return.is_success: reply_content tool_return.result else: reply_content tool_return.result # 3. 若不需要调用工具直接生成回复 else: reply_content self.llm.generate(f用户输入{user_input}请友好回复无需调用工具) # 4. 返回回复消息Msg格式senderAgent名称content回复内容 return Msg(senderself.name, contentreply_content, receiverx.sender)3. 步骤 3运行单智能体并测试from agentscope import run_agent from agentscope.message import Msg if __name__ __main__: # 1. 初始化工具 weather_tool WeatherQueryTool() # 2. 初始化天气助手Agent weather_agent WeatherAssistantAgent( nameWeatherAssistant, # Agent名称 tools[weather_tool] # 为Agent绑定工具 ) # 3. 模拟用户输入启动交互 user_msg Msg(senderUser, content查询西安的天气, receiverWeatherAssistant) result run_agent(weather_agent, user_msg) # 4. 打印结果 print(f\n{result.sender} 回复{result.content})运行结果模拟示例WeatherAssistant 回复【西安实时天气】多云24~28℃东南风3级湿度60%以上是模拟实际业务根据需求修改天气查询接口即可四、实战第三步多智能体协同数据分析团队单智能体仅能处理单一任务多智能体协同可拆解复杂任务如 “用户需求→数据查询→数据分析→报告生成”。下面实现一个 “数据分析团队”包含 3 个 Agent需求理解 Agent接收用户数据分析需求拆解为 “数据查询指令”数据查询 Agent调用 “数据库工具”获取原始数据报告生成 Agent基于原始数据生成可视化分析报告模拟。1. 步骤 1定义协同工具与 Agent1数据库查询工具模拟pythonfrom agentscope.tools import BaseTool, ToolReturn class DBQueryTool(BaseTool): name db_query_tool description 用于查询销售数据输入参数为时间范围如2024-01至2024-03 def _run(self, time_range: str) - ToolReturn: # 模拟数据库返回销售数据 mock_data { 2024-01至2024-03: 1月50万2月65万3月72万, 2024-04至2024-06: 4月80万5月78万6月92万 } if time_range in mock_data: return ToolReturn(resultmock_data[time_range], is_successTrue) else: return ToolReturn(resultf无{time_range}的销售数据,2三个协同 Agentpython# 1. 需求理解Agent class RequirementAgent(BaseAgent): def __init__(self, name: str): super().__init__(namename) self.llm self._load_llm(gpt-3.5-turbo) def reply(self, x: Msg) - Msg: # 拆解用户需求为数据查询指令 prompt f用户需求{x.content}请提取时间范围格式YYYY-MM至YYYY-MM输出查询指令如“查询2024-01至2024-03销售数据” query_cmd self.llm.generate(prompt) # 将指令转发给数据查询Agent return Msg(senderself.name, contentquery_cmd, receiverDataQueryAgent) # 2. 数据查询Agent class DataQueryAgent(BaseAgent): def __init__(self, name: str, tools: List[BaseTool]): super().__init__(namename, toolstools) self.llm self._load_llm(gpt-3.5-turbo) def reply(self, x: Msg) - Msg: # 从需求指令中提取时间范围 prompt f指令{x.content}提取时间范围仅返回“YYYY-MM至YYYY-MM”格式 time_range self.llm.generate(prompt).strip() # 调用数据库工具 tool_return self.call_tool(db_query_tool, time_rangetime_range) # 将数据转发给报告生成Agent return Msg(senderself.name, contenttool_return.result, receiverReportAgent) # 3. 报告生成Agent class ReportAgent(BaseAgent): def __init__(self, name: str): super().__init__(namename) self.llm self._load_llm(gpt-3.5-turbo) def reply(self, x: Msg) - Msg: # 基于原始数据生成分析报告 prompt f销售数据{x.content}请生成简洁的分析报告含趋势、峰值、建议格式清晰 report self.llm.generate(prompt) # 将报告返回给用户 return Msg(senderself.name, contentreport, receiverUser2. 步骤 2用 Scope 管理多智能体协同通过Scope定义 Agent 间的通信流程启动协同任务from agentscope import Scope from agentscope.message import Msg if __name__ __main__: # 1. 初始化工具与Agent db_tool DBQueryTool() req_agent RequirementAgent(nameRequirementAgent) query_agent DataQueryAgent(nameDataQueryAgent, tools[db_tool]) report_agent ReportAgent(nameReportAgent) # 2. 创建Scope注册所有Agent data_analysis_scope Scope(nameDataAnalysisScope) data_analysis_scope.add_agents([req_agent, query_agent, report_agent]) # 3. 发送用户需求启动多智能体协同 user_msg Msg( senderUser, content我想知道2024年第二季度4-6月的销售情况帮我分析一下, receiverRequirementAgent # 需求先发给“需求理解Agent” ) # 4. 运行Scope获取最终结果 final_result data_analysis_scope.run(user_msg) # 5. 打印最终报告 print(\n 数据分析报告 ) print(final_result.content)五、实战进阶AgentScope 核心特性应用1. 记忆管理让 Agent 记住历史对话AgentScope 支持ShortTermMemory和LongTermMemory只需在 Agent 初始化时添加记忆组件from agentscope.memory import ShortTermMemory class MemoryWeatherAgent(WeatherAssistantAgent): def __init__(self, name: str, tools: List[BaseTool]): super().__init__(namename, toolstools) # 添加短期记忆保存最近5轮对话 self.memory ShortTermMemory(max_size5) def reply(self, x: Msg) - Msg: # 将当前用户消息存入记忆 self.memory.add(x) # 从记忆中获取历史对话用于上下文理解 history self.memory.get_memory() history_str \n.join([f{msg.sender}: {msg.content} for msg in history]) # 调用LLM时传入历史对话 tool_check_prompt f历史对话{history_str}\n当前用户输入{x.content}\n判断是否需要调用工具... # 后续逻辑与之前一致...2. 工具动态绑定根据任务临时添加工具支持在运行时为 Agent 动态添加工具无需重启 Agent# 初始化一个空工具的Agent empty_agent BaseAgent(nameDynamicAgent, tools[]) # 动态添加天气查询工具 empty_agent.add_tool(WeatherQueryTool()) # 此时Agent可调用weather_query_tool六、总结与扩展通过本文实战您已掌握 AgentScope 的核心流程工具定义→Agent 开发→Scope 协同。AgentScope 的优势在于低门槛无需从零构建 Agent 通信、记忆管理逻辑高灵活支持自定义工具、LLM 对接、协同规则易扩展可快速集成到现有系统如客服平台、数据分析平台。扩展方向对接真实工具将模拟工具天气、数据库替换为高德天气 API、MySQL 数据库接口多模态支持让 Agent 生成可视化图表复杂协同基于Scope定义更复杂的任务流如 “异常数据→自动告警→人工介入”。初学者欢迎大家指点批评指正。# 单智能体开发(天气助手) from agentscope.tools import BaseTool, ToolReturn from agentscope.agents import BaseAgent from agentscope.message import Msg from typing import List import json # 导入json模块替换eval # ---------------------- 1. 修正工具参数类型校验 ---------------------- class WeatherQueryTool(BaseTool): name weather_query_tool description 用于查询指定城市的实时天气输入参数为城市名称如北京、上海 def _run(self, city: str) - ToolReturn: # 新增参数类型校验 if not isinstance(city, str): return ToolReturn(result城市名称必须是字符串如“北京”, is_successFalse) mock_weather { 北京: 晴22~30℃东北风2级湿度45%, 上海: 多云24~28℃东南风3级湿度60%, 广州: 雷阵雨26~32℃南风1级湿度75% } if city in mock_weather: result f【{city}实时天气】{mock_weather[city]} return ToolReturn(resultresult, is_successTrue) else: return ToolReturn(resultf暂不支持查询{city}的天气, is_successFalse) # ---------------------- 2. 修正用json解析LLM输出替换eval ---------------------- class WeatherAssistantAgent(BaseAgent): def __init__(self, name: str, tools: List[BaseTool]): super().__init__(namename, toolstools) self.llm self._load_llm(gpt-3.5-turbo) def reply(self, x: Msg) - Msg: user_input x.content # 优化Prompt强制LLM输出标准JSON无多余内容 tool_check_prompt f 任务判断是否需要调用工具查询天气严格按以下格式输出JSON禁止任何多余文字 {{ need_tool: true/false, // 需要调用工具为true否则为false tool_name: weather_query_tool仅need_tooltrue时填写, tool_params: {{city: 城市名}}仅need_tooltrue时填写如{{city: 北京}} }} 用户输入{user_input} llm_result self.llm.generate(tool_check_prompt).strip() # 新增异常捕获避免LLM输出非JSON格式 try: tool_decision json.loads(llm_result) except json.JSONDecodeError: return Msg( senderself.name, content抱歉暂时无法理解您的需求请重新描述城市名称, receiverx.sender ) if tool_decision.get(need_tool, False): tool_name tool_decision.get(tool_name) tool_params tool_decision.get(tool_params, {}) # 校验工具参数是否完整 if not tool_params.get(city): return Msg( senderself.name, content请明确您要查询天气的城市名称如“查询北京天气”, receiverx.sender ) # 调用工具 tool_return self.call_tool(tool_name, **tool_params) reply_content tool_return.result else: reply_content self.llm.generate(f用户输入{user_input}友好回复无需调用工具) return Msg(senderself.name, contentreply_content, receiverx.sender) # ---------------------- 3. 修正run_agent用法单次交互直接调用reply ---------------------- if __name__ __main__: # 初始化工具和Agent weather_tool WeatherQueryTool() weather_agent WeatherAssistantAgent(nameWeatherAssistant, tools[weather_tool]) # 模拟用户输入单次交互直接调用Agent的reply方法 user_msg Msg(senderUser, content查询上海的天气, receiverWeatherAssistant) result weather_agent.reply(user_msg) # 修正替换run_agent直接调用reply # 打印结果 print(f\n{result.sender} 回复{result.content})# 多智能体数据分析团队 from agentscope.tools import BaseTool, ToolReturn from agentscope.agents import BaseAgent from agentscope import Scope from agentscope.message import Msg from typing import List import json # ---------------------- 1. 数据库工具保留参数校验 ---------------------- class DBQueryTool(BaseTool): name db_query_tool description 用于查询销售数据输入参数为时间范围如2024-01至2024-03 def _run(self, time_range: str) - ToolReturn: if not isinstance(time_range, str): return ToolReturn(result时间范围必须是字符串如“2024-01至2024-03”, is_successFalse) mock_data { 2024-01至2024-03: 1月50万2月65万3月72万, 2024-04至2024-06: 4月80万5月78万6月92万 } if time_range in mock_data: return ToolReturn(resultmock_data[time_range], is_successTrue) else: return ToolReturn(resultf无{time_range}的销售数据, is_successFalse) # ---------------------- 2. 三个协同Agent保留JSON解析优化 ---------------------- class RequirementAgent(BaseAgent): def __init__(self, name: str): super().__init__(namename) self.llm self._load_llm(gpt-3.5-turbo) def reply(self, x: Msg) - Msg: prompt f 任务从用户需求中提取时间范围严格按以下格式输出仅返回时间范围无多余文字 YYYY-MM至YYYY-MM如“2024-04至2024-06” 用户需求{x.content} time_range self.llm.generate(prompt).strip() # 生成数据查询指令 query_cmd f查询{time_range}销售数据 return Msg(senderself.name, contentquery_cmd, receiverDataQueryAgent) class DataQueryAgent(BaseAgent): def __init__(self, name: str, tools: List[BaseTool]): super().__init__(namename, toolstools) self.llm self._load_llm(gpt-3.5-turbo) def reply(self, x: Msg) - Msg: prompt f 任务从指令中提取时间范围仅返回“YYYY-MM至YYYY-MM”格式无多余文字 指令{x.content} time_range self.llm.generate(prompt).strip() # 调用工具 tool_return self.call_tool(db_query_tool, time_rangetime_range) return Msg(senderself.name, contenttool_return.result, receiverReportAgent) class ReportAgent(BaseAgent): def __init__(self, name: str): super().__init__(namename) self.llm self._load_llm(gpt-3.5-turbo) def reply(self, x: Msg) - Msg: prompt f 任务基于销售数据生成分析报告包含“原始数据、趋势分析、关键结论、建议”4部分格式清晰 销售数据{x.content} report self.llm.generate(prompt) # 明确接收者为User用于后续提取最终结果 return Msg(senderself.name, contentreport, receiverUser) # ---------------------- 3. 修正Scope.run() 结果提取遍历日志找User接收者 ---------------------- if __name__ __main__: # 初始化工具和Agent db_tool DBQueryTool() req_agent RequirementAgent(nameRequirementAgent) query_agent DataQueryAgent(nameDataQueryAgent, tools[db_tool]) report_agent ReportAgent(nameReportAgent) # 创建Scope并注册Agent data_analysis_scope Scope(nameDataAnalysisScope) data_analysis_scope.add_agents([req_agent, query_agent, report_agent]) # 发送用户需求 user_msg Msg( senderUser, content我想知道2024年第二季度4-6月的销售情况帮我分析一下, receiverRequirementAgent ) # 运行Scope返回所有交互日志需提取最终给User的报告 interaction_logs data_analysis_scope.run(user_msg) # 修正遍历日志找到接收者为User的消息即最终报告 final_report None for msg in interaction_logs: if msg.receiver User: final_report msg.content break # 打印结果 print(\n 数据分析报告 ) print(final_report if final_report else 抱歉数据分析失败请重试)