AI 大模型与 Agent 开发实战

13.7 完整项目实战：智能文档问答 Agent

生活类比：前面学的所有章节，就像分别学了切菜、炒菜、调味、摆盘。现在该做一道完整的菜了——从买菜到端上桌，把所有技能串起来。这个项目就是你的"毕业作品"。

本节我们从零构建一个完整的智能文档问答 Agent。它包含前端界面、后端 API、RAG Pipeline 和流式输出——麻雀虽小，五脏俱全。这也是全书的最后一个实战项目。

13.7.1 项目架构

┌──────────┐     ┌──────────────┐     ┌──────────────┐
│  前端 UI  │ ←→  │  FastAPI     │ ←→  │  OpenAI API  │
│  (HTML)  │     │  (Backend)   │     │  (LLM)       │
└──────────┘     └──────┬───────┘     └──────────────┘
                        │
                  ┌─────┴──────┐
                  │  Chroma DB │
                  │  (向量存储) │
                  └────────────┘

技术栈选择：

后端：FastAPI——异步、高性能、自带 API 文档
RAG：LangChain——文档处理和链式调用
向量库：Chroma——轻量、嵌入式、无需额外服务
LLM：OpenAI——GPT-4o-mini 性价比高
前端：原生 HTML/JS——不依赖框架，专注核心逻辑

13.7.2 项目结构

doc-qa-agent/
├── main.py          # FastAPI 主程序
├── rag_pipeline.py  # RAG Pipeline
├── static/
│   └── index.html   # 前端界面
├── requirements.txt # 依赖
└── data/            # 文档存放目录

13.7.3 RAG Pipeline 实现（rag_pipeline.py）

python

import os
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import TextLoader, PyPDFLoader
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain.schema import Document
from typing import List, AsyncIterator
import glob

class DocQAPipeline:
    """文档问答 Pipeline——负责索引构建和查询"""

    def __init__(self, persist_dir: str = "./chroma_db"):
        # 嵌入模型：把文本转成向量
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
        # 聊天模型：生成回答
        self.llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)  # 低温度减少幻觉
        self.persist_dir = persist_dir
        self.vectorstore = None

    def ingest_documents(self, data_dir: str = "./data"):
        """从 data 目录加载所有文档并构建索引"""
        all_docs = []

        # 递归遍历 data 目录下所有文件
        for filepath in glob.glob(f"{data_dir}/**/*", recursive=True):
            if not os.path.isfile(filepath):       # 跳过目录
                continue

            try:
                # 根据文件类型选择加载器
                if filepath.endswith('.pdf'):
                    loader = PyPDFLoader(filepath)
                elif filepath.endswith(('.txt', '.md', '.py')):
                    loader = TextLoader(filepath, encoding='utf-8')
                else:
                    continue                       # 跳过不支持的格式

                docs = loader.load()
                for doc in docs:
                    doc.metadata["source"] = filepath  # 记录来源文件路径
                all_docs.extend(docs)
                print(f"  ✓ {filepath} ({len(docs)} chunks)")
            except Exception as e:
                print(f"  ✗ {filepath}: {e}")       # 加载失败但不中断

        print(f"\n总计: {len(all_docs)} 个文档片段")

        # 文档分块
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,                        # 每块 1000 字符
            chunk_overlap=200,                       # 重叠 200 字符
            separators=["\n\n", "\n", "。", "！", "？", "；", " ", ""]
        )
        chunks = splitter.split_documents(all_docs)
        print(f"分块: {len(chunks)} 个 chunk")

        # 构建向量索引
        self.vectorstore = Chroma.from_documents(
            documents=chunks,
            embedding=self.embeddings,
            persist_directory=self.persist_dir
        )
        print(f"索引已保存到 {self.persist_dir}")

    def query(self, question: str, k: int = 5) -> dict:
        """查询文档——检索 + 生成"""
        if not self.vectorstore:
            # 如果还没加载索引，从持久化目录加载
            self.vectorstore = Chroma(
                persist_directory=self.persist_dir,
                embedding_function=self.embeddings
            )

        # 创建检索器，检索 top-k 相关文档
        retriever = self.vectorstore.as_retriever(search_kwargs={"k": k})
        docs = retriever.invoke(question)            # 执行检索

        # 把检索到的文档块拼接成上下文
        context = "\n\n---\n\n".join([
            f"[来源: {d.metadata.get('source', 'unknown')}]\n{d.page_content}"
            for d in docs
        ])

        # 构建 Prompt 模板
        prompt = ChatPromptTemplate.from_template("""基于以下文档内容回答问题。如果文档中没有相关信息，请诚实说明。

文档内容：
{context}

问题：{question}

请提供准确、有据可查的回答，并标注信息来源。""")

        # 构建问答链并执行
        chain = prompt | self.llm | StrOutputParser()
        answer = chain.invoke({"context": context, "question": question})

        return {
            "question": question,
            "answer": answer,
            "sources": [
                {
                    "source": d.metadata.get("source", ""),
                    "content_preview": d.page_content[:200]
                }
                for d in docs
            ]
        }

    async def query_stream(self, question: str, k: int = 5) -> AsyncIterator[str]:
        """流式查询——逐字返回结果"""
        if not self.vectorstore:
            self.vectorstore = Chroma(
                persist_directory=self.persist_dir,
                embedding_function=self.embeddings
            )

        retriever = self.vectorstore.as_retriever(search_kwargs={"k": k})
        docs = retriever.invoke(question)

        context = "\n\n---\n\n".join([
            f"[来源: {d.metadata.get('source', 'unknown')}]\n{d.page_content}"
            for d in docs
        ])

        prompt = ChatPromptTemplate.from_template("""基于以下文档内容回答问题：

文档内容：
{context}

问题：{question}

回答：""")

        chain = prompt | self.llm                     # 不加 StrOutputParser，保留原始 chunk

        # 异步流式输出
        async for chunk in chain.astream({"context": context, "question": question}):
            if chunk.content:
                yield chunk.content

