Agentic RAG 与 Context Engineering

从"流水线"到"工具调用"的范式转移

第 1–10 章里我们学的是RAG 流水线：分块 → Embedding → 检索 → Rerank → 生成。这是预定义的、固定的步骤。

Agentic RAG 和 Context Engineering 共享一个更激进的想法：让 LLM 自己决定何时检索、检索什么、是否还要再检索一次——把"检索"从流水线节点降级为一种 tool_call。

方向 ②：Agentic RAG

核心模式：自反思检索循环

Agentic RAG 最常见的实现是 Self-RAG / CRAG（Corrective RAG） 风格的循环：检索 → LLM 评估检索质量 → 不够好则重写查询再检索 → 直到满意才生成。

┌──────────────────────────────────────────────────┐
│              Agentic RAG 控制流                    │
│                                                    │
│   user_question                                    │
│        │                                           │
│        ▼                                           │
│   [Plan Agent] 拆分子问题                          │
│        │                                           │
│        ├──→ [Retrieve Agent] 向量检索              │
│        │           │                               │
│        │           ▼                               │
│        ├──→ [Grade Agent] 这些 chunk 真的有用吗？  │
│        │           │                               │
│        │       ┌───┴───┐                           │
│        │      够用   不够用                         │
│        │       │      │                            │
│        │       │      └──→ [Rewrite] 改写查询      │
│        │       │           ↑     │                 │
│        │       │           └─────┘ 循环            │
│        │       ▼                                   │
│        └──→ [Synthesize Agent] 综合多源信息       │
│                  │                                 │
│                  ▼                                 │
│              最终答案                              │
└──────────────────────────────────────────────────┘

用 LangGraph 实现最小骨架

from langgraph.graph import StateGraph, END
from typing import TypedDict, List
from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-sonnet-4-6")

class RAGState(TypedDict):
    question: str
    documents: List[str]
    grade: str             # "useful" | "not_useful"
    rewrites: int          # 重写次数防止死循环

def retrieve(state: RAGState) -> RAGState:
    docs = vector_store.search(state["question"], top_k=5)
    return {"documents": docs}

def grade_documents(state: RAGState) -> RAGState:
    """LLM 自评估检索质量"""
    prompt = f"""问题：{state['question']}
检索到的文档：{state['documents']}
这些文档能回答问题吗？只回答 useful 或 not_useful。"""
    grade = llm.invoke(prompt).content.strip().lower()
    return {"grade": grade}

def rewrite_query(state: RAGState) -> RAGState:
    """检索失败时重写查询"""
    prompt = f"""原问题：{state['question']}
检索失败。请改写问题，使其更容易匹配文档（例如换关键词、加领域术语）。"""
    new_q = llm.invoke(prompt).content
    return {"question": new_q, "rewrites": state["rewrites"] + 1}

def generate(state: RAGState) -> RAGState:
    answer = llm.invoke(f"基于以下文档回答：\n{state['documents']}\n问题：{state['question']}")
    return {"answer": answer.content}

def decide_next(state: RAGState) -> str:
    if state["grade"] == "useful" or state["rewrites"] >= 3:
        return "generate"
    return "rewrite"

# 编排图
g = StateGraph(RAGState)
g.add_node("retrieve", retrieve)
g.add_node("grade", grade_documents)
g.add_node("rewrite", rewrite_query)
g.add_node("generate", generate)
g.set_entry_point("retrieve")
g.add_edge("retrieve", "grade")
g.add_conditional_edges("grade", decide_next, {"generate": "generate", "rewrite": "rewrite"})
g.add_edge("rewrite", "retrieve")         # 闭环
g.add_edge("generate", END)

agentic_rag = g.compile()
result = agentic_rag.invoke({"question": "...", "rewrites": 0})

Agentic RAG 的成本警告

方向 ③：Context Engineering

它和 RAG 的根本区别

Context Engineering 不是一个具体技术，是 2024 下半年 Anthropic / Cursor / Cognition AI 提出的思维方式。一句话概括：

Coding Agent 是 Context Engineering 的标杆

看一下 Claude Code、Cursor、Aider 这类 Coding Agent 怎么处理"代码库问答"——它们几乎不用向量库。

Cursor 的 @codebase 算混合方案——保留向量索引但同时给了 Agent 直接的文件读写工具。详见 Cursor 教程第 3 章 @codebase 向量检索原理。

Context Engineering 的最小骨架

from anthropic import Anthropic
import subprocess, json

client = Anthropic()

