1. 引言

2024-2025 年，AI Agent 从概念验证走向工程落地。无论是单 Agent 的推理-行动循环，还是多 Agent 协作系统，亦或是多模态检索增强生成，都已成为构建下一代 AI 应用的核心范式。本文从论文研读、源码剖析、核心问题三个维度，系统梳理 Agentic 工作流、多智能体系统、多模态 RAG 三大方向的关键技术。

2. Agentic 工作流

2.1 必读论文与核心思想

论文	核心贡献	适用场景
ReAct (Yao et al., 2023)	将推理（Reasoning）与行动（Acting）交织，通过 Thought-Action-Observation 循环实现动态决策	知识密集型问答、工具调用
Plan-and-Solve (Wang et al., 2023)	先制定完整计划再逐步执行，减少 ReAct 的试错成本	长流程任务、多步骤推理
Tree of Thoughts (Yao et al., 2023)	在推理树的多个分支上并行探索，用 BFS/DFS 搜索最优路径	数学推理、创意写作
Gorilla (Patil et al., 2023)	通过检索增强的 API 调用，让 LLM 学会调用数千个工具	工具调用、API 编排
Self-Refine (Madaan et al., 2023)	生成→反馈→改进的迭代循环，无需外部监督	代码生成、文本优化

2.2 源码研读：超越 LangChain

LangChain 提供了便捷的 Agent 抽象，但真正的工程细节藏在 AutoGPT 和 BabyAGI 的核心循环中。

AutoGPT 核心循环（简化伪代码）：

代码语言：python

先拆解一下 AutoGPT 的核心循环逻辑，这其实是理解 AI Agent 运作机制的关键。整个过程可以概括为三个步骤：

# AutoGPT 核心循环 while True: # 1. 状态管理：维护上下文窗口 context = build_context( recent_messages=memory.recent(5), # 最近5条 relevant_memories=memory.query(current_task), # 向量检索 system_prompt=load_prompt("agent.yaml") # 系统提示词 ) # 2. 决策：LLM 输出 JSON 格式的下一步行动 action = llm.generate( prompt=context, response_format={"type": "json_object"} ) # 3. 执行与错误处理 try: result = execute_action(action) memory.add(action, result) except APIError as e: # 关键：错误重试 上下文压缩 if e.is_retryable(): action = retry_with_backoff(action, max_retries=3) else: memory.summarize_and_prune() # 压缩上下文 action = {"thought": "策略调整", "tool": "replan"}

理解了这套机制，我们再来看 BabyAGI 在状态管理上的独特亮点：

使用 TaskQueue 维护优先级队列，支持任务依赖上下文压缩策略：当 token 超限时，用 LLM 对历史做摘要错误隔离：单个子任务失败不影响整体流程

AutoGPT execute_action 函数实现示例：

代码语言：python

复制

import jsonimport timefrom typing import Any, Dict, Optionaldef execute_action(action: Dict[str, Any], max_retries: int = 3) -> str:"""执行 AutoGPT 决策出的行动，包含重试与错误处理逻辑。Args:action: LLM 输出的 JSON 行动指令，格式如：{"thought": "需要查询天气", "tool": "web_search", "args": {"query": "北京今日天气"}}max_retries: 最大重试次数Returns:执行结果字符串"""tool_name = action.get("tool", "unknown")args = action.get("args", {})for attempt in range(max_retries):try:# 模拟工具调用（实际应接入真实 API）if tool_name == "web_search":result = f"【搜索结果】{args.get('query', '')} 的相关信息已获取"elif tool_name == "code_executor":result = f"【代码执行】已运行代码片段，输出：Hello World"elif tool_name == "file_write":result = f"【文件写入】已将内容写入 {args.get('filename', 'output.txt')}"else:result = f"【未知工具】{tool_name} 未注册，跳过执行"# 模拟成功返回return f"[{tool_name}] 执行成功: {result}"except ConnectionError as e:# 网络错误：指数退避重试if attempt < max_retries - 1:wait_time = 2 ** attempt# 1s, 2s, 4sprint(f"连接错误，{wait_time}秒后重试 (第{attempt 1}/{max_retries}次)")time.sleep(wait_time)else:return f"[{tool_name}] 执行失败: 已达最大重试次数，错误: {e}"except ValueError as e:# 参数错误：不重试，直接返回错误return f"[{tool_name}] 参数错误: {e}"return f"[{tool_name}] 执行失败: 未知错误"# 使用示例action_example = {"thought": "用户想了解 Python 列表推导式","tool": "web_search","args": {"query": "Python list comprehension 教程"}}result = execute_action(action_example)print(result)# 输出: [web_search] 执行成功: 【搜索结果】Python list comprehension 教程 的相关信息已获取

BabyAGI summarize_and_prune 上下文压缩方法实现示例：

