跳到主要内容

第 1 章 · 单个 Agent 的五步循环

本章讲 DB-GPT 最核心的东西:一个 Agent 接到任务后,内部怎么一步步把活干完、还能自己纠错。这是整个框架的灵魂,读懂这一章,其余都是它的扩展。

1.1 它要解决的小问题​

大模型会写 SQL,但它不知道自己写的 SQL 跑不跑得通。直接把模型吐的 SQL 给用户,经常是字段名拼错、表关联错、或者条件太死查不到数据。

DB-GPT 的答案:别信模型,去执行,用执行结果当裁判。模型写完 SQL,系统真的连库去跑;跑挂了或查不到数据,就把报错原文当成新的"观察"塞回去,让模型读着报错改一版。这就是"自我纠错"。

1.2 思路/直觉:五步闭环​

把一次回复拆成五步,失败就回到第一步重来:

   ┌──────────────────────────────────────────────┐
   │                                                │
   ▼                                                │ 不通过:把失败原因
 ① think ──► ② review ──► ③ act ──► ④ verify ──┐    │ 当新"观察"塞回
  问 LLM      自审合规      真执行     对答案       │    │
  要答案                  (跑SQL/                  │    │
                          跑代码)                  ├────┘
                                                   │
                                          通过 ──►  写记忆 → 返回

怎么读: 横着是正常流;④verify 不通过时,沿右边的回边把"为什么错"塞回 ①,最多重试 max_retry_count 次。

1.3 原理演示(示意,非源码)​

先用简化代码把这个闭环演出来,建立直觉:

# 示意,非源码:把五步循环的骨架抽出来
async def generate_reply(received):
    observation = received.content          # 一开始的"观察"就是用户问题
    for attempt in range(MAX_RETRY):
        msgs = build_prompt(observation)    # ① 组装 prompt(含记忆/资源)
        llm_reply = await think(msgs)       #    问 LLM 要一版答案
        approve, _ = await review(llm_reply)# ② 自审:合规吗
        act_out = await act(llm_reply)      # ③ 真执行:跑 SQL / 跑代码
        ok, reason = await verify(act_out)  # ④ 对答案:成功?有数据?
        if ok:
            await write_memory(...)         #    成功:记一笔,跳出
            return act_out
        observation = reason                # ⑤ 失败:把"为什么错"当新观察
    return failed()                         #    重试用尽,带着失败返回

重点看 observation = reason 这一行:失败原因变成了下一轮的输入——这就是自我纠错的全部秘密。

1.4 真实实现​

真正的循环在 ConversableAgent.generate_reply(dbgpt/agent/core/base_agent.py:436-736),核心是那个 while current_retry_counter < self.max_retry_count 循环。把五步对到真实代码:

① think —— 先组装思考消息,再问 LLM(base_agent.py:536-571):

# base_agent.py:566-571  真实片段
llm_reply, model_name = await self.thinking(
    thinking_messages,
    sender,
    stream_callback=_llm_stream_callback,
)

thinking(base_agent.py:738-...)内部把每条 AgentMessage 转成 LLM 消息,调 self.llm_client.create(...),自带 3 次网络重试抗限流/抖动(base_agent.py:757 的 while retry_count < 3)。

② review —— 自审这版答案是否合规(base_agent.py:616-620)。基类 review 默认放行,子类可覆盖做内容审查。

③ act —— 把 LLM 文本变成动作并执行(base_agent.py:640-647 调 self.act)。act(base_agent.py:795-839)遍历挂在 Agent 上的每个 Action,先 parse_action 把文本解析成结构化动作,再 run 执行:

# base_agent.py:824-835  真实片段
real_action = action.parse_action(ai_message, default_action=action, **kwargs)
if real_action is None:
    continue
last_out = await real_action.run(
    ai_message=message.content if message.content else "",
    resource=None, rely_action_out=last_out, **kwargs,
)

④ verify —— 对答案(base_agent.py:670-672 调 self.verify)。基类 verify(base_agent.py:847-872)做三层检查:review 没批准 → 失败;action_output.is_exe_success 为假 → 失败;执行结果为空 → 失败;最后落到 correctness_check 由子类做语义校验。

