企业官网建设流程全解析

DB-GPT Agent 框架深度解析：ConversableAgent 架构、多智能体协作与三层记忆体系

【免费下载链接】DB-GPTopen-source agentic AI data assistant for the next generation of AI + Data products.项目地址: https://gitcode.com/GitHub_Trending/db/DB-GPT

本文以 DB-GPT 官方的 Agent 框架概念文档为骨架，完整讲解其数据驱动的多智能体架构：包括 ConversableAgent 的五大核心模块（Profile / Memory / Planning / Action / Resource）、消息驱动的协作模式、分级记忆体系，以及如何用几行代码构建并运行一个自定义 Agent。读完后你可以直接对照源码理解"接收消息 → 思考 → 审查 → 行动 → 验证"这一完整对话循环的实现机制，并能在仓库示例的基础上搭建自己的多智能体应用。

一、框架定位：数据驱动的多智能体框架

DB-GPT 提供的是一套数据驱动的多智能体框架（data-driven multi-agent framework），用于构建能够自主协作、调用工具、访问数据库并在多轮会话中保持记忆的 AI Agent。与只做单轮对话补全的系统不同，该框架的 Agent 具备完整的"感知—思考—行动—反思"闭环，并且原生面向数据场景（SQL 查询、数据分析、图表生成）设计。

从源码结构看，整个 Agent 框架位于packages/dbgpt-core/src/dbgpt/agent/目录下，核心实现分层如下：

二、Agent 架构：五大核心模块

官方文档给出的架构如下：每个 Agent 都围绕五个核心模块构建，ConversableAgent将其统一封装，对外连接用户、LLM、外部工具、数据库与知识库。

这五个模块在ConversableAgent中体现为一组显式字段，见 ConversableAgent 类定义：

• agent_context（AgentContext）：会话运行上下文，携带conv_id、语言、采样温度等运行参数；
• actions（List[Action]）：行动模块，可绑定一个或多个 Action，按顺序执行并以前一个 Action 的输出为下一个的输入；
• resource（Resource）：资源模块，在build()阶段通过preload_resource()预加载，并通过get_prompt()把知识库/表结构等信息注入提示词；
• llm_config（LLMConfig）：模型配置，封装LLMClient与模型选择策略；
• memory（AgentMemory）：记忆模块，默认为带 GptsMemory 会话记忆的实现。

此外还有若干重要的运行控制字段：max_retry_count（默认 3，单轮回复失败重试上限）、max_timeout（默认 600 秒）、run_mode（AgentRunMode.DEFAULT或LOOP，循环模式下 Agent 会持续迭代直到收到终止信号）以及stream_out（是否流式输出）。

2.1 Profile：角色的定义与约束

从源码结构看，Role基类（见 Role 类定义）持有profile: ProfileConfig与memory: AgentMemory两个必填/默认字段，并通过一组只读属性对外暴露角色信息：name、role、goal、constraints、retry_goal、retry_constraints、desc、examples、expand_prompt。这些属性正是系统提示词模板中的渲染变量——Role.prompt_template()方法会解析模板中未声明的变量，把上述角色参数与运行时参数一起渲染进提示词（见 prompt_template 实现）。

值得注意的是安全细节：模板渲染统一走SandboxedEnvironment（Jinja2 沙箱），注释中明确说明这是为了防止用户可控内容（例如所选 Skill 的指令文本）通过模板注入导致 SSTI（服务端模板注入）问题（见 build_system_prompt 实现）。

2.2 Planning 与 Action：任务分解和执行

Planning 模块负责把复杂任务拆成可执行步骤，agent/core/plan/下提供了几种策略：

• planner_agent.py：规划型 Agent，通过 LLM 生成子任务计划；
• team_auto_plan.py：Auto-Plan 团队，实现"管理器—工人"式协作（对应下文 Manager-Worker 模式）；
• plan/awel/：基于 AWEL 图引擎的 Agent 编排算子（agent_operator.py、team_awel_layout.py），可把 Agent 团队构造成 DAG 在流程画布中运行。