tools = [
    {
        "name": "grep",
        "description": "在文件系统中搜索文本（精确匹配）",
        "input_schema": {"type": "object",
            "properties": {"pattern": {"type": "string"},
                          "path": {"type": "string"}},
            "required": ["pattern"]}
    },
    {
        "name": "read_file",
        "description": "读取文件内容，可指定行号区间",
        "input_schema": {"type": "object",
            "properties": {"path": {"type": "string"},
                          "start": {"type": "integer"},
                          "end": {"type": "integer"}},
            "required": ["path"]}
    },
]

def execute_tool(name: str, args: dict) -> str:
    if name == "grep":
        out = subprocess.run(["grep", "-rn", args["pattern"], args.get("path", ".")],
                             capture_output=True, text=True)
        return out.stdout[:3000]
    if name == "read_file":
        with open(args["path"]) as f:
            lines = f.readlines()
        s, e = args.get("start", 1) - 1, args.get("end", len(lines))
        return "".join(lines[s:e])

def context_engineering_query(question: str) -> str:
    """不预建索引，让模型自己探索代码库"""
    messages = [{"role": "user", "content": question}]
    while True:
        resp = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=2048,
            tools=tools,
            messages=messages,
        )
        if resp.stop_reason == "end_turn":
            return resp.content[0].text
        # 模型决定调用工具
        tool_use = [b for b in resp.content if b.type == "tool_use"][0]
        result = execute_tool(tool_use.name, tool_use.input)
        messages.append({"role": "assistant", "content": resp.content})
        messages.append({"role": "user", "content": [
            {"type": "tool_result",
             "tool_use_id": tool_use.id,
             "content": result}
        ]})

方向 ④：GraphRAG 与多模态检索

GraphRAG：给检索加上"关系"

纯向量检索回答不了"X 引用的论文里，哪些被 Y 又引用过"这种多跳关系查询。GraphRAG（微软 2024 年开源）把文档先抽成知识图谱，再做检索：

┌─────────────────────────────────────────────────┐
│  GraphRAG 索引阶段（一次性）                      │
│                                                  │
│  原始文档 ─→ LLM 抽取 ─→ (实体, 关系, 实体)        │
│                          (论文A) -[引用]-> (论文B) │
│                          (作者X) -[发表]-> (论文A) │
│                                                  │
│  ─→ 构建图（Neo4j）                               │
│  ─→ 社区检测（Leiden 算法）找出主题集群           │
│  ─→ LLM 给每个社区生成摘要                        │
└─────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────┐
│  GraphRAG 查询阶段                                │
│                                                  │
│  用户问题                                        │
│       │                                          │
│       ├─→ Local Search：实体邻居遍历             │
│       │     适合"X 的具体属性"类问题              │
│       │                                          │
│       └─→ Global Search：基于社区摘要             │
│             适合"整个语料的主题趋势"类问题         │
└─────────────────────────────────────────────────┘

第 10 章 混合存储：向量 + 图数据库 给了 Neo4j 的最小代码。完整 GraphRAG 实现成本不低（实体抽取要消耗大量 LLM 调用），只在确实需要关系查询时上。

ColPali：视觉文档检索

朴素 RAG 处理 PDF 的最大痛点：分块切片把表格、公式、架构图的图注全切碎了。ColPali（2024 年 Faysse 等提出）走了另一条路——不做 OCR，不做分块，直接用视觉模型把 PDF 每一页编码成多向量，检索时用 late interaction 计算页面与查询的相似度。

对于 PDF 密集的领域（医疗、法律、金融研报），这是当前最优解。完整教程见 ColPali 视觉文档检索教程，10 章覆盖架构原理、byaldi 上手、多向量索引、领域微调、ViDoRe 评估。

四个方向的组合：一个真实生产蓝图

用户问题: "对比2023和2024年我们和竞品在自动驾驶领域的专利布局"
    │
    ▼
[Plan Agent] 拆分子问题
    │
    ├─→ 子问题1: 我们公司的自动驾驶专利？
    │     └─→ [GraphRAG Local Search]（专利-公司图）
    │
    ├─→ 子问题2: 竞品 A/B/C 的自动驾驶专利？
    │     └─→ [GraphRAG Local Search]
    │
    ├─→ 子问题3: 专利全文里的技术细节？
    │     └─→ [ColPali 视觉检索]（保留专利图、流程图）
    │
    ├─→ 子问题4: 行业研报背景？
    │     └─→ [Prompt Caching CAG]（5 篇核心研报常驻缓存）
    │
    ▼
[Synthesize Agent] 综合 → 输出对比报告

这就是 第 11 章 反复强调的"真实生产系统是混合体"。

三章总览：你现在拥有的工具箱