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

AgentScope 是阿里巴巴推出的轻量级、高扩展性智能体开发框架，核心目标是降低智能体（Agent）开发门槛，支持单智能体快速迭代与多智能体协同调度，尤其在对话交互、任务拆解、工具调用等场景中表现突出。本文将从环境搭建→单智能体开发→多智能体协同→实战案例四个维度，带您快速掌握 AgentScope 的核心用法。

一、前置知识：AgentScope 的核心概念

在实战前，需先理解 AgentScope 的 3 个核心组件，它们是搭建智能体系统的基础：

1. Agent（智能体）：框架的核心执行单元，负责接收输入、决策逻辑（如是否调用工具）、生成输出。支持自定义角色（如 “数据分析师”“客服专员”）与能力（如工具调用、记忆管理）。
2. Tool（工具）：Agent 可调用的外部能力载体，如 API 接口（天气查询、数据查询）、本地函数（文件读写、数据计算）、第三方服务（数据库连接、LLM 模型）。
3. Scope（作用域）：用于管理多智能体的协同规则，定义 Agent 间的通信方式（如 “请求 - 响应”“广播”）、任务分配逻辑（如 “轮询”“指定转发”），确保多智能体有序协作。

二、实战第一步：环境搭建与依赖安装

AgentScope 基于 Python 开发，支持 Windows/macOS/Linux 系统，推荐 Python 3.8+ 版本。

1. 安装 AgentScope

通过 pip 直接安装官方最新版本（截至 2024 年 5 月，稳定版为 0.3.0）：

pip 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(result=result, is_success=True) else: return ToolReturn(result=f"暂不支持查询{city}的天气", is_success=False)

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__(name=name, tools=tools) # 加载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格式：sender=Agent名称，content=回复内容） return Msg(sender=self.name, content=reply_content, receiver=x.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( name="WeatherAssistant", # Agent名称 tools=[weather_tool] # 为Agent绑定工具 ) # 3. 模拟用户输入，启动交互 user_msg = Msg(sender="User", content="查询西安的天气", receiver="WeatherAssistant") 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：定义协同工具与 Agent

（1）数据库查询工具（模拟）

python

from 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(result=mock_data[time_range], is_success=True) else: return ToolReturn(result=f"无{time_range}的销售数据",

（2）三个协同 Agent

python

# 1. 需求理解Agent class RequirementAgent(BaseAgent): def __init__(self, name: str): super().__init__(name=name) 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(sender=self.name, content=query_cmd, receiver="DataQueryAgent") # 2. 数据查询Agent class DataQueryAgent(BaseAgent): def __init__(self, name: str, tools: List[BaseTool]): super().__init__(name=name, tools=tools) 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_range=time_range) # 将数据转发给报告生成Agent return Msg(sender=self.name, content=tool_return.result, receiver="ReportAgent") # 3. 报告生成Agent class ReportAgent(BaseAgent): def __init__(self, name: str): super().__init__(name=name) 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(sender=self.name, content=report, receiver="User"

2. 步骤 2：用 Scope 管理多智能体协同

通过Scope定义 Agent 间的通信流程，启动协同任务：

from agentscope import Scope from agentscope.message import Msg if __name__ == "__main__": # 1. 初始化工具与Agent db_tool = DBQueryTool() req_agent = RequirementAgent(name="RequirementAgent") query_agent = DataQueryAgent(name="DataQueryAgent", tools=[db_tool]) report_agent = ReportAgent(name="ReportAgent") # 2. 创建Scope，注册所有Agent data_analysis_scope = Scope(name="DataAnalysisScope") data_analysis_scope.add_agents([req_agent, query_agent, report_agent]) # 3. 发送用户需求，启动多智能体协同 user_msg = Msg( sender="User", content="我想知道2024年第二季度（4-6月）的销售情况，帮我分析一下", receiver="RequirementAgent" # 需求先发给“需求理解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__(name=name, tools=tools) # 添加短期记忆（保存最近5轮对话） self.memory = ShortTermMemory(max_size=5) 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(name="DynamicAgent", tools=[]) # 动态添加天气查询工具 empty_agent.add_tool(WeatherQueryTool()) # 此时Agent可调用weather_query_tool

六、总结与扩展

通过本文实战，您已掌握 AgentScope 的核心流程：工具定义→Agent 开发→Scope 协同。AgentScope 的优势在于：

• 低门槛：无需从零构建 Agent 通信、记忆管理逻辑；
• 高灵活：支持自定义工具、LLM 对接、协同规则；
• 易扩展：可快速集成到现有系统（如客服平台、数据分析平台）。

扩展方向：

1. 对接真实工具：将模拟工具（天气、数据库）替换为高德天气 API、MySQL 数据库接口；
2. 多模态支持：让 Agent 生成可视化图表；
3. 复杂协同：基于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_success=False) 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(result=result, is_success=True) else: return ToolReturn(result=f"暂不支持查询{city}的天气", is_success=False) # ---------------------- 2. 修正：用json解析LLM输出（替换eval） ---------------------- class WeatherAssistantAgent(BaseAgent): def __init__(self, name: str, tools: List[BaseTool]): super().__init__(name=name, tools=tools) 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_tool=true时填写）, "tool_params": {{"city": "城市名"}}（仅need_tool=true时填写，如{{"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( sender=self.name, content="抱歉，暂时无法理解您的需求，请重新描述城市名称", receiver=x.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( sender=self.name, content="请明确您要查询天气的城市名称（如“查询北京天气”）", receiver=x.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(sender=self.name, content=reply_content, receiver=x.sender) # ---------------------- 3. 修正：run_agent用法（单次交互直接调用reply） ---------------------- if __name__ == "__main__": # 初始化工具和Agent weather_tool = WeatherQueryTool() weather_agent = WeatherAssistantAgent(name="WeatherAssistant", tools=[weather_tool]) # 模拟用户输入（单次交互：直接调用Agent的reply方法） user_msg = Msg(sender="User", content="查询上海的天气", receiver="WeatherAssistant") 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_success=False) 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(result=mock_data[time_range], is_success=True) else: return ToolReturn(result=f"无{time_range}的销售数据", is_success=False) # ---------------------- 2. 三个协同Agent（保留JSON解析优化） ---------------------- class RequirementAgent(BaseAgent): def __init__(self, name: str): super().__init__(name=name) 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(sender=self.name, content=query_cmd, receiver="DataQueryAgent") class DataQueryAgent(BaseAgent): def __init__(self, name: str, tools: List[BaseTool]): super().__init__(name=name, tools=tools) 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_range=time_range) return Msg(sender=self.name, content=tool_return.result, receiver="ReportAgent") class ReportAgent(BaseAgent): def __init__(self, name: str): super().__init__(name=name) 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(sender=self.name, content=report, receiver="User") # ---------------------- 3. 修正：Scope.run() 结果提取（遍历日志找User接收者） ---------------------- if __name__ == "__main__": # 初始化工具和Agent db_tool = DBQueryTool() req_agent = RequirementAgent(name="RequirementAgent") query_agent = DataQueryAgent(name="DataQueryAgent", tools=[db_tool]) report_agent = ReportAgent(name="ReportAgent") # 创建Scope并注册Agent data_analysis_scope = Scope(name="DataAnalysisScope") data_analysis_scope.add_agents([req_agent, query_agent, report_agent]) # 发送用户需求 user_msg = Msg( sender="User", content="我想知道2024年第二季度（4-6月）的销售情况，帮我分析一下", receiver="RequirementAgent" ) # 运行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 "抱歉，数据分析失败，请重试")