3、核心架构解析:Agent循环机制、任务队列管理、上下文窗口策略

AutoGPT 的核心,说白了就是一个永不停止的思考-行动循环。我刚开始研究这个项目时,第一反应是:这不就是个 while True 吗?后来深入源码才发现,里面的门道比我想象的多得多。

今天咱们就把这个循环拆开来看。我会结合自己踩过的坑,把 Agent 循环、任务队列和上下文窗口这三个核心机制讲透。

3.1 Agent 主循环:思考-行动-观察

AutoGPT 的 Agent 循环,本质上是一个感知-决策-执行的闭环。每次循环,Agent 都会经历三个步骤:

  1. 思考(Think):分析当前状态,决定下一步做什么
  2. 行动(Act):执行一个具体动作(调用工具、写文件、搜索等)
  3. 观察(Observe):获取行动结果,更新内部状态

我最初实现这个循环时,犯过一个低级错误——把思考步骤放在循环外面。结果 Agent 第一次思考后就再也不更新计划了,像个复读机一样重复执行同一个动作。嗯,这个坑你们别踩。

下面是一个简化版的 Agent 循环代码:

class AutoGPTAgent:
    def __init__(self):
        self.memory = []
        self.task_queue = TaskQueue()
        self.context = ContextWindow(max_tokens=4096)
    
    def run_cycle(self):
        while True:
            # 1. 思考:分析当前状态
            thought = self.think()
            
            # 2. 行动:执行决策
            action = self.decide_action(thought)
            result = self.execute_action(action)
            
            # 3. 观察:记录结果
            observation = self.observe(result)
            self.update_state(thought, action, observation)
            
            # 4. 检查是否应该继续
            if self.should_stop():
                break

关键点:每次循环都必须包含完整的思考-行动-观察链条。跳过任何一步,Agent 都会失去上下文连贯性。

3.2 任务队列管理:优先级与依赖

任务队列是 Agent 的「待办事项清单」。但和普通队列不同,AutoGPT 的任务队列需要处理优先级和依赖关系。

我记得有一次,Agent 同时收到了「搜索资料」和「写报告」两个任务。如果先写报告再搜索,报告内容就是空的。这就是典型的依赖问题。

我常用的任务队列实现方式:

class TaskQueue:
    def __init__(self):
        self.tasks = []  # 使用列表模拟优先级队列
        self.dependencies = {}  # 任务ID -> [依赖任务ID列表]
    
    def add_task(self, task, priority=0, depends_on=None):
        self.tasks.append({
            'id': task.id,
            'content': task,
            'priority': priority,
            'depends_on': depends_on or []
        })
        self.tasks.sort(key=lambda t: t['priority'], reverse=True)
    
    def get_next_task(self):
        """获取下一个可执行的任务"""
        for task in self.tasks:
            if all(self.is_completed(dep) for dep in task['depends_on']):
                return task
        return None

这里有个实际经验:优先级不要超过 3 级。我试过 5 级优先级,结果 Agent 总是在纠结「哪个任务更重要」,反而浪费了大量 token。

优先级级别 适用场景 示例
0(紧急) 必须立即执行 错误恢复、用户中断响应
1(高) 当前步骤的关键任务 获取必要数据、执行核心操作
2(普通) 常规任务 数据整理、日志记录
3(低) 可延后执行 优化、美化、非必要检查

避坑指南:我曾经把「检查网络连接」设为高优先级任务,结果 Agent 每次循环都去检查网络,导致主任务迟迟无法推进。后来我把这类「检查性任务」统一放到低优先级,只在空闲时执行。

3.3 上下文窗口策略:Token 的生死抉择

上下文窗口是 AutoGPT 最头疼的问题之一。LLM 的上下文长度有限(比如 GPT-4 的 8K/32K),但 Agent 运行时间越长,积累的历史就越多。

你想想看,如果 Agent 运行了 100 轮,每轮产生 500 token 的对话历史,那就是 50K token。早就超出上下文限制了。

我常用的上下文管理策略有三种:

  1. 滑动窗口:只保留最近 N 轮对话
  2. 摘要压缩:定期对历史进行摘要,用摘要替代原始内容
  3. 关键信息提取:只保留与当前任务相关的信息

下面是我在实际项目中使用的上下文窗口类:

class ContextWindow:
    def __init__(self, max_tokens=4096):
        self.max_tokens = max_tokens
        self.history = []
        self.summary = ""
        self.current_tokens = 0
    
    def add(self, content, token_count):
        self.history.append({
            'content': content,
            'tokens': token_count
        })
        self.current_tokens += token_count
        
        # 如果超出限制,触发压缩
        if self.current_tokens > self.max_tokens * 0.8:
            self.compress()
    
    def compress(self):
        """压缩历史记录"""
        # 策略1:移除最早的历史
        while self.current_tokens > self.max_tokens * 0.6:
            oldest = self.history.pop(0)
            self.current_tokens -= oldest['tokens']
        
        # 策略2:生成摘要(实际项目中会调用LLM)
        self.summary = self.generate_summary(self.history)
        
        # 策略3:保留摘要 + 最近3轮完整历史
        self.history = self.history[-3:]

注意:压缩策略不能太激进。我试过只保留摘要,结果 Agent 丢失了大量细节信息,回答变得非常模糊。建议保留至少 2-3 轮完整历史,再加上摘要。

3.4 三者如何协同工作

Agent 循环、任务队列、上下文窗口不是孤立的。它们之间的关系可以用一张图来表示:

AutoGPT 核心架构协同图 Agent 主循环 思考 (Think) 行动 (Act) 观察 (Observe) 任务队列 优先级 + 依赖管理 上下文窗口 Token 预算 + 压缩策略 任务队列提供「做什么」 上下文窗口提供「记得什么」

从图中可以看出:

  • Agent 循环是驱动引擎,负责调度思考-行动-观察
  • 任务队列为循环提供「下一步做什么」的决策依据
  • 上下文窗口为思考和观察提供历史信息

三者配合得好,Agent 就能持续稳定地运行。配合不好,就会出现各种奇怪的问题。

我的经验:调试这三个组件时,建议先单独测试每个组件,再组合测试。我曾经直接组合测试,结果出了问题根本不知道是哪个环节的锅。分开测,每个组件都跑通,再合起来,效率高得多。

3.5 实战中的调优建议

最后分享几个我在实际项目中总结的调优经验:

  1. 循环频率控制:不要无限制地跑循环。我习惯每 10 轮检查一次「是否达成目标」,避免 Agent 陷入死循环。
  2. 任务队列清理:已完成的任务要及时移除。我见过一个项目,任务队列里积累了上千条已完成任务,每次遍历都卡半天。
  3. 上下文窗口预留:不要把 token 用满。我一般留 20% 的余量给突发情况(比如 Agent 突然需要处理一个长文档)。
  4. 日志记录:每次循环都记录关键指标(token 消耗、任务完成数、循环次数)。出了问题,日志就是你的救命稻草。

小技巧:在 Agent 循环中加入一个「心跳」机制。如果连续 5 轮没有实质进展(比如任务队列没有减少),就触发一个警告。这个机制帮我抓到了好几次 Agent 卡在某个步骤上的问题。

好了,核心架构就讲到这里。这三个机制是 AutoGPT 的骨架,理解透了,后面的调优才能有的放矢。