一、从“检索”到“决策”：为什么需要Agent？

2023年以来，RAG（检索增强生成）几乎成了大模型落地企业知识库的标配方案。它的逻辑足够直白：用户提问 → 向量检索召回相关文档 → LLM基于文档生成回答。这套链路解决了LLM“闭卷考试”式的幻觉问题——模型不再依赖自身参数记忆，而是有了外部证据支撑。

但RAG的局限在企业真实场景中很快暴露出来。典型困境是：当用户问“去年Q3的基金持仓变化和今年Q1的对比”，传统RAG要么答非所问，要么直接说无法回答。根本原因在于，RAG只是一个被动的检索-生成管道——它只做一次检索，无法追问、无法规划、无法主动调用外部工具获取实时数据。

这就催生了从RAG到Agent的升级：在RAG基础上为LLM增加记忆模块（记住历史对话）、工具调用模块（调用API获取实时数据）、任务规划模块（将复杂问题拆解为子任务）。Agent不再是“回答问题”，而是“解决问题”。

接下来，我们用LangChain + FastAPI + FAISS，一步步构建一个具备RAG能力的轻量级Agent，涵盖从文档向量化到多轮对话的完整链路。

二、系统架构概览

整体架构分为三层：

┌─────────────────────────────────────────────────────┐
│                    API 层 (FastAPI)                  │
│              /upload  /ask  /history               │
├─────────────────────────────────────────────────────┤
│                   Agent 核心层                       │
│  ┌──────────┐  ┌──────────┐  ┌──────────────┐     │
│  │  Planner │→ │ Retriever│→ │  Generator   │     │
│  │ (意图判断)│  │ (多路检索)│  │ (生成回答)   │     │
│  └──────────┘  └──────────┘  └──────────────┘     │
│                      ↓                              │
│  ┌─────────────────────────────────────────────┐   │
│  │          记忆模块 (ConversationBufferMemory) │   │
│  └─────────────────────────────────────────────┘   │
├─────────────────────────────────────────────────────┤
│                   存储层                            │
│     FAISS (向量索引)  +  SQLite (元数据)           │
└─────────────────────────────────────────────────────┘

Agent的核心能力在于：当用户提问时，Planner先判断问题类型——如果是简单事实查询，走常规RAG检索；如果是复杂任务（如“对比A和B的差异”），则拆解为多次检索并整合结果。整个过程由LangChain的ConversationalRetrievalChain编排。

三、核心技术实现

3.1 文档加载与切分

文档处理是RAG质量的第一道关口。切分策略直接影响检索精度：chunk_size过大会引入噪声，过小则会切断语义。对中文技术文档，推荐chunk_size=800~1200，chunk_overlap=100~150。

# document_processor.py
import os
from typing import List
from langchain_community.document_loaders import PyPDFLoader, UnstructuredMarkdownLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.documents import Document

def load_document(file_path: str) -> List[Document]:
    """支持 PDF 和 Markdown 的多格式加载"""
    ext = os.path.splitext(file_path)[1].lower()
    if ext == ".pdf":
        loader = PyPDFLoader(file_path)  # 自动记录页码到 metadata
    elif ext == ".md":
        loader = UnstructuredMarkdownLoader(file_path)
    else:
        raise ValueError(f"不支持的文件格式: {ext}")
    return loader.load()

def split_documents(docs: List[Document]) -> List[Document]:
    """递归切分，按语义边界分段"""
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,
        chunk_overlap=100,
        separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""],
    )
    chunks = splitter.split_documents(docs)
    # 过滤空块，避免噪声污染向量库
    return [c for c in chunks if c.page_content.strip()]

3.2 向量化存储

采用FAISS作为本地向量索引，适合中小规模知识库（万级文本块以内）。Embedding模型建议与LLM的语言偏好保持一致——中文场景选用中文优化模型。

# vector_store.py
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_core.documents import Document
from typing import List, Optional
import os

def get_embeddings():
    return DashScopeEmbeddings(
        model="text-embedding-v3",
        dashscope_api_key=os.getenv("DASHSCOPE_API_KEY"),
    )