Action 模块则是"手"。ConversableAgent.act()的默认实现（见 act 方法）遍历self.actions列表：先调用action.parse_action()从 LLM 回复中解析出真正要执行的动作，再调用real_action.run()得到ActionOutput；若一个 Action 未解析出可执行动作则continue，最终返回最后一个 Action 的输出。ActionOutput携带执行结果（content/observations）、是否成功（is_exe_success）、是否可重试（have_retry）、思考过程（thoughts）等字段，是验证与记忆写入的关键数据源。

三、ConversableAgent：对话循环的完整实现

ConversableAgent是所有 Agent 的基类（见 base_agent.py#L39），它继承自Role并实现抽象接口Agent（send/receive/generate_reply/thinking/review/act/verify，定义于 agent.py#L16）。文档中概括的"接收消息 → think（plan）→ act → respond"循环，在generate_reply()中被展开为带重试的六阶段流程（见 generate_reply 实现）：

1. 初始化回复消息：_init_reply_message()基于收到的消息创建回复消息，轮次rounds + 1；
2. Thinking（思考）：_load_thinking_messages()组装提示词——读取记忆（read_memories()）、注入任务进度摘要（task_progress）、加载资源提示词（数据库表结构、知识库等）、构建 system/user prompt；随后thinking()调用 LLM（内置 3 次模型级重试，失败后换模型并休眠 10 秒重试，见 thinking 方法）。若 LLM 报"上下文超长"错误且启用了上下文管理（_context_manager），会自动触发响应式压缩（Layer 4 reactive compaction）后重试；
3. Review（审查）：review()检查回复是否合规，返回(approve, comments)写入AgentReviewInfo；
4. Act（行动）：执行第二节所述的 Action 链，产出ActionOutput并挂到消息的action_report上；
5. Verify（验证）：verify()依次检查审查是否通过、Action 是否执行成功、结果是否为空，最后调用子类可覆写的correctness_check()；
6. 自我修正循环：若verify不通过且act_out.have_retry为真，把失败原因作为新的observation写入记忆并进入下一轮（直至max_retry_count）；通过则写入记忆并（非 LOOP 模式或动作终止时）结束。

整个循环由max_retry_count（默认 3）和max_timeout（默认 600s）双重保护，每轮失败都会通过send()把带失败原因的消息回传给发件人以组织新的求解指令。

AgentMessage是智能体间通信的消息对象（见 AgentMessage 定义），除文本content外还携带rounds（轮次）、action_report、review_info、current_goal、model_name、success等字段，to_llm_message()负责把消息转成 LLM 可消费的 dict。

3.1 AgentContext 运行上下文

AgentContext（定义于 agent.py#L197-L227）是每次会话的运行参数容器，常用字段及默认值：

build()阶段（见 build 方法）会完成三件准备工作：预加载资源、check_available()校验（身份、Action 所需资源、LLM 配置齐全），以及初始化记忆会话f"{conv_id}_{role}_{name}"并从GptsMemory恢复历史 Action 输出。校验逻辑见 check_available 实现——非人类 Agent 若缺少 Action 模块、LLM 配置或 Action 声明的资源类型，都会在这里抛出ValueError。

四、多智能体协作模式

文档定义了三种协作拓扑，在源码中均有对应实现：

• Sequential（顺序协作）：Agent 按顺序把结果传给下一个。底层就是send()/receive()消息传递链——initiate_chat()发起对话后，回复方通过receive()生成回复并send()回发件人（见 receive 实现），消息中的rely_messages参数支持显式指定"依赖消息"（例如引用上游 Agent 的执行结果），在_load_thinking_messages()中会被格式化为Question:/Observation:段落注入提示词。
• Parallel（并行协作）：多个 Agent 同时处理子任务。ConversableAgent内置executor（默认单线程ThreadPoolExecutor，见 字段定义），blocking_func_to_async()可把阻塞函数放入执行器并发运行；团队层面的并行编排可结合plan/awel/中的 AWEL 算子实现。
• Manager-Worker（管理器—工人）：规划 Agent 把任务委派给专家 Agent。对应实现是agent/core/plan/team_auto_plan.py的 Auto-Plan 团队（PlannerAgent负责拆解与调度），可参考示例 auto_plan_agent_dialogue_example.py 查看完整用法。

