:::info 备注：这是老版本文档，已经过期，新版本文档见： https://docs.dbgpt.site/docs/latest/agents/introduction/

:::

1.认识DBGPT Agent

1.1 DBGPT Agent结构

DB-GPT Agent开发文档(过期） - 图2

1.2 DBGPT 运行机制

DB-GPT Agent开发文档(过期） - 图3

1.3 DBGPT Multi-Agents协作机制

  • 单插件模式
  • 自动规划模式
  • Awel编排模式

1.4 Memory机制

  • Common Memory 集体记忆

1.5 人类交互机制

  • 暂未实现

1.6 资源绑定加载机制

2.开发一个Agent

2.1 开发Agent

2.1.1 初始化Agent

DBGPT的Agent通过继承Agent基类实现，参考如下代码:

  1. from dbgpt.agent.agents.base_agent_new import Agent, ConversableAgent
  3. class TestAssistantAgent(ConversableAgent):
  5.     def __init__(self, **kwargs):
  6.         super().__init__(**kwargs)
2.1.2 定义身份

每个Agent设计之前需要给Agent定义好他的角色、身份、和功能作用，具体定义如下：

  1. from dbgpt.agent.agents.base_agent_new import Agent, ConversableAgent
  3. class TestAssistantAgent(ConversableAgent):
  4.     name = "Edgar" # agent的名字，
  5.     profile: str = "DataScientist" # agent的身份定义
  6.     goal: str = "xxx." # agent的核心功能目标，告诉LLM可以用他来做什么
  7.     desc: str = "xxx" # 对agent的介绍和描述，用来进行任务分派和展示，如果为空就会使用goal内容
  8.     def __init__(self, **kwargs):
  9.         super().__init__(**kwargs)
2.1.3 补充Prompt约束逻辑

Agent的Prompt默认使用固定模板组装(如果有特殊要求可绑定外挂模板)，主要包含：

  1. 1. 身份定义(自动构建)
  2. 2. 资源信息（自动装载）
  3. 3. 约束逻辑
  4. 4. 参考案例（非必须）
  5. 5. 输出格式模板和约束(自动构建)

可以看到默认开发，prompt部分可以只准备约束逻辑就可以，参考如下：

  1. from dbgpt.agent.agents.base_agent_new import Agent, ConversableAgent
  3. class TestAssistantAgent(ConversableAgent):
  4.     name = "Edgar" # agent的名字，
  5.     profile: str = "DataScientist" # agent的身份定义
  6.     goal: str = "xxx." # agent的核心功能目标，告诉LLM可以用他来做什么
  7.     desc: str = "xxx" # 对agent的介绍和描述，用来进行任务分派和展示，如果为空就会使用goal内容
  9.     # 参考如下 可以包含多个约束和推理限制逻辑,支持使用参数模板{param_name}
  10.     constraints: List[str] = [
  11.         "Please check the generated SQL carefully. Please strictly abide by the data structure definition given. It is prohibited to use non-existent fields and data values. Do not use fields from table A to table B. You can perform multi-table related queries.",
  12.         "Please select an appropriate one from the supported display methods for data display. If no suitable display type is found, table display is used by default. Supported display types: \n {display_type}",
  13.         "xxx{param_x}",
  14.     ]
  15.     def __init__(self, **kwargs):
  16.         super().__init__(**kwargs)

  • prompt动态参数赋值

    如果prompt里用到了动态的参数，需要实际对话过程组装值，需要重载实现如下接口(_init_reply_message):

  1. class TestAssistantAgent(ConversableAgent):
  2.     ### ...
  3.     ### 其他代码
  4.     ### ...
  5.     def _init_reply_message(self, recive_message):
  6.         reply_message = super()._init_reply_message(recive_message)
  7.         reply_message["context"] = {
  8.             "display_type": self.actions[0].render_prompt(),
  9.             "dialect": self.resource_loader.get_resesource_api(
  10.                 self.actions[0].resource_need
  11.             ).get_data_type(self.resources[0]),
  12.         }
  13.         return reply_message
2.1.4 资源预加载 (非必须)

如果存在一些特定资源，在Agent初始化时候要预先把绑定资源加载起来可以参考如下实现,基于资源实际情况决定，大部分情况下不需要

  1. class TestAssistantAgent(ConversableAgent):
  2.     ### ...
  3.     ### 其他代码
  4.     ### ...
  5.     async def a_preload_resource(self):
  6.         plugin_loader_client: ResourcePluginClient = (
  7.             self.resource_loader.get_resesource_api(ResourceType.Plugin)
  8.         )
  9.         for item in self.resources:
  10.             if item.type == ResourceType.Plugin:
  11.                 self.plugin_generator = await plugin_loader_client.a_load_plugin(
  12.                     item.value, self.plugin_generator
  13.                 )
2.1.5 结果检查(非必须)

如果对Action执行结果需要进行严格的校验和验证，分为代码逻辑验证和LLM验证两种模式，当然验证非必须，不实现默认通过，参考如下处理:

  1. class TestAssistantAgent(ConversableAgent):
  2.     ### ...
  3.     ### 其他代码
  4.     ### ...
  6.     async def a_correctness_check(self, message: Optional[Dict]):
  7.         task_gogal = message.get("current_goal", None)
  8.         action_report = message.get("action_report", None)
  9.         task_result = ""
  10.         if action_report:
  11.             task_result = action_report.get("content", "")
  13.         check_result, model = await self.a_thinking(
  14.             messages=[
  15.                 {
  16.                     "role": ModelMessageRoleType.HUMAN,
  17.                     "content": f"""Please understand the following task objectives and results and give your judgment:
  18.                         Task Gogal: {task_gogal}
  19.                         Execution Result: {task_result}
  20.                     """,
  21.                 }
  22.             ],
  23.             prompt=CHECK_RESULT_SYSTEM_MESSAGE,
  24.         )
  25.         success = str_to_bool(check_result)
  26.         fail_reason = None
  27.         if not success:
  28.             fail_reason = f"Your answer was successfully executed by the agent, but the goal cannot be completed yet. Please regenerate based on the failure reason:{check_result}"
  29.         return success, fail_reason

2.2 开发Action

2.2.1 初始化Action

Agent所有对外部环境、现实世界的操作都通过Action实现，Action定义了Agent的输出内容结构和实际去执行对应的操作，具体Action实现继承Action基类，参考如下:

  1. from dbgpt.agent.actions.action import Action, ActionOutput, T
  3. ## 当前Agent需要执行的Action需要输出的参数对象，
  4. class TestActionInput(BaseModel):
  5.     param_xx: str = Field(
  6.         ...,
  7.         description="The name of a indicator.",
  8.     )
  11. class TestAction(Action[TestActionInput]):
  12.     def __init__(self, **kwargs):
  13.         super().__init__(**kwargs)
  14.         ## 当前Agent需要展示内容使用的VIS协议标签
  15.         self._render_protocal = VisPlugin()
  17.     ## 当前Agent需要使用的资源类型
  18.     @property
  19.     def resource_need(self) -> Optional[ResourceType]:
  20.         return ResourceType.Knowledge
  23.     ## 指定当前Action需要的参数对象
  24.     @property
  25.     def out_model_type(self):
  26.         return TestActionInput
  29.     ## 实际执行Action的入口，模型推理后会自动发起Action执行
  30.     async def a_run(
  31.         self,
  32.         ai_message: str,
  33.         resource: Optional[AgentResource] = None,
  34.         rely_action_out: Optional[ActionOutput] = None,
  35.         need_vis_render: bool = True,
  36.     ) -> ActionOutput:
  37.         ## 具体执行逻辑
  38.         ...
2.2.2 绑定Action

Agent和Action开发定义完成后，将Action绑定到对应Agent

  1. from dbgpt.agent.agents.base_agent_new import Agent, ConversableAgent
  3. class TestAssistantAgent(ConversableAgent):
  4.     ### Agent身份定义
  5.     ...
  6.     def __init__(self, **kwargs):
  7.         super().__init__(**kwargs)
  8.         self._init_actions([TestAction])
2.2.3 Action扩展参数处理

因为实际的Action需要和现实环境、API等交互，除了依赖模型输出参数和资源信息外可能还需要其他额外的环境参数等,对于Action的额外扩展参数处理参考如下：

  1. from dbgpt.agent.agents.base_agent_new import Agent, ConversableAgent
  3. class TestAssistantAgent(ConversableAgent):
  4.     ### Agent身份定义
  5.     ...
  6.     def __init__(self, **kwargs):
  7.         super().__init__(**kwargs)
  8.         self._init_actions([TestAction])
  10.     ### action扩展参数
  11.     def prepare_act_param(self, recive_message: Optional[Dict],   sender: ConversableAgent) -> Optional[Dict]:
  12.         return {"plugin_generator": self.plugin_generator}
2.2.4 Action复杂参数（多级结构）LLM输出模板处理

3.Agent注册与使用

3.1 Agent注册

Agent开发完成后，如果要在应用或者Awel里可见和被使用，需要进行Agent注册：

具体代码如下:

  1. from dbgpt.agent.agents.agents_manage import agent_manage
  3. ## 代码路径 dbgpt.serve.agent.agents.controller， 也可以放在其他地方，确保会被加载的代码处即可c
  5. agent_manage.register_agent(TestAssistantAgent)

3.2 Agent使用

3.2.1 应用中配置使用（单Agent对话、 自动规划模式）

注册成功后这里可以看到新的Agent

DB-GPT Agent开发文档(过期） - 图4

给Agent绑定资源，完成应用开发，就可以进行对话

DB-GPT Agent开发文档(过期） - 图5

3.2.2 Awel应用中配置使用

DB-GPT Agent开发文档(过期） - 图6

3.3 Agent使用效果

DB-GPT Agent开发文档(过期） - 图7

DB-GPT Agent开发文档(过期） - 图8