AgentScope源码解剖:Agent框架透视

一、AgentScope是什么？

AgentScope 是阿里巴巴通义实验室开源的多智能体（Multi-Agent）开发框架，核心定位是让多智能体应用从"能跑通"到"能上线"——它不只是提供几个 Agent 聊天的 Demo，而是围绕工程化落地做了完整的工具链和运行时设计。解决的是"多个 AI 怎么协同干活"的工程问题——它不只是让 Agent 能对话，而是提供了让这套对话系统稳定、可维护、可扩展的完整基础设施。

二、为什么选AgentScope？

不是因为它最火，而是因为它最"稳";

AgentScope的设计哲学非常明确：工程化。

它不追求最炫的对话涌现效果，也不追求最短的入门代码，它追求的是——当10个Agent同时在线、当某个模型API突然超时、当工具调用返回异常格式、当Token消耗即将爆表时，你的系统不会崩。

根据官方定位，AgentScope是"面向生产环境、易于上手的智能体框架"。

它旨在解决"协调智能体合作以及大型语言模型性能不稳定所带来的挑战"，提供了"内置和可定制的容错机制"以及"基于actor的分布式框架"。

换句话说，AutoGen是实验室里的天才，LangGraph是图纸上的建筑师，而AgentScope是工地上的项目经理——它知道怎么让一群"天才"在deadline前把楼盖完。

先解析一下agentScope的三个先进性：

1.消息驱动架构=天然并发友好

不同于AutoGen的"对话即工作流"（容易聊飞），也不同于LangGraph的"状态机"（需要预先定义所有分支），AgentScope选择了消息交换作为核心通信机制。每个Agent通过标准化的Msg对象通信，这意味着：你可以随时插入新的Agent而不破坏现有流程，消息可以被追踪、重放、持久化，天然支持异步和分布式部署

2.组合式架构=可维护性

AgentScope没有采用继承式的设计（"我的Agent继承自ReActAgent"），而是组合式架构：Agent由Model、Formatter、Toolkit、Memory等模块组合而成。

这在企业级代码库中意味着——你可以独立升级模型层而不影响业务逻辑，可以替换记忆后端而不改Agent行为。

3.从POC到工程化的无缝切换

AgentScope内置了OpenTelemetry追踪、多存储后端（JSON/Redis/TableStore）、Token计数与成本预估、Ray分布式评估。这些不是"锦上添花"，而是从第一天就考虑到了上线后的运维需求。

三、它能干什么？

不是"功能列表"，而是"能力矩阵"，打开AgentScope的源码目录，你会看到18个一级模块。但别被数量吓到，它们不是简单的功能堆砌，而是围绕"智能体生命周期"的完整闭环：

感知→推理→行动→记忆→评估→优化

1.智能体核心：ReActAgent的"洋葱模型"

src/agentscope/agent/目录下，_react_agent.py是框架的心脏。它的执行流程是一个经典的Reasoning-Acting循环：

reply(msg)

├─►memory.add(msg)#记录输入

├─►_retrieve_from_long_term_memory()#长期记忆检索

├─►_retrieve_from_knowledge()#RAG知识库检索

├─►[循环max_iters次]

│├─►_compress_memory_if_needed()#记忆压缩（Token超限触发）

│├─►_reasoning(tool_choice)#调用LLM推理

│├─►[_acting(tool_call)...]#执行工具调用（可并行）

│└─►检查退出条件

└─►返回reply_msg

这个流程的精妙之处在于"防御性设计"：

记忆压缩：当对话历史超过Token阈值，自动用LLM生成结构化摘要，保留最近N条原始消息。这不是"优化项"，而是"保命项"——没有它，长对话必然触发模型上下文超限。

工具调用并行化：_acting支持并行执行多个工具，这在需要同时查数据库、查知识图谱、调API、向量检索、读文件的场景下，能显著降低延迟。

流式输出：stream=True时，推理过程可以实时返回，这对用户体验至关重要。

2.模型抽象层：一次接入，到处使用

src/agentscope/model/目录实现了统一的ChatModelBase接口：

classChatModelBase:

@abstractmethod

asyncdef__call__(

self,

messages:list[dict],

tools:list[dict]|None=None,

tool_choice:str|None=None,

structured_model:Type[BaseModel]|None=None,

)->ChatResponse|AsyncGenerator[ChatResponse,None]:

目前支持OpenAI、Anthropic、通义千问、Gemini、Ollama等主流模型，以及实时语音模型（OpenAIRealtime、DashScopeRealtime）。

更值得关注的是Formatter模块——它负责将框架内部的Msg对象转换为各厂商API所需的格式。

这意味着多模态内容（图片、音频、视频）的转换逻辑被集中管理，而不是散落在每个Agent的业务代码里。

3.工具系统：不仅是"调用函数"，而是"管理生态"，从普通的工具函数到MCP，再到skills，智能体的手脚接入均支持

