第十七章：Agent 架构模式深度解析#
mindmap
root((Agent架构模式))
ReAct模式
推理+行动
交替执行
Plan-and-Execute
先规划后执行
任务分解
Reflexion模式
自我反思
经验学习
Tool Use模式
工具选择
工具调用
RAG增强
知识检索
上下文注入
记忆系统
短期记忆
长期记忆
工作记忆
多Agent协作
层级式
对等式
竞争式
17.1 Agent 架构模式概览#
Agent 的架构模式决定了它如何思考、决策和行动。不同的架构模式适用于不同类型的任务，理解这些模式是构建高效 Agent 系统的基础。本章将深入解析六种核心架构模式，并为每种模式提供完整的 Python 代码实现。
┌─────────────────────────────────────────────────────┐
│ Agent 架构模式谱系 │
├─────────────────────────────────────────────────────┤
│ │
│ ReAct ──── 推理与行动交替，最基础的 Agent 模式 │
│ │ │
│ Plan-and-Execute ── 先规划后执行，适合复杂任务 │
│ │ │
│ Reflexion ── 加入自我反思，从失败中学习 │
│ │ │
│ Tool Use ── 专注于工具选择与编排 │
│ │ │
│ RAG-Agent ── 检索增强的 Agent │
│ │ │
│ Multi-Agent ── 多个 Agent 协作 │
│ │
└─────────────────────────────────────────────────────┘
17.2 ReAct 模式：推理与行动的交替#
ReAct（Reasoning + Acting）是最经典的 Agent 架构模式，由 Yao et al. (2022) 提出。其核心思想是让 LLM 在推理（Thought）和行动（Action）之间交替进行，每次行动后观察结果（Observation），再进行下一轮推理。
import json
from openai import OpenAI
from typing import Dict, Any, Callable, Optional
class ReActAgent:
"""ReAct 模式 Agent 实现"""
SYSTEM_PROMPT = """你是一个遵循 ReAct 模式的 AI Agent。
对于每个问题，你需要交替进行思考和行动。
可用工具:
{tools_description}
请严格按照以下 JSON 格式输出:
{{
"thought": "你的思考过程",
"action": "工具名称（如果需要调用工具）或 'finish'（如果任务完成）",
"action_input": "工具的输入参数或最终答案"
}}
"""
def __init__(self, model: str = "gpt-4o"):
self.client = OpenAI()
self.model = model
self.tools: Dict[str, Callable] = {}
self.tools_description: Dict[str, str] = {}
def register_tool(self, name: str, func: Callable, description: str):
"""注册工具"""
self.tools[name] = func
self.tools_description[name] = description
def run(self, query: str, max_steps: int = 8) -> str:
"""执行 ReAct 循环"""
tools_desc = "\n".join(
f"- {name}: {desc}" for name, desc in self.tools_description.items()
)
messages = [
{"role": "system", "content": self.SYSTEM_PROMPT.format(
tools_description=tools_desc
)},
{"role": "user", "content": query}
]
for step in range(max_steps):
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
response_format={"type": "json_object"},
temperature=0.1
)
result = json.loads(response.choices[0].message.content)
thought = result.get("thought", "")
action = result.get("action", "")
action_input = result.get("action_input", "")
print(f"\n--- Step {step + 1} ---")
print(f"Thought: {thought}")
print(f"Action: {action}")
print(f"Action Input: {action_input}")
if action == "finish":
return action_input
# 执行工具调用
if action in self.tools:
try:
observation = self.tools[action](action_input)
except Exception as e:
observation = f"工具执行错误: {str(e)}"
else:
observation = f"未知工具: {action}"
print(f"Observation: {observation}")
# 将结果加入对话历史
messages.append({"role": "assistant", "content": json.dumps(result, ensure_ascii=False)})
messages.append({"role": "user", "content": f"Observation: {observation}"})
return "达到最大步骤数，任务未完成"
# 使用示例
def search_web(query: str) -> str:
"""模拟网络搜索"""
mock_results = {
"Python 3.12 新特性": "Python 3.12 引入了改进的错误消息、f-string 改进、类型参数语法等",
"LangChain 最新版本": "LangChain 0.3.x 已发布，重构了核心架构",
}
for key, value in mock_results.items():
if key in query:
return value
return f"搜索 '{query}' 的结果：未找到相关信息"
def calculator(expression: str) -> str:
"""安全计算器"""
try:
allowed_chars = set("0123456789+-*/.() ")
if all(c in allowed_chars for c in expression):
return str(eval(expression))
return "不安全的表达式"
except Exception as e:
return f"计算错误: {e}"
# agent = ReActAgent()
# agent.register_tool("search", search_web, "搜索网络信息")
# agent.register_tool("calculator", calculator, "数学计算")
# result = agent.run("Python 3.12 有哪些新特性？请列出至少3个。")
17.3 Plan-and-Execute 模式：先规划后执行#
Plan-and-Execute 模式将任务处理分为两个明确的阶段：首先由 Planner 制定完整的执行计划，然后由 Executor 逐步执行。这种模式特别适合需要多步骤协调的复杂任务。
from dataclasses import dataclass, field
from typing import List
from enum import Enum
class StepStatus(Enum):
PENDING = "pending"
RUNNING = "running"
COMPLETED = "completed"
FAILED = "failed"
SKIPPED = "skipped"
@dataclass
class PlanStep:
"""计划中的单个步骤"""
step_id: int
description: str
tool: str
tool_input: str
dependencies: List[int] = field(default_factory=list)
status: StepStatus = StepStatus.PENDING
result: Optional[str] = None
@dataclass
class ExecutionPlan:
"""执行计划"""
goal: str
steps: List[PlanStep] = field(default_factory=list)
class PlanAndExecuteAgent:
"""Plan-and-Execute 模式 Agent"""
def __init__(self, model: str = "gpt-4o"):
self.client = OpenAI()
self.model = model
self.tools: Dict[str, Callable] = {}
def plan(self, goal: str) -> ExecutionPlan:
"""第一阶段：制定计划"""
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": """你是一个任务规划专家。
请将用户的目标分解为具体的执行步骤。
输出 JSON 格式:
{
"steps": [
{
"step_id": 1,
"description": "步骤描述",
"tool": "工具名称",
"tool_input": "工具输入",
"dependencies": []
}
]
}"""},
{"role": "user", "content": f"目标: {goal}\n可用工具: {list(self.tools.keys())}"}
],
response_format={"type": "json_object"}
)
plan_data = json.loads(response.choices[0].message.content)
steps = [PlanStep(**s) for s in plan_data["steps"]]
return ExecutionPlan(goal=goal, steps=steps)
def execute(self, plan: ExecutionPlan) -> str:
"""第二阶段：执行计划"""
results = {}
for step in plan.steps:
# 检查依赖是否满足
deps_met = all(
plan.steps[d - 1].status == StepStatus.COMPLETED
for d in step.dependencies
)
if not deps_met:
step.status = StepStatus.SKIPPED
continue
step.status = StepStatus.RUNNING
print(f"执行步骤 {step.step_id}: {step.description}")
try:
if step.tool in self.tools:
# 替换输入中的变量引用
tool_input = step.tool_input
for sid, res in results.items():
tool_input = tool_input.replace(f"{{step_{sid}_result}}", str(res))
step.result = self.tools[step.tool](tool_input)
step.status = StepStatus.COMPLETED
results[step.step_id] = step.result
else:
step.status = StepStatus.FAILED
step.result = f"工具 {step.tool} 不存在"
except Exception as e:
step.status = StepStatus.FAILED
step.result = str(e)
return self._synthesize_results(plan, results)
def _synthesize_results(self, plan: ExecutionPlan, results: dict) -> str:
"""综合所有步骤结果生成最终答案"""
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "根据执行结果生成最终答案。"},
{"role": "user", "content": f"目标: {plan.goal}\n执行结果: {json.dumps(results, ensure_ascii=False)}"}
]
)
return response.choices[0].message.content
def run(self, goal: str) -> str:
"""完整的 Plan-and-Execute 流程"""
plan = self.plan(goal)
return self.execute(plan)
17.4 Reflexion 模式：自我反思与改进#
Reflexion 模式（Shinn et al., 2023）在 ReAct 基础上增加了自我反思机制。Agent 在执行失败或结果不理想时，会分析失败原因并生成改进策略，然后在下一次尝试中应用这些经验。
@dataclass
class ReflexionMemory:
"""反思记忆"""
attempt: int
task: str
trajectory: List[Dict[str, Any]]
outcome: str
success: bool
reflection: str # 反思总结
class ReflexionAgent:
"""Reflexion 模式 Agent"""
def __init__(self, model: str = "gpt-4o"):
self.client = OpenAI()
self.model = model
self.tools: Dict[str, Callable] = {}
self.reflections: List[ReflexionMemory] = []
def reflect(self, task: str, trajectory: list, outcome: str) -> str:
"""对执行过程进行反思"""
past_reflections = "\n".join(
f"尝试 {r.attempt}: {r.reflection}" for r in self.reflections[-3:]
)
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": """你是一个善于反思的 AI。
分析之前的执行过程，找出失败原因，并提出具体的改进策略。
重点关注：
1. 哪些步骤是正确的？
2. 哪些步骤导致了错误？
3. 下次应该如何改进？"""},
{"role": "user", "content": f"""
任务: {task}
执行轨迹: {json.dumps(trajectory, ensure_ascii=False)}
结果: {outcome}
之前的反思: {past_reflections if past_reflections else '无'}
"""}
]
)
return response.choices[0].message.content
def evaluate(self, task: str, result: str) -> bool:
"""评估结果是否满足任务要求"""
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "评估结果是否正确完成了任务。只回答 true 或 false。"},
{"role": "user", "content": f"任务: {task}\n结果: {result}"}
]
)
return "true" in response.choices[0].message.content.lower()
def run(self, task: str, max_attempts: int = 3) -> str:
"""带反思的执行循环"""
for attempt in range(max_attempts):
# 构建包含历史反思的提示
reflections_context = ""
if self.reflections:
reflections_context = "从之前的尝试中学到的经验:\n" + "\n".join(
f"- {r.reflection}" for r in self.reflections[-3:]
)
# 执行任务（使用 ReAct 内循环）
trajectory = []
result = self._execute_with_react(task, reflections_context, trajectory)
# 评估结果
success = self.evaluate(task, result)
if success:
print(f"✅ 第 {attempt + 1} 次尝试成功!")
return result
# 反思失败原因
reflection = self.reflect(task, trajectory, result)
self.reflections.append(ReflexionMemory(
attempt=attempt + 1,
task=task,
trajectory=trajectory,
outcome=result,
success=False,
reflection=reflection
))
print(f"❌ 第 {attempt + 1} 次尝试失败，反思: {reflection[:100]}...")
return f"经过 {max_attempts} 次尝试仍未成功"
def _execute_with_react(self, task, context, trajectory) -> str:
"""内部 ReAct 执行循环"""
# 简化实现，实际中复用 ReActAgent 逻辑
messages = [
{"role": "system", "content": f"完成以下任务。{context}"},
{"role": "user", "content": task}
]
response = self.client.chat.completions.create(
model=self.model, messages=messages
)
result = response.choices[0].message.content
trajectory.append({"action": "generate", "result": result})
return result
17.5 Tool Use 模式：工具选择与编排#
Tool Use 模式专注于如何让 Agent 高效地选择和使用工具。OpenAI 的 Function Calling 是这一模式的典型实现：
class ToolUseAgent:
"""Tool Use 模式 Agent - 基于 OpenAI Function Calling"""
def __init__(self, model: str = "gpt-4o"):
self.client = OpenAI()
self.model = model
self.tool_functions: Dict[str, Callable] = {}
self.tool_schemas: List[Dict] = []
def register_tool(self, name: str, func: Callable, schema: dict):
"""注册工具及其 JSON Schema"""
self.tool_functions[name] = func
self.tool_schemas.append({
"type": "function",
"function": {
"name": name,
**schema
}
})
def run(self, query: str) -> str:
"""执行工具调用循环"""
messages = [{"role": "user", "content": query}]
while True:
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
tools=self.tool_schemas if self.tool_schemas else None,
)
msg = response.choices[0].message
messages.append(msg)
if msg.tool_calls:
for tool_call in msg.tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
print(f"调用工具: {func_name}({func_args})")
if func_name in self.tool_functions:
result = self.tool_functions[func_name](**func_args)
else:
result = f"未知工具: {func_name}"
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": str(result)
})
else:
return msg.content
17.6 RAG 深度解析：从原理到 LlamaIndex 实战#
RAG（Retrieval-Augmented Generation，检索增强生成）是 Agent 系统中最重要的架构模式之一。它解决了 LLM 的核心痛点：知识截止日期和幻觉问题。通过在生成前检索相关文档，RAG 让 Agent 能够基于真实数据回答问题。
17.6.1 为什么需要 RAG#
LLM 有三个根本性限制：
知识截止：训练数据有截止日期，无法回答最新问题
幻觉问题：对不确定的问题会"编造"看似合理的答案
私有数据：无法访问企业内部文档、代码库、数据库
RAG 的核心思想很简单：先搜索，再回答。
传统 LLM 流程：
用户提问 → LLM 生成答案（可能幻觉）
RAG 流程：
用户提问 → 检索相关文档 → 将文档作为上下文 → LLM 基于文档生成答案
17.6.2 RAG 的完整架构#
一个生产级 RAG 系统包含两个阶段：
离线阶段（Indexing）：
原始文档 → 文档加载 → 文本分块 → 向量化（Embedding）→ 存入向量数据库
│ │ │ │ │
PDF Loader Chunking Embedding Model Vector Store
Word Strategy (text-embedding-3) (Chroma/Pinecone)
HTML
Code
Markdown
在线阶段（Querying）：
用户查询 → 查询改写 → 向量化 → 相似度检索 → 重排序 → 构建 Prompt → LLM 生成
│ │ │ │ │ │ │
原始问题 Query Embedding Top-K Reranker Context Response
Rewrite Retrieval Assembly
17.6.3 LlamaIndex 框架介绍#
LlamaIndex（原 GPT Index）是最流行的 RAG 框架，专注于将私有数据与 LLM 连接。它提供了从数据加载到查询的完整工具链。
为什么选择 LlamaIndex：
数据连接器丰富：支持 PDF、Word、HTML、数据库、API 等 160+ 数据源
索引策略多样：向量索引、关键词索引、知识图谱索引、树索引
查询引擎强大：支持子问题分解、多步推理、SQL 查询
与 Agent 深度集成：可作为 Agent 的工具使用
生产就绪：支持流式输出、异步、可观测性
安装：
pip install llama-index llama-index-llms-openai llama-index-embeddings-openai
pip install llama-index-vector-stores-chroma
pip install llama-index-readers-file
17.6.4 实战一：基础 RAG — 30 行代码构建文档问答#
"""
最简 RAG 示例：加载文档 → 建索引 → 查询
"""
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding
# 1. 配置 LLM 和 Embedding 模型
Settings.llm = OpenAI(model="gpt-4o", temperature=0.1)
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")
# 2. 加载文档（支持 PDF、TXT、MD、DOCX 等）
documents = SimpleDirectoryReader("./docs").load_data()
print(f"加载了 {len(documents)} 个文档")
# 3. 构建向量索引（自动分块 + 向量化 + 存储）
index = VectorStoreIndex.from_documents(documents)
# 4. 创建查询引擎
query_engine = index.as_query_engine(similarity_top_k=5)
# 5. 查询
response = query_engine.query("项目的技术架构是什么？")
print(response)
# 查看检索到的源文档
for node in response.source_nodes:
print(f" 来源: {node.metadata.get('file_name', '未知')}")
print(f" 相关度: {node.score:.4f}")
print(f" 内容片段: {node.text[:100]}...")
这 30 行代码背后，LlamaIndex 自动完成了：
文档解析（识别文件格式，提取文本）
文本分块（默认 1024 token，重叠 200 token）
向量化（调用 Embedding API）
存储（默认内存存储）
检索（余弦相似度搜索）
Prompt 构建（将检索结果注入 Prompt）
LLM 调用（生成最终答案）
17.6.5 实战二：生产级 RAG — 精细控制每个环节#
"""
生产级 RAG：精细控制分块、索引、检索、生成的每个环节
"""
from llama_index.core import (
VectorStoreIndex, StorageContext, Settings, PromptTemplate,
)
from llama_index.core.node_parser import (
SentenceSplitter, SemanticSplitterNodeParser,
)
from llama_index.core.extractors import (
TitleExtractor, SummaryExtractor, QuestionsAnsweredExtractor,
)
from llama_index.core.ingestion import IngestionPipeline
from llama_index.core.postprocessor import (
SimilarityPostprocessor, KeywordNodePostprocessor,
)
from llama_index.core.response_synthesizers import get_response_synthesizer
from llama_index.core.retrievers import VectorIndexRetriever
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.llms.openai import OpenAI
from llama_index.vector_stores.chroma import ChromaVectorStore
import chromadb
# ── 第一步：配置 ──────────────────────────────
Settings.llm = OpenAI(model="gpt-4o", temperature=0.1)
Settings.embed_model = OpenAIEmbedding(
model="text-embedding-3-small",
dimensions=512, # 降维以节省存储和提高速度
)
# ── 第二步：文档加载与预处理 ──────────────────
from llama_index.readers.file import PDFReader, DocxReader, MarkdownReader
documents = []
documents.extend(PDFReader().load_data("./docs/architecture.pdf"))
documents.extend(MarkdownReader().load_data("./docs/api-guide.md"))
# 为文档添加元数据
for doc in documents:
doc.metadata["project"] = "my-project"
doc.metadata["indexed_at"] = "2026-03-04"
# ── 第三步：文本分块策略（关键！）────────────
# 策略 A：固定大小分块（简单高效）
sentence_splitter = SentenceSplitter(
chunk_size=512, # 每块最大 512 token
chunk_overlap=50, # 块间重叠 50 token
paragraph_separator="\n\n",
)
# 策略 B：语义分块（按语义边界切分，效果更好但更慢）
semantic_splitter = SemanticSplitterNodeParser(
buffer_size=1,
breakpoint_percentile_threshold=95,
embed_model=Settings.embed_model,
)
nodes = sentence_splitter.get_nodes_from_documents(documents)
print(f"分块后得到 {len(nodes)} 个节点")
# ── 第四步：元数据增强（提高检索质量的关键）──
pipeline = IngestionPipeline(
transformations=[
sentence_splitter,
TitleExtractor(nodes=3),
SummaryExtractor(summaries=["self"]),
QuestionsAnsweredExtractor(questions=3),
Settings.embed_model,
]
)
enriched_nodes = pipeline.run(documents=documents)
# ── 第五步：持久化向量存储（Chroma）──────────
chroma_client = chromadb.PersistentClient(path="./chroma_db")
chroma_collection = chroma_client.get_or_create_collection("rag_docs")
vector_store = ChromaVectorStore(chroma_collection=chroma_collection)
storage_context = StorageContext.from_defaults(vector_store=vector_store)
index = VectorStoreIndex(enriched_nodes, storage_context=storage_context)
# ── 第六步：高级检索 + 后处理 ────────────────
retriever = VectorIndexRetriever(index=index, similarity_top_k=10)
postprocessors = [
SimilarityPostprocessor(similarity_cutoff=0.7),
KeywordNodePostprocessor(
required_keywords=["架构"],
exclude_keywords=["废弃"],
),
]
# ── 第七步：自定义 Prompt ────────────────────
qa_prompt = PromptTemplate(
"""你是一个专业的技术文档助手。请基于以下参考资料回答用户的问题。
要求：
1. 只基于提供的参考资料回答，不要编造信息
2. 如果参考资料不足以回答问题，请明确说明
3. 引用具体的来源文档
4. 使用清晰的结构化格式
参考资料：
{context_str}
用户问题：{query_str}
回答："""
)
response_synthesizer = get_response_synthesizer(
response_mode="compact",
text_qa_template=qa_prompt,
)
# ── 第八步：组装查询引擎 ────────────────────
query_engine = RetrieverQueryEngine(
retriever=retriever,
response_synthesizer=response_synthesizer,
node_postprocessors=postprocessors,
)
response = query_engine.query("系统的微服务架构是如何设计的？")
print("回答:", response)
for i, node in enumerate(response.source_nodes):
print(f" [{i+1}] {node.metadata.get('file_name', '未知')} "
f"(相关度: {node.score:.3f})")
17.6.6 分块策略深度对比#
分块（Chunking）是 RAG 中最影响效果的环节。错误的分块策略会导致检索到不相关或不完整的内容。
策略 |
原理 |
优势 |
适用场景 |
|---|---|---|---|
固定大小 |
按 token 数切分 |
简单快速，可预测 |
通用场景，快速原型 |
句子分割 |
按句子边界切分 |
保持语义完整性 |
自然语言文档 |
语义分块 |
按语义相似度切分 |
语义边界最准确 |
高质量要求场景 |
递归分割 |
按层级分隔符递归切分 |
保持文档结构 |
结构化文档（Markdown/代码） |
文档特定 |
按文档结构切分（标题/章节） |
保持逻辑完整性 |
技术文档、论文 |
from llama_index.core.node_parser import (
SentenceSplitter, SemanticSplitterNodeParser,
MarkdownNodeParser, CodeSplitter,
)
# 代码文件专用分块器
code_splitter = CodeSplitter(
language="python", chunk_lines=40, chunk_lines_overlap=5, max_chars=1500,
)
# Markdown 文件专用分块器（按标题层级切分）
md_splitter = MarkdownNodeParser()
# 根据文件类型自动选择分块器
def get_splitter_for_file(file_path: str):
if file_path.endswith('.py'):
return CodeSplitter(language="python", chunk_lines=40)
elif file_path.endswith('.md'):
return MarkdownNodeParser()
elif file_path.endswith('.pdf'):
return SentenceSplitter(chunk_size=512, chunk_overlap=50)
else:
return SentenceSplitter(chunk_size=256, chunk_overlap=30)
17.6.7 高级 RAG 技术#
基础 RAG 的检索质量往往不够理想。以下是提升效果的关键技术：
1. 查询改写（HyDE）
from llama_index.core.query_engine import TransformQueryEngine
from llama_index.core.indices.query.query_transform import HyDEQueryTransform
# HyDE：先让 LLM 生成"假设性答案"，用答案去检索
# 效果比直接用问题检索好 20-30%
hyde = HyDEQueryTransform(include_original=True)
hyde_query_engine = TransformQueryEngine(query_engine, hyde)
response = hyde_query_engine.query("如何优化数据库查询性能？")
2. 子问题分解
from llama_index.core.tools import QueryEngineTool, ToolMetadata
from llama_index.core.query_engine import SubQuestionQueryEngine
tools = [
QueryEngineTool(
query_engine=architecture_engine,
metadata=ToolMetadata(
name="architecture_docs",
description="系统架构设计文档"
),
),
QueryEngineTool(
query_engine=api_engine,
metadata=ToolMetadata(name="api_docs", description="API 接口文档"),
),
]
sub_question_engine = SubQuestionQueryEngine.from_defaults(
query_engine_tools=tools,
)
# 复杂问题自动分解为子问题，分别检索后综合回答
response = sub_question_engine.query(
"系统的用户认证是如何设计的？包括架构层面和 API 层面"
)
3. 混合检索（Hybrid Search）
from llama_index.core.retrievers import QueryFusionRetriever
# 融合向量检索（语义匹配）和关键词检索（精确匹配）
hybrid_retriever = QueryFusionRetriever(
retrievers=[vector_retriever, keyword_retriever],
similarity_top_k=5,
num_queries=4,
mode="reciprocal_rerank",
)
4. 重排序（Reranking）
from llama_index.postprocessor.cohere_rerank import CohereRerank
reranker = CohereRerank(api_key="your-key", top_n=5)
query_engine = index.as_query_engine(
similarity_top_k=20, # 先粗检索 20 个
node_postprocessors=[reranker], # 再精排到 5 个
)
17.6.8 RAG 与 Agent 的结合#
RAG 可以作为 Agent 的工具，让 Agent 在需要时主动检索知识：
from llama_index.core.tools import QueryEngineTool
from llama_index.core.agent import ReActAgent
tools = [
QueryEngineTool.from_defaults(
query_engine=code_index.as_query_engine(similarity_top_k=5),
name="search_codebase",
description="搜索项目代码库，了解代码实现细节",
),
QueryEngineTool.from_defaults(
query_engine=docs_index.as_query_engine(similarity_top_k=5),
name="search_docs",
description="搜索项目文档，了解架构设计和业务逻辑",
),
]
agent = ReActAgent.from_tools(
tools=tools, llm=OpenAI(model="gpt-4o"),
verbose=True, max_iterations=10,
)
# Agent 自动决定何时使用哪个知识库
response = agent.chat("用户登录接口的实现逻辑是什么？")
17.6.9 RAG 评估#
from llama_index.core.evaluation import (
FaithfulnessEvaluator, RelevancyEvaluator,
)
faithfulness_eval = FaithfulnessEvaluator(llm=OpenAI(model="gpt-4o"))
relevancy_eval = RelevancyEvaluator(llm=OpenAI(model="gpt-4o"))
query = "系统支持哪些认证方式？"
response = query_engine.query(query)
# 忠实度：答案是否基于检索到的文档（防幻觉）
faith = faithfulness_eval.evaluate_response(query=query, response=response)
print(f"忠实度: {'通过' if faith.passing else '不通过'}")
# 批量评估
eval_questions = [
"系统的技术栈是什么？",
"如何部署到生产环境？",
"API 的认证方式有哪些？",
]
results = []
for q in eval_questions:
resp = query_engine.query(q)
faith = faithfulness_eval.evaluate_response(query=q, response=resp)
results.append(faith.passing)
print(f"整体忠实度: {sum(results)/len(results):.1%}")
17.6.10 RAG 常见问题与优化清单#
问题 |
原因 |
解决方案 |
|---|---|---|
检索不到相关文档 |
分块太大/太小；Embedding 不匹配 |
调整 chunk_size；换 Embedding 模型；用 HyDE |
检索到但答案不对 |
Prompt 模板不好；LLM 忽略上下文 |
优化 Prompt；加"只基于文档回答"指令 |
答案包含幻觉 |
检索结果不够相关 |
添加 Reranker；降低 temperature |
跨文档问题回答差 |
单次检索无法覆盖多文档 |
用 SubQuestionQueryEngine 分解问题 |
速度太慢 |
Embedding 调用多；检索范围大 |
降维；缓存；减少 top_k |
成本太高 |
每次查询都调 LLM |
缓存常见查询；小模型初筛 |
RAG 优化黄金法则
分块是基础：花 80% 的优化时间在分块策略上
检索比生成重要：检索不到正确文档，再好的 LLM 也无法回答
评估驱动优化：先建立评估基准，再做优化，用数据说话
混合检索：向量 + 关键词几乎总是比单一方式好
Reranker 是性价比最高的优化：通常能提升 10-20% 准确率
17.7 记忆系统设计#
Agent 的记忆系统是其保持上下文连贯性和从经验中学习的关键：
from datetime import datetime
import hashlib
class MemoryType(Enum):
SHORT_TERM = "short_term" # 短期记忆：当前对话上下文
LONG_TERM = "long_term" # 长期记忆：持久化的知识和经验
WORKING = "working" # 工作记忆：当前任务的中间状态
EPISODIC = "episodic" # 情景记忆：具体事件的记录
SEMANTIC = "semantic" # 语义记忆：抽象知识和概念
@dataclass
class MemoryEntry:
content: str
memory_type: MemoryType
timestamp: datetime
importance: float # 0.0 - 1.0
access_count: int = 0
last_accessed: Optional[datetime] = None
embedding: Optional[List[float]] = None
metadata: Dict[str, Any] = field(default_factory=dict)
class AgentMemorySystem:
"""Agent 记忆系统"""
def __init__(self, vector_store, embedding_model, max_short_term: int = 20):
self.short_term: List[MemoryEntry] = []
self.working: Dict[str, Any] = {}
self.vector_store = vector_store # 用于长期记忆
self.embedding_model = embedding_model
self.max_short_term = max_short_term
def store(self, content: str, memory_type: MemoryType, importance: float = 0.5, **metadata):
"""存储记忆"""
entry = MemoryEntry(
content=content,
memory_type=memory_type,
timestamp=datetime.now(),
importance=importance,
metadata=metadata
)
if memory_type == MemoryType.SHORT_TERM:
self.short_term.append(entry)
if len(self.short_term) > self.max_short_term:
self._consolidate_short_term()
elif memory_type in (MemoryType.LONG_TERM, MemoryType.EPISODIC, MemoryType.SEMANTIC):
entry.embedding = self.embedding_model.encode(content)
self.vector_store.add(entry)
def recall(self, query: str, memory_type: Optional[MemoryType] = None, top_k: int = 5) -> List[MemoryEntry]:
"""检索记忆"""
results = []
# 搜索短期记忆
if memory_type in (None, MemoryType.SHORT_TERM):
for entry in reversed(self.short_term):
if query.lower() in entry.content.lower():
results.append(entry)
# 搜索长期记忆（向量检索）
if memory_type in (None, MemoryType.LONG_TERM, MemoryType.EPISODIC):
query_embedding = self.embedding_model.encode(query)
long_term_results = self.vector_store.similarity_search(
query_embedding, top_k=top_k
)
results.extend(long_term_results)
# 按重要性和时间排序
results.sort(key=lambda x: (x.importance, x.timestamp), reverse=True)
return results[:top_k]
def _consolidate_short_term(self):
"""将短期记忆整合到长期记忆"""
# 保留最近和最重要的记忆
self.short_term.sort(key=lambda x: x.importance, reverse=True)
to_consolidate = self.short_term[self.max_short_term // 2:]
self.short_term = self.short_term[:self.max_short_term // 2]
for entry in to_consolidate:
if entry.importance > 0.3:
entry.memory_type = MemoryType.LONG_TERM
entry.embedding = self.embedding_model.encode(entry.content)
self.vector_store.add(entry)
def forget(self, threshold: float = 0.1):
"""遗忘不重要的记忆"""
self.short_term = [
m for m in self.short_term if m.importance > threshold
]
17.8 多 Agent 协作架构#
多 Agent 系统有三种主要的协作架构：
层级式（Hierarchical）#
一个 Supervisor Agent 负责任务分配和结果整合，Worker Agent 负责具体执行。
对等式（Peer-to-Peer）#
所有 Agent 地位平等，通过消息传递协作，适合需要讨论和辩论的场景。
竞争式（Competitive）#
多个 Agent 独立完成同一任务，选择最优结果，适合需要多样性的创意任务。
class CollaborationPattern(Enum):
HIERARCHICAL = "hierarchical"
PEER_TO_PEER = "peer_to_peer"
COMPETITIVE = "competitive"
class SupervisorAgent:
"""层级式协作中的 Supervisor"""
def __init__(self, workers: Dict[str, Any], llm_client):
self.workers = workers
self.llm = llm_client
def delegate(self, task: str) -> str:
"""分析任务并委派给合适的 Worker"""
# 1. 分析任务需要哪些能力
analysis = self.llm.chat.completions.create(
model="gpt-4o",
messages=[{
"role": "user",
"content": f"分析任务并选择合适的执行者。\n任务: {task}\n可用执行者: {list(self.workers.keys())}"
}],
response_format={"type": "json_object"}
)
assignment = json.loads(analysis.choices[0].message.content)
# 2. 分发子任务
results = {}
for subtask in assignment.get("subtasks", []):
worker_name = subtask["worker"]
if worker_name in self.workers:
results[worker_name] = self.workers[worker_name].run(subtask["task"])
# 3. 整合结果
return self._synthesize(task, results)
def _synthesize(self, original_task: str, results: dict) -> str:
response = self.llm.chat.completions.create(
model="gpt-4o",
messages=[{
"role": "user",
"content": f"原始任务: {original_task}\n各执行者结果: {json.dumps(results, ensure_ascii=False)}\n请整合为最终答案。"
}]
)
return response.choices[0].message.content
17.9 架构模式选型指南#
模式 |
适用场景 |
优势 |
劣势 |
|---|---|---|---|
ReAct |
通用问答、简单任务 |
实现简单、灵活 |
长任务易迷失方向 |
Plan-and-Execute |
复杂多步骤任务 |
结构清晰、可追踪 |
规划阶段耗时、难以应对意外 |
Reflexion |
需要高准确率的任务 |
能从失败中学习 |
多次尝试增加成本 |
Tool Use |
工具密集型任务 |
工具调用高效 |
依赖工具质量 |
RAG-Agent |
知识密集型任务 |
减少幻觉 |
检索质量影响大 |
Multi-Agent |
大型复杂项目 |
专业化分工 |
协调开销大 |
17.10 本章小结#
本章深入解析了 Agent 的六种核心架构模式。每种模式都有其适用场景和权衡：
ReAct 是最基础的模式，适合快速原型开发
Plan-and-Execute 适合需要系统性规划的复杂任务
Reflexion 通过自我反思提高任务成功率
Tool Use 专注于高效的工具选择和编排
RAG-Agent 将检索增强与 Agent 能力结合
记忆系统是所有模式的基础设施，决定了 Agent 的长期表现
在实际项目中，这些模式往往需要组合使用。下一章我们将介绍主流的 Agent 开发框架，它们封装了这些模式的实现细节，让开发者能够更高效地构建 Agent 系统。