代码语言：python

复制

from typing import List, Dictdef summarize_and_prune(conversation_history: List[Dict[str, str]],max_tokens: int = 2000,summary_model=None) -> List[Dict[str, str]]:"""当上下文窗口超限时，对历史对话进行智能压缩。策略：保留最近 N 条完整记录，对更早的历史做 LLM 摘要。Args:conversation_history: 对话历史列表，每条含 role 和 contentmax_tokens: 允许的最大 token 数（估算）summary_model: 用于生成摘要的 LLM 接口（可选）Returns:压缩后的对话历史"""# 1. 估算当前 token 数（简单按字符数估算，实际可用 tiktoken）total_chars = sum(len(msg.get("content", "")) for msg in conversation_history)estimated_tokens = total_chars // 4# 粗略估算# 2. 如果未超限，直接返回if estimated_tokens <= max_tokens:return conversation_history# 3. 保留最近 3 条完整记录（保证上下文连贯性）recent_count = 3recent_messages = conversation_history[-recent_count:]older_messages = conversation_history[:-recent_count]# 4. 对早期历史进行摘要压缩if older_messages and summary_model:# 将早期历史拼接成文本older_text = "".join(f"[{msg.get('role', 'user')}]: {msg.get('content', '')}"for msg in older_messages)# 调用 LLM 生成摘要（模拟调用）summary_prompt = f"请对以下对话历史做简洁摘要（保留关键决策和结果）：{older_text}"# summary = summary_model.generate(summary_prompt)# 实际调用summary = f"[摘要] 早期对话共 {len(older_messages)} 条，涉及任务规划与工具调用，关键决策已保留。"# 模拟结果# 5. 构建压缩后的上下文compressed = [{"role": "system", "content": f"以下是早期对话摘要：{summary}"}] recent_messageselse:# 无摘要模型时，直接截断早期历史compressed = recent_messages# 6. 再次检查 token 数compressed_chars = sum(len(msg.get("content", "")) for msg in compressed)print(f"上下文压缩完成: {total_chars} 字符 → {compressed_chars} 字符 "f"(压缩率: {compressed_chars / total_chars:.1%})")return compressed# 使用示例history = [{"role": "user", "content": "帮我写一个 Python 爬虫"},{"role": "assistant", "content": "好的，我来设计爬虫架构..."},{"role": "tool", "content": "已执行代码，返回页面 HTML"},{"role": "user", "content": "需要添加反爬处理"},{"role": "assistant", "content": "添加 User-Agent 轮换和袋里 IP..."},{"role": "tool", "content": "反爬策略已生效，请求成功"},{"role": "user", "content": "把结果保存到 CSV 文件"},{"role": "assistant", "content": "正在写入 CSV..."},]compressed_history = summarize_and_prune(history, max_tokens=100)print(f"压缩后共 {len(compressed_history)} 条消息")for msg in compressed_history:print(f"[{msg['role']}]: {msg['content'][:50]}...")

2.3 核心问题：ReAct vs. Plan-and-Execute 可靠性对比

维度	ReAct	Plan-and-Execute
长流程可靠性	低——每一步都可能偏离，累积错误	高——计划作为"锚点"约束执行
容错能力	强——可随时调整方向	弱——计划僵化，偏离后需重规划
Token 消耗	高——大量试错轨迹	中——计划 执行，但计划本身占 token
适用场景	探索性强、路径不确定的任务	步骤明确、可预见的流程

混合方案：动态回滚工作流引擎设计

代码语言：python

复制

class RollbackWorkflowEngine:"""支持动态回滚的工作流引擎"""def __init__(self):self.execution_graph = nx.DiGraph()# 有向无环图self.checkpoints = {}# 检查点快照self.rollback_stack = []def execute_step(self, step_id: str, context: dict):# 1. 创建检查点checkpoint = deepcopy(context)self.checkpoints[step_id] = checkpoint# 2. 执行步骤try:result = self._run_step(step_id, context)self.execution_graph.add_node(step_id, status="success", result=result)return resultexcept Exception as e:# 3. 失败时回滚到最近的成功检查点self.execution_graph.add_node(step_id, status="failed", error=str(e))rollback_to = self._find_safe_rollback_point(step_id)context = self.checkpoints[rollback_to]# 4. 重试或重新规划if self._should_retry(e):return self.execute_step(step_id, context)else:return self._replan_from(rollback_to, context)def _find_safe_rollback_point(self, failed_step: str) -> str:"""找到不依赖失败步骤的最近检查点"""ancestors = nx.ancestors(self.execution_graph, failed_step)for step in reversed(self.rollback_stack):if step in ancestors:continueif self.execution_graph.nodes[step].get("status") == "success":return stepreturn "root"

3. 多智能体系统

3.1 必读论文与架构对比