src/agentscope/tool/目录下的Toolkit是框架的工程化亮点：

注册机制：

#注册函数工具

toolkit.register_tool_function(tool_func=my_func,group_name="group")

#注册MCP工具（标准化协议）

awaittoolkit.register_mcp_client(mcp_client=client)

#注册技能（Skill）——可复用的Agent技能包

toolkit.register_agent_skill(skill_dir="/path/to/skill")

动态调整：

toolkit.reset_equipped_tools(agent,tools_info)#运行时调整Agent可用工具

这意味着你可以构建一个"工具市场"：不同Agent根据当前任务动态加载不同的工具集，而不是一次性加载所有工具导致上下文膨胀。

4.记忆系统：短期记忆+长期记忆+向量检索

src/agentscope/memory/实现了双层记忆架构：

工作记忆（WorkingMemory）：当前对话上下文，支持标记系统（HINT临时提示、COMPRESSED已压缩消息）

长期记忆（Long-termMemory）：持久化存储，支持Mem0/Reme集成，向量检索

标记系统是个容易被忽视但极实用的设计：你可以给某条消息打上HINT标记，表示"这条消息只在当前回合有效，用完即删"，从而避免临时提示污染长期记忆。

5.评估与优化：闭环才是生产力

src/agentscope/evaluate/和src/agentscope/tuner/构成了"评估→优化→再评估"的闭环：

Evaluate模块：

BenchmarkBase抽象基类：定义标准化评估流程

RayEvaluator：支持分布式并行评估（大规模基准测试必备）

自动聚合LLM/Tool/Token使用量统计

Tuner模块：

强化学习优化（基于Trinity-RFT框架的GRPO算法）

监督微调（SFT）

提示词自动优化（基于DSPy的MIPROv2）

这意味着AgentScope不只是"让你搭一个Agent"，而是"让你搭一个能持续进化的Agent"。

6.协议支持：MCP+A2A=互联互通

MCP（ModelContextProtocol）：让Agent可以调用外部标准化工具服务，支持SSE和StreamableHTTP传输

A2A（Agent-to-Agent）：Google推出的Agent间通信协议，支持跨平台Agent协作和服务发现

这两个协议的接入，标志着AgentScope不是"闭门造车"的框架，而是积极拥抱行业标准的生态参与者。

四、怎么用？——从源码到实战的"最小可行路径"

理解了"为什么"和"能干什么"，接下来是"怎么用"。基于源码结构，我梳理了一条从入门到生产的渐进路径。

第一步：5分钟搭一个单Agent（验证可行性）

from agentscope.agentimportReActAgent

from agentscope.modelimportOpenAIChatModel

from agentscope.formatterimportOpenAIChatFormatter

from agentscope.memoryimportInMemoryMemory

from agentscope.messageimportMsg

#1.初始化模型

model=OpenAIChatModel(api_key="sk-xxx",model_name="qwen-32")

formatter=OpenAIChatFormatter()

#2.创建Agent（组合式配置）

agent=ReActAgent(

name="Assistant",

sys_prompt="你好，我是GEOAI小明，一个被时空智能体深深吸引的技术爱好者",

model=model,

formatter=formatter,

memory=InMemoryMemory(),

max_iters=10,#最大推理-行动循环次数

)

#3.对话

msg=Msg("user","帮我查一下明天的天气","user")

response=awaitagent.reply(msg)

print(response.get_text_content())

关键认知：AgentScope的Agent不是"继承出来的"，而是"组装出来的"。ReActAgent的构造函数接收model、formatter、toolkit、memory等组件，这种组合式设计让单元测试和模块替换变得异常简单。

第二步：给Agent装上工具（扩展能力边界）

from agentscope.tool import Toolkit

#初始化工具箱

toolkit=Toolkit()

#注册自定义工具

def search_weather(city:str,date:str)->str:

"""查询指定城市的天气"""

#...实现逻辑

return"晴天，25°C"

toolkit.register_tool_function(

tool_func=search_weather,

group_name="weather_tools"

)

#重新配置Agent

agent=ReActAgent(

...,

toolkit=toolkit,#注入工具箱

)

此时Agent的ReAct循环会自动判断：用户问天气→需要调用search_weather→执行工具→将结果反馈给LLM→生成自然语言回答。

第三步：多Agent协作（Pipeline编排）

from agentscope.pipeline import SequentialPipeline,FanOutPipeline

from agentscope.agent import ReActAgent,UserAgent

#定义角色

researcher=ReActAgent(name="Researcher",sys_prompt="你是研究员...",...)

writer=ReActAgent(name="Writer",sys_prompt="你是写手...",...)

reviewer=ReActAgent(name="Reviewer",sys_prompt="你是审稿人...",...)

user=UserAgent("User")

#顺序流水线：研究员→写手→审稿人

pipeline=SequentialPipeline(agents=[researcher,writer,reviewer])

#或者并发执行：三个Agent同时分析同一问题