13.7.4 FastAPI 主程序（main.py）

python

from fastapi import FastAPI, HTTPException, UploadFile, File
from fastapi.responses import StreamingResponse, HTMLResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel
import json
import os
import shutil
import uvicorn

from rag_pipeline import DocQAPipeline

app = FastAPI(title="智能文档问答 Agent")
pipeline = DocQAPipeline()                          # 全局 Pipeline 实例


# 定义请求/响应模型
class QueryRequest(BaseModel):
    question: str
    k: int = 5                                      # 默认检索 5 个文档块


class IngestResponse(BaseModel):
    status: str
    message: str


@app.on_event("startup")
async def startup():
    """服务启动时创建必要目录"""
    os.makedirs("./data", exist_ok=True)             # 文档存放目录
    os.makedirs("./chroma_db", exist_ok=True)        # 向量库目录


@app.get("/", response_class=HTMLResponse)
async def root():
    """返回前端页面"""
    with open("static/index.html", "r") as f:
        return f.read()


@app.post("/ingest")
async def ingest_documents():
    """重建索引——扫描 data 目录所有文档"""
    try:
        pipeline.ingest_documents("./data")
        return {"status": "success", "message": "文档索引完成"}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.post("/upload")
async def upload_document(file: UploadFile = File(...)):
    """上传文档"""
    file_path = f"./data/{file.filename}"
    with open(file_path, "wb") as f:
        shutil.copyfileobj(file.file, f)              # 写入文件
    return {"status": "success", "filename": file.filename}


@app.post("/query")
async def query(request: QueryRequest):
    """文档问答——一次性返回"""
    result = pipeline.query(request.question, request.k)
    return result


@app.post("/query/stream")
async def query_stream(request: QueryRequest):
    """流式文档问答——SSE 推送"""
    async def event_generator():
        """SSE 事件生成器"""
        async for chunk in pipeline.query_stream(request.question, request.k):
            yield f"data: {json.dumps({'content': chunk}, ensure_ascii=False)}\n\n"
        yield "data: [DONE]\n\n"                     # 结束标记

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream"
    )


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

13.7.5 前端界面（static/index.html）

html