论文	核心架构	通信方式	关键创新
AutoGen (微软)	ConversableAgent GroupChatManager	消息路由	灵活的对话控制流
MetaGPT	角色化 Agent（产品经理/架构师/工程师）	结构化 SOP	软件工程全流程模拟
ChatDev	编程双 Agent（CEO CTO）	自然语言对话	角色扮演驱动开发
Generative Agents (斯坦福小镇)	记忆流 反思 计划	环境交互	可信的长期行为模拟

3.2 源码研读：AutoGen 对话控制流

AutoGen 的核心在于 ConversableAgent 和 GroupChatManager 的配合：

代码语言：python

复制

# AutoGen 核心控制流（简化）class GroupChatManager:"""负责调度多 Agent 对话的管理器"""def __init__(self, agents: list, max_round: int = 50):self.agents = agentsself.max_round = max_roundself.round = 0self.speaker_history = []def select_next_speaker(self, last_message: dict) -> ConversableAgent:"""决定下一位发言的 Agent"""# 策略1：轮流发言if self.round % len(self.agents) == 0:return self.agents[0]# 策略2：按内容路由（由 LLM 决策）prompt = f"""当前对话历史：{self.speaker_history[-3:]}最后消息：{last_message['content']}可用 Agent：{[a.name for a in self.agents]}请选择最合适回复的 Agent："""selected = llm.choose(prompt, choices=self.agents)return selecteddef run(self, initial_message: str):"""主循环——这里包含死锁检测机制"""self.speaker_history.append({"role": "user", "content": initial_message})while self.round < self.max_round:# 死锁检测：如果连续 3 轮都是同一个 Agent 发言if self._detect_deadlock():print("检测到死锁，开始触发干预策略")self._resolve_deadlock()continuespeaker = self.select_next_speaker(self.speaker_history[-1])reply = speaker.generate_reply(self.speaker_history)self.speaker_history.append({"role": speaker.name, "content": reply})self.round = 1

3.3 核心问题：死锁与无限循环处理

先看死锁检测策略：

代码语言：python

复制

def _detect_deadlock(self) -> bool:"""检测三种死锁模式"""recent = self.speaker_history[-6:]# 模式1：同一 Agent 重复发言if len(recent) >= 3:speakers = [msg["role"] for msg in recent[-3:]]if len(set(speakers)) == 1:return True# 模式2：两个 Agent 来回重复if len(recent) >= 4:pattern = [msg["role"] for msg in recent[-4:]]if pattern == pattern[0:2] * 2:# A-B-A-Breturn True# 模式3：内容重复（语义相似度 > 0.95）if len(recent) >= 2:sim = cosine_similarity(embed(recent[-1]["content"]),embed(recent[-2]["content"]))if sim > 0.95:return Truereturn Falsedef _resolve_deadlock(self):"""死锁解决策略"""# 策略1：引入第三方 Agent 仲裁arbiter = self.agents[-1]# 最后一个 Agent 作为仲裁者summary = arbiter.summarize(self.speaker_history[-10:])self.speaker_history.append({"role": "arbiter","content": f"仲裁意见：{summary}，建议下一步：..."})# 策略2：强制切换话题# 策略3：降低 max_round 并输出当前结果

Agent 间通信协议设计：JSON 结构 vs. 自然语言

维度	JSON 结构化协议	自然语言协议
解析效率	高——可直接反序列化	低——需 LLM 理解意图
歧义性	低——字段明确	高——依赖上下文
灵活性	低——需预定义 schema	高——可表达任意内容
Token 消耗	低	高
推荐场景	工具调用、状态同步	创意讨论、需求澄清

推荐方案：混合协议

代码语言：json

复制

{"protocol_version": "2.0","sender": "architect_agent","receiver": "coder_agent","intent": "request_code_review","payload": {"task_id": "TASK-0042","code_snippet": "def process(data): ...","review_focus": ["performance", "security"]},"natural_language_note": "这段代码在数据量大时可能 OOM，请重点检查内存管理。","metadata": {"timestamp": "2025-06-10T08:30:00Z","context_window_used": 3200}}

4. 多模态 RAG

4.1 必读论文与技术演进

论文	核心能力	技术路线
CLIP (OpenAI, 2021)	图文对齐	对比学习，双塔结构
BLIP-2 (Salesforce, 2023)	图文理解 生成	Q-Former 桥接视觉和语言
LLaVA (微软, 2023)	视觉对话	视觉编码器 LLM 投影
UniIR (2024)	统一多模态检索	多任务指令微调
ColPali (2024)	视觉文档检索	视觉模型直接理解 PDF

4.2 ColPali 核心技术解析

ColPali 彻底抛弃了传统的 OCR 文本 Embedding pipeline，直接用视觉模型理解 PDF 页面。

传统方案 vs. ColPali：

传统方案：