class VectorStoreManager:
    def __init__(self, persist_path: str = "./faiss_index"):
        self.persist_path = persist_path
        self.embeddings = get_embeddings()
        self.store: Optional[FAISS] = None
    
    def build_index(self, documents: List[Document]):
        """构建向量索引并持久化"""
        self.store = FAISS.from_documents(documents, self.embeddings)
        self.store.save_local(self.persist_path)
    
    def load_index(self):
        """从磁盘加载已有索引"""
        if os.path.exists(self.persist_path):
            self.store = FAISS.load_local(
                self.persist_path, 
                self.embeddings,
                allow_dangerous_deserialization=True
            )
            return True
        return False
    
    def get_retriever(self, k: int = 4):
        """获取检索器，k为召回文档数"""
        if not self.store:
            raise ValueError("向量索引未初始化，请先构建或加载")
        return self.store.as_retriever(
            search_kwargs={"k": k}
        )
    
    def add_documents(self, documents: List[Document]):
        """增量追加新文档"""
        if not self.store:
            self.build_index(documents)
        else:
            self.store.add_documents(documents)
            self.store.save_local(self.persist_path)

召回数k的选择：过小会漏掉关键信息，过大会引入噪声稀释LLM注意力。推荐k=3~5，并配合Rerank做二次精排。

3.3 记忆与检索链

多轮对话是Agent区别于基础RAG的关键能力。LangChain的ConversationBufferWindowMemory可以保留最近N轮对话，保证上下文连贯性。但需注意：窗口过大可能导致Prompt超长，推荐k=3~8。

# qa_chain.py
from langchain_openai import ChatOpenAI
from langchain.memory import ConversationBufferWindowMemory
from langchain.chains import ConversationalRetrievalChain
from langchain.prompts import PromptTemplate
from vector_store import VectorStoreManager

# Prompt 设计：关键约束是禁止编造
_QA_PROMPT = PromptTemplate.from_template(
    """你是一个专业的知识助手。请基于以下参考资料回答用户问题。

【参考资料】
{context}

【对话历史】
{chat_history}

【用户问题】
{question}

【回答要求】
1. 仅基于参考资料回答，不要编造不存在的信息
2. 如果参考资料中没有相关信息，请明确告知"未查询到相关信息"
3. 回答要简洁、有条理
4. 如引用具体资料，请注明来源

你的回答："""
)

def create_qa_chain(retriever, llm_model: str = "qwen-plus"):
    """创建带记忆的对话检索链"""
    llm = ChatOpenAI(
        model=llm_model,
        api_key=os.getenv("LLM_API_KEY"),
        base_url=os.getenv("LLM_BASE_URL"),
        temperature=0.1,
    )
    
    memory = ConversationBufferWindowMemory(
        k=5,  # 保留最近5轮对话
        memory_key="chat_history",
        return_messages=True,
        output_key="answer",  # 指定写入记忆的输出字段
    )
    
    chain = ConversationalRetrievalChain.from_llm(
        llm=llm,
        retriever=retriever,
        memory=memory,
        combine_docs_chain_kwargs={"prompt": _QA_PROMPT},
        return_source_documents=True,
        verbose=False,
    )
    return chain

关键设计点：output_key="answer"配合return_source_documents=True时必须设置，否则记忆模块无法识别应写入哪个输出字段。

3.4 API服务层

用FastAPI封装Agent能力，提供文档上传、问答、历史查看三个核心接口。

# app.py
from fastapi import FastAPI, UploadFile, File, HTTPException
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from typing import List, Optional
import tempfile
import os

from document_processor import load_document, split_documents
from vector_store import VectorStoreManager
from qa_chain import create_qa_chain

app = FastAPI(title="RAG Agent Service")

# 全局状态
vector_manager = VectorStoreManager()
chain = None

class AskRequest(BaseModel):
    question: str
    session_id: Optional[str] = None

class AskResponse(BaseModel):
    answer: str
    sources: List[str]
    success: bool

