Skip to content

12|两层循环:会话与单轮分离

前两阶段让 agent 能听懂任务、看到对的材料。现在让它动起来:调模型 → 模型说查日志 → 执行查询 → 回填结果 → 再调模型。

这个循环怎么设计?最直觉的写法是一个 while 把所有事都干了:调模型、执行工具、存历史、判断停止。能跑,但混在一起会出问题。本篇讲第一个循环设计决策:把会话管理和单轮执行分成两层。

单层循环为什么不够

一个典型的单层循环:

python
def run_agent(messages):
    while True:
        response = client.responses.create(model=settings.model, input=messages)
        if response.output_text and not response.output[0].type == "function_call":
            return response.output_text   # 没有工具调用,结束
        tool_call = response.output[0]
        result = run_tool_call(tool_call)
        messages.append(tool_call)
        messages.append({"type": "function_call_output", "call_id": tool_call.call_id, "output": json.dumps(result)})
        save_session(messages)   # 持久化混在执行里

工具少、不需要持久化、没有子 agent 时,这个循环完全够用。但需求一长就乱:要支持会话恢复,持久化逻辑混在执行循环里;要加错误恢复,恢复逻辑和工具执行纠缠;要派子 agent,子 agent 复用哪段逻辑不清楚。

问题出在把两件事混在了一起:会话级的管理(多轮状态、持久化、整个会话何时结束)和单轮的执行(一次 API 调用、工具执行、本轮是否继续)。这两件事变化频率不同、关注点不同,混在一起改一处就容易碰到另一处。

分成两层

把循环拆成两层:QueryEngine 层和 queryLoop 层。

AIOps 概念图:会话层与单轮层

QueryEngine 层管会话:多轮状态、transcript 持久化、SDK 协议适配、usage 统计、整个会话何时结束。它不关心单轮内部怎么执行。

queryLoop 层管单轮:一次 API 调用、工具执行、错误恢复、本轮继续还是终止。它不关心会话级的状态怎么存。

python
class QueryEngine:
    def __init__(self, client, settings):
        self.client = client
        self.settings = settings
        self.state = SessionState(...)

    def run(self, user_input):
        self.state.messages.append({"role": "user", "content": user_input})
        result = self.query_loop()
        save_session(self.state)   # 持久化在会话层
        return result

    def query_loop(self):
        while True:
            response = self.client.responses.create(
                model=self.settings.model, input=self.state.messages,
            )
            if not self._has_tool_call(response):
                self.state.messages.append({"role": "assistant", "content": response.output_text})
                return response.output_text   # 本轮无工具调用,单轮结束
            tool_call = response.output[0]
            result = run_tool_call(tool_call)
            self.state.messages.append(tool_call)
            self.state.messages.append({
                "type": "function_call_output",
                "call_id": tool_call.call_id,
                "output": json.dumps(result),
            })
            # 有工具调用,继续下一轮 API 调用

QueryEngine 的 runquery_loopquery_loop 内部循环执行单轮。持久化在 QueryEngine 层做,不在 queryLoop 里。

分离的价值

两层分离的价值不在"架构好看",在三个具体好处。

第一,独立测试。queryLoop 输入一组消息,验证输出是否符合预期,不需要启动完整会话。QueryEngine 用 mock 的 queryLoop 验证会话状态转换。两层各自测,问题定位快。

第二,独立演进。改 queryLoop 的工具执行策略(比如加并发),不会碰到 QueryEngine 的持久化逻辑。改 QueryEngine 的存储方式(比如从文件换 Redis),不影响 queryLoop。

第三,子 agent 复用。后面讲多 agent 时,子 agent 复用的就是 queryLoop 这套执行机制,只是上下文、工具集合、权限不同。如果执行逻辑和会话管理混在一起,子 agent 没法干净地复用。

不是为架构而架构

这里要纠正一个倾向:不是所有场景都要分两层。

系统工具不多、没有持久化需求、没有子 agent 时,单层循环完全够用。引入两层模型的时机是出现"同一套执行逻辑要在多个会话里复用"或"会话状态需要独立于执行策略演进"的需求。过早分层是过度设计,过晚分层是技术债。

判断标准是看混在一起的代价是否已经显现:改执行逻辑经常碰会话状态、加错误恢复无处下手、想派子 agent 发现逻辑抽不出来。这些痛感出现了,才值得分两层。

状态快照

QueryEngine 层有个设计点:状态快照的保存频率。

AIOps 概念图:状态快照

每轮都写盘能保证最高可靠性,但带来 I/O 延迟和并发竞争。内存中维护状态,每几轮或遇到关键操作(写操作工具执行前)时同步落盘。N 的取值取决于故障恢复要求——能接受丢失最近几轮对话的,N 取大些;不能丢失任何一轮的,每轮同步落盘。这是可靠性和性能的取舍。

这套两层循环加状态快照的设计,在 LangGraph 里对应 StateGraph——节点是处理步骤、边是流转、状态显式管理、检查点持久化。本系列手写一遍理解原理,生产里用 LangGraph 等于用了它的现成实现(第 25 篇展开框架对应关系)。

循环之后

两层循环的骨架搭起来了:QueryEngine 管会话,queryLoop 管单轮。但循环跑着会出错——上下文超长、网络超时、工具执行失败。这些错误怎么处理?

下一篇讲错误处理。核心设计决策:有些错误内部先扣留自己修,有些直接抛外层。这个判断比"捕获所有异常"复杂得多。