DB-GPT Agent Action 模块深度解析把 LLM 决策转化为真实世界影响的最后一公里【免费下载链接】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 体系中Action行动模块位于整条推理链的最下游负责把 profile画像、memory记忆与 planning规划模块产出的决策真正翻译为对环境的可执行操作。本文以 docs/docs/agents/modules/action/action.md 为骨架结合packages/dbgpt-core/src/dbgpt/agent下的源码实现系统讲解 Action 的四大视角目标、生成、空间、影响、Action基类与ActionOutput的设计、如何编写并绑定自定义 Action以及 Action 如何与 Resource/Tool 体系协作帮助你从零构建一个可落地的自定义 Agent。Action 模块在 DB-GPT Agent 架构中的定位在 DB-GPT 的 Agent 架构中action模块承担着把 Agent 的决策翻译成具体结果的职责。它处于整个链路的最下游直接与外部环境交互同时受到profile、memory与planning三个上游模块的直接影响——画像决定了 Agent 是谁、要做什么记忆提供了过往上下文规划给出了行动蓝图而 Action 则是把这一切落到实处的执行器。在 DB-GPT 中任何 Agent 都必须拥有一个 Action这是 Agent 能够对外部世界产生影响的唯一出口。因此理解 Action 的抽象模型、执行协议与资源协作方式是自定义 Agent 开发中绕不开的一环。理解 Action 的四个核心视角按照学术界对大语言模型自主 Agent 的经典总结见 A survey on large language model based autonomous agents可以从以下四个维度刻画 ActionAction Goals行动想达成什么行动目标回答Agent 通过该动作希望实现什么。DB-GPT 中归纳为三类任务完成Task Completion执行具体任务例如在软件开发中编写一个函数或者在游戏里制造一把铁镐沟通交流Communication与其他 Agent 进行信息交互这是多 Agent 协作场景的基础环境探索Environment exploration探索陌生环境以扩展感知在探索与利用之间取得平衡。Action Production行动如何产生行动生成策略回答动作是怎么被生成出来的主要有两条路径基于记忆回忆Action via memory recollection根据当前任务从 Agent 记忆中提取相关信息将任务与回忆出的记忆共同作为提示词触发 Agent 产生动作基于计划跟随Action via plan followingAgent 严格按照预先规划好的计划逐步执行动作。这两条路径在 DB-GPT 中并非互斥memory 模块为前者提供素材planning 模块如 plan 模块文档为后者提供蓝图而 Action 层则统一负责把它们落地为一次具体的执行。Action Space有哪些可用动作行动空间指 Agent 可以执行的全部可能动作集合可粗略分为两大类外部工具external toolsAgent 调用的真实工具如数据库查询、代码解释器、Shell、网页检索等LLM 内部知识internal knowledge of the LLMs模型自身的推理与知识输出能力。在 DB-GPT 中外部工具以Resource资源的形式暴露给 Action。从 ResourceType 枚举定义 可以看到资源体系覆盖了database、knowledge、internet、tool、skill、plugin、各类文件text/excel/image/audio/video、awel_flow、app、pack与connector等丰富类型。Tool只是其中一种Resource而ToolAction正是消费这种资源的核心 Action。Action Impact行动带来什么后果行动影响指动作执行后产生的后果可以涵盖大量具体实例主要包括改变环境Changing environmentsAgent 直接改变环境状态例如移动位置、收集物品、建造建筑改变内部状态Altering internal states行动反过来改变 Agent 自身包括更新记忆、形成新计划、获取新知识等触发新动作Triggering new actions在任务完成过程中一个 Agent 的动作可以触发另一个动作形成动作链。在 DB-GPT 中触发新动作与改变内部状态分别对应ActionOutput中的next_speakers指定下一个发言者与memory_fragments会话记忆片段等字段体现了动作之间、动作与记忆之间的闭环联动。深入 Action 基类决策如何变成 ActionOutput在 DB-GPT 中所有动作都继承自Action抽象基类。它是整个 action 模块的契约中心通过范型Action[T]约束输出模型类型并定义了如下关键成员成员类型作用resourceResource当前 Action 绑定的资源实例通过init_resource()注入resource_needResourceType声明该 Action 需要何种资源类型不需要则返回NonenamestrAction 名称默认取类名并去掉Action后缀out_model_typeType[BaseModel]声明 LLM 输出应解析成的结构化模型类型ai_out_schema/ai_out_schema_jsonstr基于out_model_type自动生成输出 JSON 格式约束与示例注入 Agent 提示词render_protocolVis可视化渲染协议用于图表、仪表盘等结果的展示render_prompt()str渲染协议对应的提示词_input_convert()T从 AI 消息文本中提取 JSON 并转换为out_model_type实例parse_action()Action根据消息解析实际要执行的 Action默认返回 default_actionrun()ActionOutput抽象方法真正执行动作的入口其中_input_convert使用find_json_objects从模型输出中稳健地提取 JSON 片段再通过 pydantic 的model_validate完成结构化转换——这也是为什么自定义 Action 中所有 LLM 输出都能被安全解析的关键。ActionOutput动作执行的统一结果载体ActionOutput是run()的唯一返回值类型也是 Agent 对话循环中action_report的数据来源。其核心字段如下字段说明content动作执行后的文本结果必填is_exe_success执行是否成功默认Trueview可视化渲染产物HTML/图表等resource_type/resource_value与动作关联的资源类型与值action/action_input动作名称与其输入参数thoughts动作执行前的思考摘要observations观察结果执行失败时会自动用content回填have_retry是否允许重试默认Trueask_user是否需要向用户提问默认Falsenext_speakers若当前 Agent 能确定下一发言者在此指定terminate是否终止对话循环特殊动作置True会强制结束循环memory_fragments当前对话的记忆片段用于随时恢复会话persisted_path当工具结果超过大小阈值被落盘持久化时记录文件路径此时content中为persisted-output预览块此外ActionOutput.pre_fill验证器实现了自动容错当is_exe_successFalse且未显式提供observations时会自动用content回填observations避免下游读取空值。编写你的第一个自定义 ActionSummaryAction 实战参考 自定义 Agent 文档中的 Create a Custom Action 小节我们来实现一个总结器 Agent 使用的SummaryAction。它把 LLM 生成的总结文本解析成结构化输入并做内容是否与问题无关的兜底判断。第一步定义 Action 的输出参数模型Action 的输出参数对象用 pydantic 声明description会被用于生成 AI 输出格式约束from typing import Optional from pydantic import BaseModel, Field from dbgpt.agent import Action, ActionOutput, AgentResource, ResourceType from dbgpt.agent.util import cmp_string_equal NOT_RELATED_MESSAGE Did not find the information you want. # 当前 Agent 需要执行的 Action 需要输出的参数对象 class SummaryActionInput(BaseModel): summary: str Field( ..., descriptionThe summary content, )第二步继承 Action 基类并实现 run()run()是动作的实际执行入口会在模型推理结束后被自动触发class SummaryAction(Action[SummaryActionInput]): def __init__(self, **kwargs): super().__init__(**kwargs) property def resource_need(self) - Optional[ResourceType]: # 当前 Agent 需要使用的资源类型 # 这里不需要使用资源直接返回 None return None property def render_protocol(self): # 当前 Agent 需要使用的可视化渲染协议 # 这里不需要可视化渲染直接返回 None return None property def out_model_type(self): return SummaryActionInput async def run( self, ai_message: str, resource: Optional[AgentResource] None, rely_action_out: Optional[ActionOutput] None, need_vis_render: bool True, **kwargs, ) - ActionOutput: 执行动作模型推理结束后自动触发的入口。 try: # 解析输入消息 param: SummaryActionInput self._input_convert(ai_message, SummaryActionInput) except Exception: return ActionOutput( is_exe_successFalse, contentThe requested correctly structured answer could not be found, fai message: {ai_message}, ) # 检查总结内容是否与用户问题无关 if param.summary and cmp_string_equal( param.summary, NOT_RELATED_MESSAGE, ignore_caseTrue, ignore_punctuationTrue, ignore_whitespaceTrue, ): return ActionOutput( is_exe_successFalse, contentthe provided text content is not related to user questions at all. fai message: {ai_message}, ) else: return ActionOutput( is_exe_successTrue, contentparam.summary, )这个示例完整展示了自定义 Action 的四个必备要素resource_need资源需求声明、render_protocol渲染协议、out_model_type输出结构与run()执行逻辑。其中cmp_string_equal用于对 LLM 输出做忽略大小写、标点与空白的语义比较属于 Agent 工具库dbgpt.agent.util中的实用函数。第三步将 Action 绑定到 Agent开发完 Agent 与 Action 后通过_init_actions([SummaryAction])把动作绑定到对应 Agentfrom pydantic import BaseModel from dbgpt.agent import Action, ConversableAgent class SummaryActionInput(BaseModel): ... class SummaryAction(Action[SummaryActionInput]): ... class MySummarizerAgent(ConversableAgent): def __init__(self, **kwargs): super().__init__(**kwargs) self._init_actions([SummaryAction])第四步Action 扩展参数传递如果 Action 需要读取 Agent 侧传入的扩展参数可以通过kwargs传递。Agent 侧重写prepare_act_param提供参数Action 侧在run()中读取from typing import Optional, Dict, Any, List from dbgpt.agent import Agent, Action, AgentMessage, ActionOutput, AgentResource, ConversableAgent class SummaryAction(Action[SummaryActionInput]): ... async def run(self, ai_message, resourceNone, rely_action_outNone, need_vis_renderTrue, **kwargs) - ActionOutput: # 读取 Agent 传入的扩展参数 extra_param kwargs.get(action_extra_param_key, None) pass class MySummarizerAgent(ConversableAgent): def __init__(self, **kwargs): super().__init__(**kwargs) self._init_actions([SummaryAction]) def prepare_act_param( self, received_message: Optional[AgentMessage], sender: Agent, rely_messages: Optional[List[AgentMessage]] None, **kwargs, ) - Dict[str, Any]: return {action_extra_param_key: this is extra param}Action 与 Resource 的协作ToolAction 源码剖析Action 模块与资源模块的协作在ToolAction中体现得最为充分。它声明resource_need ResourceType.Tool即消费 Tool 类型的资源并通过ToolPack执行真实工具class ToolAction(Action[ToolInput]): property def resource_need(self) - Optional[ResourceType]: 返回该 Action 需要的资源类型。 return ResourceType.Tool property def render_protocol(self): return self._render_protocol # VisPlugin 渲染协议 property def out_model_type(self): return ToolInput property def ai_out_schema(self) - Optional[str]: out_put_schema { thought: Summary of thoughts to the user, tool_name: The name of a tool that can be used to answer the current question or solve the current task., args: {arg name1: arg value1, arg name2: arg value2}, } return fPlease response in the following json format: {json.dumps(out_put_schema, indent2, ensure_asciiFalse)} Make sure the response is correct json and can be parsed by Python json.loads. 其执行链路run_tool()完整展示了Action Resource 渲染协议的协作模式通过ToolPack.from_resource(resource)从资源中恢复工具集合调用tool_pack.async_execute(resource_namename, **args)异步执行工具记录Status.RUNNING → COMPLETE / FAILED状态机结果超阈值时通过当前存储get_current_storage()将完整结果落盘content中仅保留persisted-output预览块完整内容保留在observations供数据库持久化并返回persisted_path通过VisPlugin渲染协议生成可视化视图最终封装为ActionOutput返回并携带terminate工具为终结性动作时为True。ToolAction还内置了容错回退机制当 LLM 未产出严格 JSON 时会尝试两种兜底——纯数字结果直接接受或从tool_name与expression ...模式中提取工具名与表达式重新执行。DB-GPT 内置的 Tool 资源在 工具概览文档 中有系统说明包括load_skill、code_interpreter、shell_interpreter、sql_query、html_interpreter五大内置工具它们正是经由ToolAction这类 Action 才能被 Agent 真正调用。从核心到扩展DB-GPT 内置 Action 一览除Action基类外仓库中实际提供了若干可直接使用或作为范本的内置 ActionAction文件说明BlankActioncore/action/blank_action.py空动作不做任何加工直接原样返回 AI 消息ai_out_schema为NoneToolActionexpand/actions/tool_action.py调用 Tool 类型资源执行真实工具CodeActionexpand/actions/code_action.py执行代码类任务ChartActionexpand/actions/chart_action.py生成图表DashboardActionexpand/actions/dashboard_action.py生成仪表盘IndicatorActionexpand/actions/indicator_action.py指标计算类动作WebSearchActionexpand/actions/websearch_action.py联网检索类动作ReactActionexpand/actions/react_action.pyReAct 范式推理-行动循环动作InsertActionexpand/actions/insert_action.py数据写入类动作其中BlankAction是理解Action协议最简单的最小实现——它用极简代码演示了ai_out_schema与run()两个最核心的覆写点。所有扩展 Action 都位于 expand/actions 目录它们共同印证了一个事实在 DB-GPT 中Agent 的一切对外行为——代码执行、工具调用、图表渲染、联网检索——都是 Action 这一统一抽象的具体实例化。完整可运行示例绑定 Action 的总结 Agent将上述知识点串联起来便得到了一个完整可运行的总结 Agent完整代码见 自定义 Agent 文档。核心装配流程如下import asyncio import os from dbgpt.agent import ( Agent, Action, ActionOutput, AgentContext, AgentMemory, AgentMessage, AgentResource, ConversableAgent, LLMConfig, ProfileConfig, ResourceType, UserProxyAgent, ) from dbgpt.agent.util import cmp_string_equal from dbgpt.core import ModelMessageRoleType from dbgpt.model.proxy import OpenAILLMClient from dbgpt.vis import Vis from pydantic import BaseModel, Field # ... SummaryActionInput / SummaryAction / MySummarizerAgent 定义见上文 async def main(): llm_client OpenAILLMClient( model_aliasgpt-3.5-turbo, # 或其他模型如 gpt-4o api_baseos.getenv(OPENAI_API_BASE), api_keyos.getenv(OPENAI_API_KEY), ) context: AgentContext AgentContext(conv_idsummarize) agent_memory: AgentMemory AgentMemory() agent_memory.gpts_memory.init(conv_idsummarize) summarizer ( await MySummarizerAgent() .bind(context) .bind(LLMConfig(llm_clientllm_client)) .bind(agent_memory) .build() ) user_proxy await UserProxyAgent().bind(agent_memory).bind(context).build() await user_proxy.initiate_chat( recipientsummarizer, revieweruser_proxy, messageI want to summarize advantages of Nuclear Power according to the following content. ..., ) print(await agent_memory.gpts_memory.app_link_chat_message(summarize)) if __name__ __main__: asyncio.run(main())该示例完整覆盖了前文讨论的 Action 生命周期AgentContext提供会话上下文AgentMemory提供记忆对应 Action Production 中的记忆回忆路径LLMConfig注入模型客户端MySummarizerAgent通过_init_actions([SummaryAction])绑定动作UserProxyAgent作为用户代理发起对话最终整个决策-执行-记忆回写的闭环在Action.run()处完成落地。小结Action 模块是 DB-GPT Agent 架构中连接思考与行动的枢纽。从本文的梳理可以看到架构定位Action 位于最下游、直接与环境交互受 profile/memory/planning 模块影响理论视角通过目标Goals、生成Production、空间Space、影响Impact四个维度可以完整刻画一个动作实现机制Action[T]基类 ActionOutput构成了统一契约out_model_type约束结构化输出_input_convert完成 JSON 解析run()承载真实执行资源协作Action 通过resource_need声明资源需求与ResourceType枚举tool、knowledge、database 等深度集成ToolAction是消费 Tool 资源的典型代表实践路径继承Action、声明输出模型、实现run()、通过_init_actions()绑定 Agent即可快速打造自定义 Agent 的专属动作。在此基础上你可以进一步参考 Resource 模块文档、数据库资源、知识库资源 与 Memory 模块结合render_protocol可视化协议让 Agent 的动作不仅是执行正确更能呈现得体。【免费下载链接】DB-GPTopen-source agentic AI data assistant for the next generation of AI Data products.项目地址: https://gitcode.com/GitHub_Trending/db/DB-GPT创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考