<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>智能文档问答 Agent</title>
    <style>
        * { margin: 0; padding: 0; box-sizing: border-box; }
        body { font-family: -apple-system, sans-serif; background: #f5f5f5; }
        .container { max-width: 800px; margin: 0 auto; padding: 20px; }
        .header { background: #2563eb; color: white; padding: 20px;
                  border-radius: 12px; margin-bottom: 20px; }
        .chat { background: white; border-radius: 12px; padding: 20px;
                min-height: 400px; max-height: 500px; overflow-y: auto; }
        .message { margin-bottom: 16px; padding: 12px;
                   border-radius: 8px; max-width: 85%; }
        .user { background: #2563eb; color: white; margin-left: auto; }
        .assistant { background: #f0f0f0; }
        .input-area { display: flex; gap: 10px; margin-top: 20px; }
        .input-area input { flex: 1; padding: 12px; border: 1px solid #ddd;
                           border-radius: 8px; font-size: 16px; }
        .input-area button { padding: 12px 24px; background: #2563eb;
                           color: white; border: none; border-radius: 8px;
                           cursor: pointer; }
        .upload-area { margin: 20px  0; }
        .source { font-size: 12px; color: #666; margin-top: 8px;
                 padding: 8px; background: #f5f5f5; border-radius: 4px; }
    </style>
</head>
<body>
    <div class="container">
        <div class="header">
            <h1>📚 智能文档问答 Agent</h1>
            <p>上传文档，提问，获取精准回答</p>
        </div>

        <div class="upload-area">
            <input type="file" id="fileInput" multiple
                   accept=".txt,.pdf,.md">
            <button onclick="uploadFiles()">上传文档</button>
            <button onclick="ingestDocs()">重建索引</button>
        </div>

        <div class="chat" id="chat"></div>

        <div class="input-area">
            <input type="text" id="questionInput"
                   placeholder="输入你的问题..."
                   onkeypress="if(event.key==='Enter')askQuestion()">
            <button onclick="askQuestion()">发送</button>
        </div>
    </div>

    <script>
        // 发送问题并接收流式回复
        async function askQuestion() {
            const input = document.getElementById('questionInput');
            const question = input.value.trim();
            if (!question) return;

            addMessage('user', question);             // 显示用户消息
            input.value = '';

            const msgDiv = addMessage('assistant', ''); // 创建空的 assistant 气泡

            // 发送请求到流式接口
            const response = await fetch('/query/stream', {
                method: 'POST',
                headers: {'Content-Type': 'application/json'},
                body: JSON.stringify({question})
            });

            // 读取流式响应
            const reader = response.body.getReader();
            const decoder = new TextDecoder();

            while (true) {
                const {done, value} = await reader.read();
                if (done) break;

                // 解析 SSE 数据
                const lines = decoder.decode(value).split('\n');
                for (const line of lines) {
                    if (line.startsWith('data: ') && line.slice(6) !== '[DONE]') {
                        const {content} = JSON.parse(line.slice(6));
                        msgDiv.innerHTML += content;   // 逐字追加到气泡
                        document.getElementById('chat').scrollTop =
                            document.getElementById('chat').scrollHeight;
                    }
                }
            }
        }

        // 添加消息气泡
        function addMessage(role, content) {
            const chat = document.getElementById('chat');
            const div = document.createElement('div');
            div.className = `message ${role}`;
            div.innerHTML = content;
            chat.appendChild(div);
            chat.scrollTop = chat.scrollHeight;
            return div;
        }

        // 上传文档
        async function uploadFiles() {
            const files = document.getElementById('fileInput').files;
            for (const file of files) {
                const formData = new FormData();
                formData.append('file', file);
                await fetch('/upload', {method: 'POST', body: formData});
            }
            alert('上传完成！请点击"重建索引"');
        }

        // 重建索引
        async function ingestDocs() {
            const resp = await fetch('/ingest', {method: 'POST'});
            const data = await resp.json();
            alert(data.message);
        }
    </script>
</body>
</html>

13.7.6 运行项目

bash

# 安装依赖
pip install fastapi uvicorn langchain langchain-community \
            langchain-openai chromadb pypdf openai

# 设置 API Key
export OPENAI_API_KEY="sk-your-api-key"

# 把文档放入 data 目录
mkdir -p data
cp /path/to/your/documents/*.txt data/
cp /path/to/your/documents/*.pdf data/

# 启动服务
python main.py

# 访问 http://localhost:8000
# 1. 点击"上传文档"上传文件
# 2. 点击"重建索引"构建向量索引
# 3. 在输入框提问，获取回答

13.7.7 扩展方向

这个项目是"最小可用"版本。如果要上生产，还需要：

对话记忆：加入会话管理，支持多轮对话。
用户认证：加 JWT 认证，区分不同用户的文档库。
文档管理：支持删除、更新文档，增量索引。
结果评估：自动评估回答质量，低质量回答转人工。
监控告警：接入 LangSmith 或自建监控，追踪延迟和错误率。
模型路由：简单问题用 mini 模型，复杂问题用旗舰模型。

常见误区

"索引建一次就行"：文档更新后必须重建索引。生产环境应支持增量更新或定期重建。
"前端不重要"：用户看不到后端代码，前端体验直接决定产品口碑。流式输出是基本要求。
"生产直接用 uvicorn"：开发时用 uvicorn 没问题，生产应该用 gunicorn + uvicorn worker，或多容器部署。
"API Key 硬编码在代码里"：绝对不行。用环境变量或密钥管理服务。
"上传不校验文件类型"：用户可能上传恶意文件。要校验文件类型、大小限制，甚至病毒扫描。

本节小结

步骤	说明
文档加载	支持 TXT/PDF/Markdown
分块索引	RecursiveCharacterTextSplitter + Chroma
问答	检索 + Prompt + LLM 生成
流式输出	SSE 流式推送
前端	原生 HTML/JS，简洁实用
部署	uvicorn 启动，生产用 gunicorn

全书结语

恭喜你读到了这里。

从第一章"什么是 Agent"开始，我们一起走过了一段不短的旅程：从 Agent 的定义与架构，到记忆机制、工具调用、多 Agent 协作；从框架选型到评测调优；从场景题到代码实战。这一章我们把所有知识融会贯通，用真实的、可运行的代码做了一个完整的智能文档问答 Agent。

回顾全书，有三条主线值得铭记：

第一条主线：Agent = 大脑 + 记忆 + 工具 + 规划

这不是一个理论公式，而是工程实践中的设计指南。你在写任何一个 Agent 时，都要问自己四个问题：它用什么模型（大脑）？它怎么记住上下文（记忆）？它能调用什么外部能力（工具）？它怎么决定下一步做什么（规划）？

第二条主线：工程思维 > 模型崇拜

这本书花了大量篇幅讲可靠性、成本、评测、监控——这些不是"无聊的工程细节"，而是决定 Agent 能否上生产的关键。一个用 GPT-4o 但没有错误处理的 Agent，不如一个用 GPT-4o-mini 但有完整护栏的 Agent 靠谱。

第三条主线：动手 > 理论

本书每一章都有代码。不是因为我们觉得代码比理论重要，而是因为 Agent 技术还在快速变化——今天学的框架 API 可能半年后就变了，但"从需求到代码"的工程方法不会变。能动手把想法变成可运行代码的人，永远不缺机会。

关于未来

Agent 技术正处于"从 demo 到产品"的关键过渡期。2024 年大家兴奋于"Agent 能做什么 demo"，2025 年开始认真思考"Agent 怎么上生产"，2026 年及以后的关键问题是"Agent 怎么做到可靠、安全、经济"。

这本书写于 2025 年，书中的一些技术细节可能在不久后就会过时——具体的框架 API 会变，模型价格会降，新的协议会出现。但 Agent 设计的核心方法论——理解需求、合理架构、工程兜底、持续评测——这些不会变。

希望这本书能成为你 Agent 开发之路上的起点，而不是终点。代码在手上，路在脚下。

祝你在 Agent 的世界里，构建出真正有用的东西。

13.7 完整项目实战：智能文档问答 Agent ​

13.7.1 项目架构 ​

13.7.2 项目结构 ​

13.7.3 RAG Pipeline 实现（rag_pipeline.py） ​

13.7.4 FastAPI 主程序（main.py） ​

13.7.5 前端界面（static/index.html） ​

13.7.6 运行项目 ​

13.7.7 扩展方向 ​

常见误区 ​

本节小结 ​

全书结语 ​

关于未来 ​

延伸阅读 ​