L3 RAG 与智能体

目标

• 不用任何框架，从零写一个最小的 RAG，理解它由哪几步组成；
• 用 LangGraph 把同样的流程编排成图；
• 了解主流智能体框架，以及 RAG 和智能体工作流中的系统优化问题。

背景

RAG（检索增强生成）：大模型不知道私有的、最新的知识。RAG 先从知识库里检索相关片段，拼进提示词，再让模型回答。流程是：

文档 → 切块 → 向量化 → 存入向量库
问题 → 向量化 → 检索最相关的 K 块 → 拼成提示词 → 大模型生成回答

智能体（Agent）：让大模型自己决定下一步做什么，例如调用搜索工具、执行代码、再根据结果继续推理。多个智能体可以分工协作。

从系统角度看，RAG 和智能体都把一次回答变成了多次模型调用和检索组成的工作流，这正是方向一和方向二研究的对象。

环境

• 完成 L1，有一个正在运行的模型服务（vLLM 或 Ollama）
• Python 包：uv pip install openai sentence-transformers numpy langgraph

第 1 步：从零写一个 RAG

准备一个文本文件 knowledge.txt，放几段你熟悉领域的资料（比如从本手册复制几页）。

# mini_rag.py
import numpy as np
from openai import OpenAI
from sentence_transformers import SentenceTransformer

# 1. 切块：按段落切，太长的段落再按固定长度切
text = open("knowledge.txt", encoding="utf-8").read()
chunks = []
for para in text.split("\n\n"):
    para = para.strip()
    for i in range(0, len(para), 300):
        if para[i:i + 300]:
            chunks.append(para[i:i + 300])

# 2. 向量化（BGE 小模型，中文效果好，CPU 也能跑）
embedder = SentenceTransformer("BAAI/bge-small-zh-v1.5")
chunk_vecs = embedder.encode(chunks, normalize_embeddings=True)

# 3. 检索：向量已归一化，内积就是余弦相似度
def retrieve(question, k=3):
    q = embedder.encode([question], normalize_embeddings=True)[0]
    scores = chunk_vecs @ q
    return [chunks[i] for i in np.argsort(-scores)[:k]]

# 4. 生成
client = OpenAI(base_url="http://localhost:8000/v1", api_key="EMPTY")

def answer(question):
    context = "\n\n".join(retrieve(question))
    prompt = (
        "根据下面的资料回答问题。资料里没有相关信息时，直接说不知道。\n\n"
        f"资料：\n{context}\n\n问题：{question}"
    )
    resp = client.chat.completions.create(
        model="qwen3-0.6b",
        messages=[{"role": "user", "content": prompt}],
        extra_body={"chat_template_kwargs": {"enable_thinking": False}},
    )
    return resp.choices[0].message.content

print(answer("课题组的总体研究目标是什么？"))

做完后思考：

• 把检索出来的片段打印出来，检索得准吗？不准的时候，问题出在切块、向量模型还是问题表述？
• 用 L1 的方法测一下每一步各花了多少时间：向量化、检索、生成。哪一步最慢？
• 知识库换成上万段文本时，把第 3 步换成 L2 的 PGVector。

第 2 步：用 LangGraph 编排

LangGraph 把工作流表示成图：节点是步骤，边是流向，状态在节点之间传递。复杂的智能体（带循环、分支、人工确认）用图来表达更清晰。

把上面的 RAG 改写成两个节点：

# rag_graph.py
from typing import TypedDict

from langgraph.graph import END, START, StateGraph

from mini_rag import client, retrieve


class State(TypedDict):
    question: str
    context: str
    answer: str


def retrieve_node(state: State):
    return {"context": "\n\n".join(retrieve(state["question"]))}


def generate_node(state: State):
    prompt = f"根据资料回答问题，资料里没有就说不知道。\n\n资料：\n{state['context']}\n\n问题：{state['question']}"
    resp = client.chat.completions.create(
        model="qwen3-0.6b",
        messages=[{"role": "user", "content": prompt}],
        extra_body={"chat_template_kwargs": {"enable_thinking": False}},
    )
    return {"answer": resp.choices[0].message.content}


graph = StateGraph(State)
graph.add_node("retrieve", retrieve_node)
graph.add_node("generate", generate_node)
graph.add_edge(START, "retrieve")
graph.add_edge("retrieve", "generate")
graph.add_edge("generate", END)
app = graph.compile()

print(app.invoke({"question": "课题组的总体研究目标是什么？"})["answer"])

（from mini_rag import ... 会执行一遍 mini_rag.py 末尾的示例问答，可以把那行 print 删掉。）

练习：加一个“判断检索结果是否相关”的节点，不相关时改写问题重新检索（用条件边 add_conditional_edges 实现循环）。

智能体框架速览

框架	特点
LangGraph	用图编排有状态的工作流，控制粒度细，适合研究和定制
AgentScope	阿里巴巴通义实验室开源，面向多智能体应用，中文文档丰富
CrewAI	按“角色 + 任务”组织多个智能体，上手快
Ray	不是智能体框架，但常用来把多智能体系统分布到多台机器上，见 L4

智能体调用外部工具和数据源，现在普遍使用 MCP（Model Context Protocol） 这一标准协议。

开箱即用的 RAG 应用（适合体验，不适合做研究改动）：RAGFlow、Langchain-Chatchat。

RAG 与智能体的系统优化

做研究时关注的不是“能不能跑通”，而是“怎么跑得更快、更省”。代表性工作：

工作	解决的问题	资源
Ayo（ASPLOS 2025）	把 RAG 等应用的整个工作流作为整体做端到端优化，而不是孤立地优化每次模型调用	代码、论文
Parrot（OSDI 2024）	用“语义变量”暴露应用层信息，让推理服务做跨请求优化
CacheBlend（EuroSys 2025）	RAG 中检索到的文档片段的 KV Cache 能否复用，以减少重复的预填充计算	论文
LMCache	KV Cache 的存储、共享与传输层，可接入 vLLM	代码

这部分工作需要先理解 RAG 流程和 KV Cache 原理（见深度学习与大模型）。