所有 Agent 之间的通信都统一走Agent抽象接口的send/receive/generate_reply三件套，并配合root_tracer的全链路 trace span（agent.send、agent.receive、agent.generate_reply.thinking/act/verify等）便于观测每一步的思考、行动与验证结果。

五、记忆体系：感知 / 短期 / 长期 / 混合

文档给出的记忆分级表：

源码中该体系位于agent/core/memory/，与文档的一一对应关系：

• base.py：SensoryMemory、ShortTermMemory、Memory抽象基类与MemoryFragment记忆片段；
• long_term.py/short_term.py：长期与短期记忆实现（短期记忆EnhancedShortTermMemory带重要性淘汰机制）；
• hybrid.py：HybridMemory（见 hybrid.py#L31）显式建模人类的短期/长期记忆——短期记忆缓冲近期感知，长期记忆固化重要信息；
• agent_memory.py：AgentMemory把上述记忆与 GptsMemory（会话消息记忆）整合为 Agent 可直接使用的统一入口。

几个值得注意的实现细节：

1. 向量库持久化：HybridMemory.from_chroma()类方法（见 hybrid.py#L69-L110）可以基于 Chroma 向量库构建长期记忆，集合默认名为agent_memory_long_term，数据默认落在DATA_DIR/agent_memory；from_vstore()则支持任意VectorStoreBase实现，说明长期记忆的存储后端是可插拔的。
2. 重要性评分：记忆片段写入时可携带importance分数与is_insight标记（见 AgentMemoryFragment），Role基类预留了memory_importance_scorer与memory_insight_extractor属性（默认由 LLM 打分/抽取，memory/llm.py），用于决定哪些片段值得长期保留。
3. 记忆读写闭环：Role.write_memories()（见 role.py#L278-L406）在每轮对话结束后把thought / action / observation等字段渲染成记忆片段并写入；read_memories()在下轮思考前按当前观测召回相关记忆注入提示词。recovering_memory()还支持从历史ActionOutput恢复记忆，配合build()中的gpts_memory.get_agent_history_memory()实现跨进程会话恢复。
4. 任务进度快照：_task_progress列表逐轮追加已完成的步骤（step / action / phase / status），并通过task_progress_summary属性渲染成 "## Task Progress" 摘要注入每次 LLM 调用，防止 Agent 在长任务中重复已完成的步骤；同时每步的完整action_input/observation会写入磁盘快照文件（step_NNN_action.json），保证上下文压缩后仍可通过read_file找回精确数值（见 _write_op_snapshot）。

六、预置 Agent 类型

文档列出的四类预置 Agent 在agent/expand/中都有实现，并各有配套示例脚本：

expand/下还有dashboard_assistant_agent.py、excel_table_agent.py、web_assistant_agent.py、data_scientist_agent.py等更专门的变体，可作为自定义 Agent 的参考模板。

七、快速上手：构建并运行一个 Agent

7.1 官方文档的最小示例

from dbgpt.agent import ConversableAgent, AgentContext # Define a simple custom agent agent = ConversableAgent( name="DataAnalyst", role="You are a data analysis expert", goal="Help users analyze data and generate insights", llm_config={"model": "chatgpt_proxyllm"}, ) # Start a conversation result = await agent.a_send("Analyze the sales trends for Q4 2024")

该示例展示了用 Profile 参数（name/role/goal）+ 模型配置定义 Agent 的最小形态。需要注意的是：从当前源码看，ConversableAgent是 Pydantic 模型，角色信息实际经由ProfileConfig传入，且完整运行必须绑定AgentContext（check_available()会强制校验 context、actions 与 llm_config 三者齐备），因此该示例应理解为概念性最小表达；可直接运行的完整写法见下一节。

7.2 仓库示例中的完整可运行写法

仓库示例 single_agent_dialogue_example.py 给出了标准工作流：创建 Agent → bind 上下文/模型/记忆 → build 校验初始化 → 通过 UserProxyAgent 发起对话：

