AI Agent 三大核心方向深度解析：Agentic 工作流、多智能体系统与多模态 RAG

1. 引言

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

2. Agentic 工作流

2.1 必读论文与核心思想

2.2 源码研读：超越 LangChain

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

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

# 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 函数实现示例：

import json
import time
from typing import Any, Dict, Optional

def 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, 4s
print(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 上下文压缩方法实现示例：

from typing import List, Dict

def 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 和 content
max_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 = 3
recent_messages = conversation_history[–recent_count:]
older_messages = conversation_history[:–recent_count]

# 4. 对早期历史进行摘要压缩
if older_messages and summary_model:
# 将早期历史拼接成文本
older_text = "\\n".join(
f"[{msg.get('role', 'user')}]: {msg.get('content', '')}"
for msg in older_messages
)

# 调用 LLM 生成摘要（模拟调用）
summary_prompt = f"请对以下对话历史做简洁摘要（保留关键决策和结果）：\\n{older_text}"
# summary = summary_model.generate(summary_prompt) # 实际调用
summary = f"[摘要] 早期对话共 {len(older_messages)} 条，涉及任务规划与工具调用，关键决策已保留。" # 模拟结果

# 5. 构建压缩后的上下文
compressed = [
{"role": "system", "content": f"以下是早期对话摘要：{summary}"}
] + recent_messages
else:
# 无摘要模型时，直接截断早期历史
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 可靠性对比

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

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 result
except 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:
continue
if self.execution_graph.nodes[step].get("status") == "success":
return step
return "root"

3. 多智能体系统

3.1 必读论文与架构对比

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

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

# AutoGen 核心控制流（简化）
class GroupChatManager:
"""管理多 Agent 对话的调度器"""

def __init__(self, agents: list, max_round: int = 50):
self.agents = agents
self.max_round = max_round
self.round = 0
self.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 selected

def 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()
continue

speaker = 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 核心问题：死锁与无限循环处理

死锁检测策略：

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-B
return True

# 模式3：内容重复（语义相似度 > 0.95）
if len(recent) >= 2:
sim = cosine_similarity(
embed(recent[–1]["content"]),
embed(recent[–2]["content"])
)
if sim > 0.95:
return True

return False

def _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. 自然语言

推荐方案：混合协议

{
"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 必读论文与技术演进

4.2 ColPali 核心技术解析

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

传统方案 vs. ColPali：

传统方案：
PDF → OCR → 文本提取 → 文本分块 → Text Embedding → 向量检索
↓
问题：表格错乱、图表丢失、公式变形

ColPali 方案：
PDF → 页面截图 → 视觉编码器 → 多模态 Embedding → 视觉检索
↓
优势：保留原始布局、理解图表、无需 OCR

ColPali 核心流程：

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):
# 直接编码整页图像，无需 OCR
embedding = 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 页面

优势与局限：

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

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

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 三大方向对比

5.2 推荐学习顺序

5.3 关键代码仓库

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