@app.on_event("startup")
async def startup():
    """启动时加载已有索引"""
    if vector_manager.load_index():
        global chain
        retriever = vector_manager.get_retriever(k=4)
        chain = create_qa_chain(retriever)
        print("✅ 已加载已有向量索引")
    else:
        print("⚠️ 未找到已有索引，请先上传文档")

@app.post("/upload")
async def upload_document(file: UploadFile = File(...)):
    """上传并索引文档"""
    # 保存临时文件
    suffix = os.path.splitext(file.filename)[1]
    with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
        content = await file.read()
        tmp.write(content)
        tmp_path = tmp.name
    
    try:
        # 加载并切分文档
        docs = load_document(tmp_path)
        chunks = split_documents(docs)
        
        # 记录来源信息
        for chunk in chunks:
            chunk.metadata["source"] = file.filename
        
        # 向量化存储
        vector_manager.add_documents(chunks)
        
        # 更新链
        global chain
        retriever = vector_manager.get_retriever(k=4)
        chain = create_qa_chain(retriever)
        
        return JSONResponse({
            "success": True,
            "message": f"成功索引 {len(chunks)} 个文档块",
            "filename": file.filename,
        })
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))
    finally:
        os.unlink(tmp_path)

@app.post("/ask", response_model=AskResponse)
async def ask(request: AskRequest):
    """智能问答接口"""
    if not chain:
        raise HTTPException(status_code=503, detail="系统尚未初始化，请先上传文档")
    
    try:
        result = chain({"question": request.question})
        answer = result["answer"]
        sources = list(set([
            doc.metadata.get("source", "未知来源") 
            for doc in result.get("source_documents", [])
        ]))
        return AskResponse(answer=answer, sources=sources, success=True)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health():
    return {"status": "ok", "index_loaded": vector_manager.store is not None}

3.5 可观测性：追踪每一步

RAG系统的质量问题往往难以定位——是检索阶段召回不相关文档？还是LLM忽略了检索结果？引入Langfuse可以自动捕获链中每一步的输入输出，可视化展示完整的调用链路。

# config.py
import os
from langfuse.langchain import CallbackHandler

def get_langfuse_handler():
    """Langfuse 3.x 通过环境变量自动配置"""
    if not os.getenv("LANGFUSE_ENABLED", "false").lower() == "true":
        return None
    return CallbackHandler(update_trace=True)

# 在调用链时注入
# result = chain({"question": question}, callbacks=[handler] if handler else None)

四、从RAG到Agent的进阶方向

上述实现是RAG+基础对话记忆，距离真正的“Agent”还有几步之遥。以下是可迭代演进的方向：

1. 多路检索与混合排序：目前是单路向量检索。生产环境建议增加关键词检索（BM25），通过RRF（Reciprocal Rank Fusion）融合多路结果，再经Rerank模型精排。

2. 查询改写：用户提问往往口语化、不完整。增加一个前置改写步骤，将问题转化为更适合向量检索的形式，能显著提升召回质量。

3. 工具调用：当Agent判断需要实时数据（如天气、股价）时，调用外部API获取信息后再回答。这需要将RAG检索封装为Agent的一个“工具”，由Planner决策何时调用。

4. 规划与多跳推理：对于“对比A和B的差异”这类问题，Agent应自动拆解为“检索A信息→检索B信息→对比生成”，而非一次性召回。

五、部署与成本控制

生产级部署建议：

• 容器化：使用Docker打包，Kubernetes编排，支持弹性伸缩
• 无状态设计：会话状态存Redis，向量索引存云存储，便于水平扩展
• 成本优化：Embedding调用可批量处理；对高频问题增加缓存层

总结

从RAG到Agent的演进，本质上是将LLM从“被动的检索-生成工具”升级为“具备记忆、规划和执行能力的决策体”。本文实现的是一个最小可行版本——完整的RAG链路 + 多轮对话记忆 + 标准化的API封装。在此基础上，你可以逐步叠加查询改写、混合检索、工具调用等能力，向真正的企业级Agent演进。