fanout=FanOutPipeline(

agents=[agent1,agent2,agent3],

enable_gather=True,#收集所有结果

)

results=awaitfanout(msg)#返回列表[result1,result2,result3]

Pipeline的设计哲学：AgentScope没有强制你用某种特定范式（如AutoGen的群聊或LangGraph的状态机），而是提供了可组合的编排原语：顺序、并发、聊天室、消息中心。

你可以根据业务复杂度自由组合。

第四步：接入RAG（知识增强）

from agentscope.ragimportKnowledgeBase

#构建知识库

kb=KnowledgeBase(

docs_path="./documents",

embedding_model=...,#向量嵌入模型

vector_store=...,#向量存储后端

)

#Agent自动检索

agent=ReActAgent(

...,

knowledge=[kb],#支持多知识库

enable_rewrite_query=True,#查询重写（优化检索质量）

)

当用户提问时，Agent会自动：

从知识库检索相关文档，将检索结果注入提示词上下文，基于增强后的上下文生成回答。

第五步：生产级部署（可观测+持久化+分布式）

from agentscope.session import RedisSession

from agentscope.tracing import setup_tracing

#1.链路追踪（OpenTelemetry）

setup_tracing(

service_name="my_agent_service",

exporter_type="otlp",#导出到Jaeger/Zipkin

)

#2.会话持久化（Redis）

session=RedisSession(

redis_url="redis://localhost:6379",

session_id="user_123",

)

#3.分布式评估（Ray）

from agentscope.evaluate import RayEvaluator

evaluator=RayEvaluator(

num_workers=10,#10个并行Worker

...

)

awaitevaluator.run(solution)

到这一步，你的Agent系统已经具备了：

可观测性：每次LLM调用、工具执行都有Trace

可恢复性：会话状态持久化，重启后无缝恢复

可扩展性：Ray分布式评估支持大规模基准测试

五、源码级洞察：三个值得深挖的设计细节

1.消息系统的"块化设计"

Msg不是简单的字符串，而是块（Block）的集合：

@dataclass

classMsg:

name:str

content:str|list[Block]#可以是文本，也可以是块列表

role:str

id:str

timestamp:str

metadata:dict|None

块类型包括：TextBlock、ImageBlock、AudioBlock、VideoBlock、ToolUseBlock、ToolResultBlock、ThinkingBlock。

这意味着AgentScope从底层就支持"一条消息里同时包含文字、图片和工具调用结果"，而不是像某些框架那样把多模态当成"附件"处理。这对构建真正的多模态Agent（如视觉分析助手）至关重要。

2.记忆压缩的"结构化摘要"

当工作记忆超过Token阈值时，AgentScope不是简单地"截断老消息"，而是：

classCompressionConfig(BaseModel):

enable:bool

trigger_threshold:int#触发阈值

keep_recent:int=3#保留最近N条不压缩

compression_prompt:str#压缩提示词

summary_schema:Type[BaseModel]#结构化摘要模型

compression_model:ChatModelBase#压缩专用模型（可与主模型不同）

这允许你用轻量级模型（如Qwen-8B）做记忆压缩，用重量级模型（如Qwen-72B）做核心推理，从而显著降低Token成本。

3.Tuner模块的"LLM-as-Judge"

在强化学习优化中，评判函数（Judge）可以用LLM充当：

async def my_judge(task,response,auxiliary_models,logger):

judge_model=auxiliary_models["judge"]#独立的评判模型

score=await evaluate_response(judge_model,task,response)

return JudgeOutput(reward=score)

这意味着AgentScope支持"自我进化"：Agent生成回答→Judge（可以是另一个LLM）评估质量→强化学习调整策略→生成更好的回答。这在客服、内容审核等需要持续优化的场景中极具价值。

五、总结：AgentScope 的"工程化坐标"

在多智能体框架的版图中，AgentScope 占据了一个独特的位置：

维度	AutoGen	LangGraph	AgentScope
设计哲学	对话驱动协作	状态机精确控制	工程化稳健运行
协作范式	群聊（涌现式）	有向图（显式控制）	消息驱动（灵活编排）
上手难度	低	中	中
生产就绪度	中	高	极高
核心优势	快速原型	复杂工作流	可观测、可扩展、可容错
最佳场景	实验/探索	工业级流程	企业级 Agent 中台

如果你正在做一个需要上线、需要监控、需要持续优化的多智能体项目，AgentScope 不是"可选项"，而是"必选项"。

最后一句忠告：框架只是工具，理解源码才是能力。AgentScope 的源码结构清晰、模块边界明确，非常适合作为"学习多智能体工程化"的教科书。建议从agent/_react_agent.py的reply()方法开始，单步跟踪一次完整的 ReAct 循环，你会对"一个生产级 Agent 是怎么思考、怎么行动、怎么记忆"有全新的认知。

本文基于 AgentScope 开源框架源码解析，框架地址：http://github.com/agentscope-ai/agentscope