import asyncio import os from dbgpt.agent import AgentContext, AgentMemory, LLMConfig, UserProxyAgent from dbgpt.agent.expand.code_assistant_agent import CodeAssistantAgent async def main(): from dbgpt.model.proxy.llms.siliconflow import SiliconFlowLLMClient # 1. 创建 LLM 客户端（此处以 SiliconFlow 为例，可替换为任意 dbgpt.model 中的客户端） llm_client = SiliconFlowLLMClient( model_alias=os.getenv( "SILICONFLOW_MODEL_VERSION", "Qwen/Qwen2.5-Coder-32B-Instruct" ), ) # 2. 会话上下文与记忆 context: AgentContext = AgentContext(conv_id="test123", gpts_app_name="代码助手") agent_memory = AgentMemory() agent_memory.gpts_memory.init(conv_id="test123") try: # 3. 链式 bind：上下文 → LLM 配置 → 记忆，然后 build() 完成校验与记忆恢复 coder = ( await CodeAssistantAgent() .bind(context) .bind(LLMConfig(llm_client=llm_client)) .bind(agent_memory) .build() ) user_proxy = await UserProxyAgent().bind(context).bind(agent_memory).build() # 4. 由人类代理（UserProxyAgent）发起对话 await user_proxy.initiate_chat( recipient=coder, reviewer=user_proxy, message="计算下321 * 123等于多少", ) finally: agent_memory.gpts_memory.clear(conv_id="test123") if __name__ == "__main__": asyncio.run(main())

bind()是ConversableAgent提供的统一装配入口（见 bind 方法），它按类型分发绑定目标：

• AgentContext→agent_context
• LLMConfig→llm_config
• AgentMemory→memory（注意：GptsMemory不可直接绑定，会抛出ValueError，请改用AgentMemory）
• Resource→resource
• ProfileConfig→profile
• Action类/实例或 Action 列表 → 追加进actions
• PromptTemplate→bind_prompt（直接覆盖系统提示词模板）
• SkillBase/FileBasedSkill→ 绑定 Skill，并把 Skill 的指令模板设为bind_prompt，让技能指令成为 Agent 的系统提示词

也就是说，同一个bind()入口覆盖了"Profile、Memory、Planning、Action、Resource"五大模块的装配，这也是框架"数据驱动"的体现：Agent 行为由绑定的数据（上下文、记忆、资源、模型策略）决定，而不是硬编码。

7.3 更多协作示例

• 摘要型单 Agent：single_summary_agent_dialogue_example.py
• SQL/数据分析 Agent：sql_agent_dialogue_example.py
• Auto-Plan 多 Agent 协作：auto_plan_agent_dialogue_example.py
• 沙箱代码执行 Agent：sandbox_code_agent_example.py

八、延伸阅读与源码导航

按官方文档的"What's next"指引，并结合当前仓库的实际文件位置：

• Agent 框架详解：docs/docs/agents/introduction.md
• 自定义 Agent：docs/docs/agents/introduction/custom_agents.md
• Agent 工具：docs/docs/agents/introduction/tools.md
• 任务分解（Planning）：docs/docs/agents/introduction/planning.md
• 核心基类：packages/dbgpt-core/src/dbgpt/agent/core/base_agent.py
• 抽象接口与上下文：packages/dbgpt-core/src/dbgpt/agent/core/agent.py
• 记忆体系：packages/dbgpt-core/src/dbgpt/agent/core/memory/hybrid.py
• 预置 Agent 集合：packages/dbgpt-core/src/dbgpt/agent/expand/

小结：DB-GPT 的 Agent 框架以ConversableAgent为中枢，用 Profile/Memory/Planning/Action/Resource 五大模块把"角色、记忆、规划、执行、资源"解耦，用send/receive/generate_reply消息循环串联单 Agent 与多 Agent 协作，并用"感知—短期—长期—混合"的分级记忆（向量库可插拔 + 重要性评分 + 任务进度快照）支撑跨会话的长程任务。理解这套结构后，你可以基于expand/下的现成模板，用bind() + build() + initiate_chat()三步骤快速搭建自己的数据智能体。

【免费下载链接】DB-GPTopen-source agentic AI data assistant for the next generation of AI + Data products.项目地址: https://gitcode.com/GitHub_Trending/db/DB-GPT