⑤ 自我纠错 —— 这是最妙的一段(base_agent.py:680-721):

# base_agent.py:680-696  真实片段(精简)
if not check_pass:
    if not act_out.have_retry:
        break                       # 这个动作声明"不可重试",直接放弃
    fail_reason = reason
    observation = fail_reason        # ← 失败原因变成下一轮的"观察"
    await self.write_memories(..., check_pass=False,
                              check_fail_reason=fail_reason, ...)
else:
    observation = act_out.observations
    await self.write_memories(..., check_pass=True, ...)
    if self.run_mode != AgentRunMode.LOOP or act_out.terminate:
        break                       # 成功且非 LOOP 模式:收工

失败时不仅把 reason 设为新观察,还把失败也写进记忆——这样模型重试时能从记忆里看到"我上次错在哪",而不是干净地从头再来。

下一轮开始前,系统会把上一版回复作为一条消息 send 回发送方(base_agent.py:718-721),让历史里留下"我试过、失败了"的痕迹。

1.5 一个真实的 verify:DataScientist 怎么对 SQL 答案​

抽象的 verify 不够直观,看一个具体 Agent 怎么覆盖 correctness_check。DataScientistAgent(dbgpt/agent/expand/data_scientist_agent.py:102-156)的校验真的连库执行 SQL:

# data_scientist_agent.py:118-149  真实片段(精简)
action_reply_obj = json.loads(action_out.content)
sql = action_reply_obj.get("sql", None)
if not sql:
    return False, "...没找到需要生成的 sql 信息"
try:
    columns, values = await self.database.query(sql=sql, db=action_out.resource_value)
    if not values or len(values) <= 0:
        return False, "...当前 SQL 查不到数据,可能是过滤字段值或条件不当"
    return True, None
except Exception as e:
    return False, f"SQL execution error, ...请重读历史信息修复此 SQL。错误信息:{e}"

关键:它不是问模型"你这 SQL 对吗",而是真去 self.database.query(sql=...) 执行。查不到数据、抛异常,都返回 (False, 具体原因)——这个原因就会沿 §1.4 的回边变成下一轮的观察,模型读着"SQL execution error: 字段 xxx 不存在"去改。这就是"执行驱动的自我纠错"落到实处的样子。

注意 DataScientistAgent 把 max_retry_count 设成 5(data_scientist_agent.py:72)——SQL 容易错,多给几次纠错机会。

1.6 关键细节 / 坑​

  • have_retry 是急刹车:ActionOutput.have_retry 为假时(base_agent.py:681-683),即使没通过也立刻 break,不再重试。用于"这类失败重试也没用"的场景,避免空转。
  • terminate 是强制终止信号:ActionOutput.terminate 为真会强行结束对话循环(action/base.py:59-62 注释、base_agent.py:704),用于"任务彻底完成"的显式收尾。
  • 超时兜底:每轮结束检查 time_cost > self.max_timeout(base_agent.py:707-713),防止单个 Agent 把整条会话拖死。
  • 异常被兜住:整个 generate_reply 包在 try 里,出异常返回一条 success=False 的消息而非崩溃(base_agent.py:728-732),保证多 Agent 编排不会因为一个 Agent 炸了而整体挂掉。
  • 全程有 trace:每一步都用 root_tracer.start_span 包了(think/review/act/verify 各一个 span),conv_id、LLM 回复、动作输出都进了 metadata——调试一次失败的纠错链路时,trace 是你的眼睛。

1.7 代码地图​

主题文件路径(相对 packages/dbgpt-core/src/)关键符号
五步循环主体dbgpt/agent/core/base_agent.pyConversableAgent.generate_reply
thinkdbgpt/agent/core/base_agent.pyConversableAgent.thinking
actdbgpt/agent/core/base_agent.pyConversableAgent.act
verify(基类三层检查)dbgpt/agent/core/base_agent.pyConversableAgent.verify
动作解析dbgpt/agent/core/action/base.pyAction.parse_action, ActionOutput
具体 verify(连库跑SQL)dbgpt/agent/expand/data_scientist_agent.pyDataScientistAgent.correctness_check
运行模式枚举dbgpt/agent/core/role.pyAgentRunMode