PDF → OCR → 文本提取 → 文本分块 → Text Embedding → 向量检索

问题：表格错乱、图表丢失、公式变形

ColPali 方案：

PDF → 页面截图 → 视觉编码器 → 多模态 Embedding → 视觉检索

优势：保留原始布局、理解图表、无需 OCR

ColPali 核心流程：

代码语言：python

复制

class ColPaliRetriever:"""基于视觉模型的 PDF 检索器"""def __init__(self, vision_model="google/paligemma-3b-mix-224"):self.encoder = load_vision_encoder(vision_model)self.index = FAISSIndex(dimension=1152)# PaliGemma 的视觉维度def index_pdf(self, pdf_path: str):"""将 PDF 页面编码为视觉向量"""images = pdf_to_images(pdf_path, dpi=150)# 每页转图片for page_num, img in enumerate(images):# 直接编码整页图像，无需 OCRembedding = self.encoder.encode(img)self.index.add_item(embedding=embedding,metadata={"page": page_num,"pdf_path": pdf_path})def search(self, query: str, top_k: int = 5):"""支持文本和图像查询"""if is_image(query):query_emb = self.encoder.encode(load_image(query))else:query_emb = self.encoder.encode_text(query)results = self.index.search(query_emb, top_k)return results# 返回匹配的 PDF 页面

优势与局限：

维度	传统 OCR Embedding	ColPali
表格理解	差——行列关系丢失	好——保留视觉布局
图表检索	差——无法理解图表语义	好——直接编码图像
多语言	依赖 OCR 语言包	好——视觉模型语言无关
推理速度	快（文本 Embedding 轻量）	慢（视觉模型计算量大）
存储成本	低（文本向量小）	高（图像向量大）
细粒度检索	好（可检索段落级）	差（目前以页为单位）

4.3 核心问题：复杂 PDF 的 RAG 系统设计

针对包含复杂表格、图表和图片的 PDF，设计一个支持"用图片搜文档"和"用图表数据生成分析"的 RAG 系统：

代码语言：python

复制

class MultiModalRAGSystem:"""支持图文互搜和图表分析的多模态 RAG"""def __init__(self):# 双通道索引self.visual_index = ColPaliRetriever()# 视觉通道self.text_index = TextEmbeddingRetriever()# 文本通道self.analyzer = ChartAnalyzer()# 图表分析器def index_document(self, pdf_path: str):"""双通道索引 PDF"""pages = pdf_to_images(pdf_path)for page_num, page_img in enumerate(pages):# 1. 视觉索引：整页编码self.visual_index.add_page(page_img, page_num)# 2. 文本索引：提取文本块text_blocks = extract_text_blocks(page_img)# 含表格文本for block in text_blocks:self.text_index.add_text(text=block["text"],metadata={"page": page_num,"bbox": block["bbox"],"type": block["type"]# text/table/chart})# 3. 图表检测与结构化提取charts = detect_charts(page_img)for chart in charts:chart_data = self.analyzer.extract_data(chart["image"])self.text_index.add_text(text=chart_data["summary"],metadata={"page": page_num,"type": "chart","chart_data": chart_data["raw_data"]})def search_by_image(self, query_image: str, top_k: int = 5):"""用图片搜文档"""return self.visual_index.search(query_image, top_k)def analyze_chart(self, query: str, page_num: int):"""用图表数据生成分析"""# 1. 检索相关图表chart_blocks = self.text_index.search(query=query,filter={"type": "chart", "page": page_num})# 2. 提取结构化数据chart_data = chart_blocks[0].metadata["chart_data"]# 3. LLM 生成分析analysis_prompt = f"""图表数据：{json.dumps(chart_data, ensure_ascii=False)}用户问题：{query}请基于图表数据生成分析报告，包含趋势、异常值和关键发现。"""return llm.generate(analysis_prompt)

5. 总结与学习路径

5.1 三大方向对比

方向	核心挑战	关键技能	推荐入门项目
Agentic 工作流	可靠性、状态管理、错误恢复	图算法、状态机设计	实现一个 ReAct Agent
多智能体系统	死锁、通信协议、角色设计	分布式系统、协议设计	复现 AutoGen 的 GroupChat
多模态 RAG	多模态对齐、检索精度、推理效率	视觉模型、向量检索	实现 ColPali 简化版

5.2 推荐学习顺序

基础阶段：掌握 ReAct 和 Plan-and-Solve 的核心循环进阶阶段：阅读 AutoGen 源码，理解对话控制流高阶阶段：实现一个支持动态回滚的工作流引擎实战阶段：构建一个多模态 RAG 系统，处理真实 PDF 文档

5.3 关键代码仓库

AutoGPT - Agent 核心循环BabyAGI - 任务管理AutoGen - 多 Agent 框架ColPali - 视觉文档检索