AI Agent 搭建全流程实战指南

2.4 向量数据库

RAG 架构的关键存储层。

# ChromaDB（轻量级，适合开发）
pip install chromadb

# Pinecone 客户端（云端生产级）
pip install pinecone-client

# FAISS（Facebook AI 检索）
pip install faiss-cpu
# 若有 CUDA 环境可安装 GPU 版
pip install faiss-gpu

2.5 Web 框架

FastAPI 性能优异且自带文档生成。

pip install fastapi uvicorn flask

3. 项目结构设计

合理的目录结构能显著提升可维护性。建议采用以下布局：

ai-agent/
├── app/
│   ├── main.py          # FastAPI 入口
│   ├── config/          # 配置管理
│   │   └── settings.py
│   ├── components/      # 核心业务组件
│   │   ├── document_loader.py
│   │   ├── text_splitter.py
│   │   ├── vector_store.py
│   │   └── llm_handler.py
│   ├── api/             # API 路由
│   │   ├── endpoints/
│   │   │   ├── docs.py
│   │   │   └── chat.py
│   │   └── routers.py
│   └── services/        # 业务逻辑层
│       ├── document_service.py
│       └── chat_service.py
├── data/                # 数据目录
│   ├── raw/
│   └── processed/
├── models/              # 模型文件
├── tests/               # 测试代码
├── requirements.txt     # 依赖列表
├── .env                 # 环境变量
├── Dockerfile           # 容器配置
└── README.md            # 项目说明

4. 配置文件设置

4.1 创建 .env 文件

敏感信息（如 API Key）严禁硬编码在代码中。

touch .env

4.2 配置环境变量

# OpenAI API 配置
OPENAI_API_KEY=your-openai-api-key
OPENAI_MODEL_NAME=gpt-3.5-turbo

# 向量数据库配置
CHROMA_DB_PATH=./data/chroma_db
PINECONE_API_KEY=your-pinecone-api-key
PINECONE_ENVIRONMENT=your-pinecone-environment
PINECONE_INDEX_NAME=your-index-name

# 应用配置
APP_NAME=AI Agent
APP_VERSION=1.0.0
DEBUG=True

4.3 配置加载器

使用 Pydantic 进行类型安全的配置管理。

# app/config/settings.py
from pydantic_settings import BaseSettings
from typing import Optional

class Settings(BaseSettings):
    # OpenAI 配置
    openai_api_key: str
    openai_model_name: str = "gpt-3.5-turbo"

    # 向量数据库配置
    chroma_db_path: str = "./data/chroma_db"
    pinecone_api_key: Optional[str] = None
    pinecone_environment: Optional[str] = None
    pinecone_index_name: Optional[str] = None

    # 应用配置
    app_name: str = "AI Agent"
    app_version: str = "1.0.0"
    debug: bool = True

class Config:
    env_file = ".env"
    env_file_encoding = "utf-8"

settings = Settings()

5. 核心组件开发

5.1 文档加载器

统一不同格式的文档读取接口。

# app/components/document_loader.py
from langchain_community.document_loaders import (
    PyPDFLoader, Docx2txtLoader, TextLoader, BSHTMLLoader,
    UnstructuredFileLoader
)
from typing import List
from langchain_core.documents import Document

class DocumentLoader:
    @staticmethod
    def load_document(file_path: str) -> List[Document]:
        """根据文件类型加载文档"""
        if file_path.endswith('.pdf'):
            loader = PyPDFLoader(file_path)
        elif file_path.endswith('.docx'):
            loader = Docx2txtLoader(file_path)
        elif file_path.endswith('.txt'):
            loader = TextLoader(file_path, encoding='utf-8')
        elif file_path.endswith('.html'):
            loader = BSHTMLLoader(file_path)
        else:
            loader = UnstructuredFileLoader(file_path)
        return loader.load()

    @staticmethod
    def load_documents(file_paths: List[str]) -> List[Document]:
        """批量加载文档"""
        documents = []
        for file_path in file_paths:
            documents.extend(DocumentLoader.load_document(file_path))
        return documents

5.2 文本分割器

切片策略直接影响检索精度，RecursiveCharacterTextSplitter 是默认优选。

# app/components/text_splitter.py
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_core.documents import Document
from typing import List

class TextSplitter:
    def __init__(self, chunk_size: int = 1000, chunk_overlap: int = 200):
        self.splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
            length_function=len,
            separators=["\n\n", "\n", " ", ""]
        )

    def split_documents(self, documents: List[Document]) -> List[Document]:
        return self.splitter.split_documents(documents)

    def split_text(self, text: str) -> List[str]:
        return self.splitter.split_text(text)

5.3 向量存储

封装向量库操作，方便后续切换后端。

# app/components/vector_store.py
from langchain_community.vectorstores import Chroma, Pinecone
from langchain_openai import OpenAIEmbeddings
from langchain_core.documents import Document
from typing import List, Optional
from app.config.settings import settings

class VectorStoreManager:
    def __init__(self):
        self.embeddings = OpenAIEmbeddings(api_key=settings.openai_api_key)
        # 可选：使用本地嵌入模型
        # self.embeddings = HuggingFaceEmbeddings(model_name="sentence-transformers/all-MiniLM-L6-v2")

    def create_chroma_vector_store(self, documents: List[Document], persist_directory: Optional[str] = None) -> Chroma:
        persist_dir = persist_directory or settings.chroma_db_path
        vector_store = Chroma.from_documents(
            documents=documents,
            embedding=self.embeddings,
            persist_directory=persist_dir
        )
        vector_store.persist()
        return vector_store

    def load_chroma_vector_store(self, persist_directory: Optional[str] = None) -> Chroma:
        persist_dir = persist_directory or settings.chroma_db_path
        return Chroma(embedding_function=self.embeddings, persist_directory=persist_dir)

    def create_pinecone_vector_store(self, documents: List[Document]) -> Pinecone:
        import pinecone
        pinecone.init(api_key=settings.pinecone_api_key, environment=settings.pinecone_environment)
        index_name = settings.pinecone_index_name
        if index_name not in pinecone.list_indexes():
            pinecone.create_index(name=index_name, dimension=1536, metric="cosine")
        vector_store = Pinecone.from_documents(documents=documents, embedding=self.embeddings, index_name=index_name)
        return vector_store

    def load_pinecone_vector_store(self) -> Pinecone:
        import pinecone
        pinecone.init(api_key=settings.pinecone_api_key, environment=settings.pinecone_environment)
        return Pinecone.from_existing_index(index_name=settings.pinecone_index_name, embedding=self.embeddings)

5.4 LLM 处理器

统一管理 Prompt 和模型调用。

# app/components/llm_handler.py
from langchain_openai import ChatOpenAI
from langchain_community.chat_models import ChatOllama
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from app.config.settings import settings

class LLMHandler:
    def __init__(self):
        self.llm = ChatOpenAI(
            api_key=settings.openai_api_key,
            model_name=settings.openai_model_name,
            temperature=0.1
        )
        # 可选：使用本地 Ollama 模型
        # self.llm = ChatOllama(model="llama2", temperature=0.1)

    def generate_response(self, prompt: str, system_prompt: Optional[str] = None) -> str:
        messages = []
        if system_prompt:
            messages.append(SystemMessage(content=system_prompt))
        messages.append(HumanMessage(content=prompt))
        response = self.llm.invoke(messages)
        return response.content

    def generate_with_template(self, template: str, input_variables: dict) -> str:
        prompt_template = ChatPromptTemplate.from_template(template)
        chain = prompt_template | self.llm | StrOutputParser()
        return chain.invoke(input_variables)

6. 业务逻辑服务

6.1 文档处理服务

将原始文件转化为向量索引。

# app/services/document_service.py
from app.components.document_loader import DocumentLoader
from app.components.text_splitter import TextSplitter
from app.components.vector_store import VectorStoreManager
from langchain_core.documents import Document
from typing import List
import os

class DocumentService:
    def __init__(self):
        self.loader = DocumentLoader()
        self.splitter = TextSplitter()
        self.vector_store_manager = VectorStoreManager()

    def process_and_store_document(self, file_path: str) -> bool:
        try:
            documents = self.loader.load_document(file_path)
            split_docs = self.splitter.split_documents(documents)
            self.vector_store_manager.create_chroma_vector_store(split_docs)
            return True
        except Exception as e:
            print(f"处理文档时出错：{e}")
            return False

    def process_and_store_documents(self, file_paths: List[str]) -> dict:
        results = {"success": [], "failed": []}
        for file_path in file_paths:
            if self.process_and_store_document(file_path):
                results["success"].append(file_path)
            else:
                results["failed"].append(file_path)
        return results

    def add_folder_documents(self, folder_path: str) -> dict:
        supported_extensions = ['.pdf', '.docx', '.txt', '.html']
        file_paths = []
        for root, dirs, files in os.walk(folder_path):
            for file in files:
                if any(file.endswith(ext) for ext in supported_extensions):
                    file_paths.append(os.path.join(root, file))
        return self.process_and_store_documents(file_paths)

6.2 聊天服务

基于 RAG 的问答链实现。

# app/services/chat_service.py
from app.components.vector_store import VectorStoreManager
from app.components.llm_handler import LLMHandler
from langchain.chains import RetrievalQA

class ChatService:
    def __init__(self):
        self.vector_store_manager = VectorStoreManager()
        self.llm_handler = LLMHandler()
        self.vector_store = self.vector_store_manager.load_chroma_vector_store()
        self.retriever = self.vector_store.as_retriever(search_type="similarity", search_kwargs={"k": 5})
        self.qa_chain = RetrievalQA.from_chain_type(
            llm=self.llm_handler.llm,
            chain_type="stuff",
            retriever=self.retriever,
            return_source_documents=True
        )

    def get_answer(self, question: str) -> dict:
        result = self.qa_chain.invoke({"query": question})
        sources = []
        for doc in result["source_documents"]:
            content = doc.page_content[:100] + "..." if len(doc.page_content) > 100 else doc.page_content
            sources.append({"page_content": content, "metadata": doc.metadata})
        return {"answer": result["result"], "sources": sources}

    def chat_with_context(self, question: str, chat_history: list = None) -> dict:
        if chat_history:
            context = "\n".join([f"用户：{msg['question']}\nAI: {msg['answer']}" for msg in chat_history])
            prompt = f"上下文:\n{context}\n\n当前问题：{question}"
        else:
            prompt = question
        return self.get_answer(prompt)

7. API 开发

7.1 主应用入口

# app/main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.api.routers import router
from app.config.settings import settings

app = FastAPI(title=settings.app_name, version=settings.app_version, debug=settings.debug)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

app.include_router(router, prefix="/api")

@app.get("/")
def root():
    return {"message": f"Welcome to {settings.app_name}", "version": settings.app_version}

@app.get("/health")
def health_check():
    return {"status": "healthy"}

7.2 路由注册

# app/api/routers.py
from fastapi import APIRouter
from app.api.endpoints import docs, chat

router = APIRouter()
router.include_router(docs.router, prefix="/docs", tags=["documents"])
router.include_router(chat.router, prefix="/chat", tags=["chat"])

7.3 文档 API 端点

# app/api/endpoints/docs.py
from fastapi import APIRouter, UploadFile, File, HTTPException
from fastapi.responses import JSONResponse
from typing import List
import os
import shutil
from app.services.document_service import DocumentService

router = APIRouter()
document_service = DocumentService()
UPLOAD_DIR = "./data/raw"
os.makedirs(UPLOAD_DIR, exist_ok=True)

@router.post("/upload")
async def upload_document(file: UploadFile = File(...)):
    try:
        file_path = os.path.join(UPLOAD_DIR, file.filename)
        with open(file_path, "wb") as buffer:
            shutil.copyfileobj(file.file, buffer)
        success = document_service.process_and_store_document(file_path)
        if success:
            return JSONResponse(status_code=200, content={"message": f"文档 {file.filename} 上传并处理成功"})
        else:
            raise HTTPException(status_code=500, detail=f"文档 {file.filename} 处理失败")
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"上传文档时出错：{str(e)}")

@router.post("/upload-multiple")
async def upload_multiple_documents(files: List[UploadFile] = File(...)):
    file_paths = []
    try:
        for file in files:
            file_path = os.path.join(UPLOAD_DIR, file.filename)
            with open(file_path, "wb") as buffer:
                shutil.copyfileobj(file.file, buffer)
            file_paths.append(file_path)
        results = document_service.process_and_store_documents(file_paths)
        return JSONResponse(status_code=200, content={"message": "多文档上传并处理完成", "results": results})
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"上传多文档时出错：{str(e)}")

@router.post("/add-folder")
async def add_folder(folder_path: str):
    try:
        if not os.path.exists(folder_path):
            raise HTTPException(status_code=404, detail=f"文件夹 {folder_path} 不存在")
        results = document_service.add_folder_documents(folder_path)
        return JSONResponse(status_code=200, content={"message": "文件夹文档添加完成", "results": results})
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"添加文件夹文档时出错：{str(e)}")

7.4 聊天 API 端点

# app/api/endpoints/chat.py
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from app.services.chat_service import ChatService

router = APIRouter()
chat_service = ChatService()

class ChatRequest(BaseModel):
    question: str
    chat_history: list = None

@router.post("/query")
async def query_agent(request: ChatRequest):
    try:
        result = chat_service.chat_with_context(request.question, request.chat_history)
        return result
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"查询时出错：{str(e)}")

8. 前端界面开发（可选）

使用 React + Vite 快速构建管理界面。

8.1 初始化项目

# 安装 Node.js 和 npm
node --version
npm --version

# 创建项目
npm create vite@latest ai-agent-frontend -- --template react
cd ai-agent-frontend
npm install axios react-dropzone

8.2 主要组件

App.jsx

import { useState } from 'react'
import './App.css'
import ChatInterface from './components/ChatInterface'
import DocumentUpload from './components/DocumentUpload'

function App() {
  const [activeTab, setActiveTab] = useState('chat')
  return (
    <div className="app">
      <header className="app-header">
        <h1>AI Agent</h1>
        <nav>
          <button className={activeTab === 'chat' ? 'active' : ''} onClick={() => setActiveTab('chat')}>聊天</button>
          <button className={activeTab === 'upload' ? 'active' : ''} onClick={() => setActiveTab('upload')}>上传文档</button>
        </nav>
      </header>
      <main className="app-main">
        {activeTab === 'chat' && <ChatInterface />}
        {activeTab === 'upload' && <DocumentUpload />}
      </main>
    </div>
  )
}
export default App

ChatInterface.jsx

import { useState, useEffect } from 'react'
import axios from 'axios'

function ChatInterface() {
  const [messages, setMessages] = useState([])
  const [input, setInput] = useState('')
  const [loading, setLoading] = useState(false)

  const sendMessage = async () => {
    if (!input.trim()) return
    const userMessage = { role: 'user', content: input }
    setMessages(prev => [...prev, userMessage])
    setInput('')
    setLoading(true)
    try {
      const response = await axios.post('http://localhost:8000/api/chat/query', {
        question: input,
        chat_history: messages.map(msg => ({
          question: msg.role === 'user' ? msg.content : '',
          answer: msg.role === 'assistant' ? msg.content : ''
        })).filter(msg => msg.question)
      })
      const assistantMessage = { role: 'assistant', content: response.data.answer }
      setMessages(prev => [...prev, assistantMessage])
    } catch (error) {
      console.error('发送消息失败:', error)
      setMessages(prev => [...prev, { role: 'assistant', content: '抱歉，我现在无法回答您的问题。请稍后再试。' }])
    } finally {
      setLoading(false)
    }
  }

  return (
    <div className="chat-interface">
      <div className="chat-messages">
        {messages.map((msg, index) => (
          <div key={index} className={`message ${msg.role}`}>
            <div className="message-content">{msg.content}</div>
          </div>
        ))}
        {loading && (
          <div className="message assistant">
            <div className="message-content">思考中...</div>
          </div>
        )}
      </div>
      <div className="chat-input">
        <input type="text" value={input} onChange={(e) => setInput(e.target.value)} onKeyPress={(e) => e.key === 'Enter' && sendMessage()} placeholder="输入您的问题..." />
        <button onClick={sendMessage} disabled={loading}>发送</button>
      
    
  )
}
export default ChatInterface

DocumentUpload.jsx

import { useCallback } from 'react'
import { useDropzone } from 'react-dropzone'
import axios from 'axios'

function DocumentUpload() {
  const [uploading, setUploading] = useState(false)
  const [result, setResult] = useState(null)

  const onDrop = useCallback(async (acceptedFiles) => {
    setUploading(true)
    setResult(null)
    const formData = new FormData()
    acceptedFiles.forEach(file => { formData.append('files', file) })
    try {
      const response = await axios.post('http://localhost:8000/api/docs/upload-multiple', formData, {
        headers: { 'Content-Type': 'multipart/form-data' }
      })
      setResult(response.data)
    } catch (error) {
      console.error('上传文件失败:', error)
      setResult({ error: '文件上传失败' })
    } finally {
      setUploading(false)
    }
  }, [])

  const { getRootProps, getInputProps, isDragActive } = useDropzone({ onDrop })

  return (
    <div className="document-upload">
      <div {...getRootProps()} className={`dropzone ${isDragActive ? 'active' : ''}`}>
        <input {...getInputProps()} />
        {isDragActive ? (
          <p>放开文件以上传</p>
        ) : (
          <p>拖拽文件到此处，或点击选择文件</p>
        )}
      </div>
      {uploading && <p className="uploading">上传中...</p>}
      {result && (
        <div className="upload-result">
          <h3>上传结果</h3>
          {result.error ? (
            <p className="error">{result.error}</p>
          ) : (
            <>
              <p>成功：{result.results.success.length} 个文件</p>
              <p>失败：{result.results.failed.length} 个文件</p>
              {result.results.success.length > 0 && (
                <div =>
                  成功文件:
                  {result.results.success.map((file, index) => ({file}))}
                
              )}
              {result.results.failed.length > 0 && (
                
                  失败文件:
                  {result.results.failed.map((file, index) => ({file}))}
                
              )}
            
          )}
        
      )}
    </div>
  )
}
export default DocumentUpload

启动开发服务器：npm run dev

9. 测试与调试

9.1 运行后端服务

uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

9.2 访问 API 文档

浏览器打开 http://localhost:8000/docs 查看 Swagger UI。

9.3 测试 API 端点

curl -X POST -H "Content-Type: application/json" -d '{"question": "你好"}' http://localhost:8000/api/chat/query

9.4 单元测试

pip install pytest
pytest tests/

10. 部署与监控

10.1 Docker 部署

Dockerfile

FROM python:3.10-slim
WORKDIR /app
RUN apt-get update && apt-get install -y build-essential curl && rm -rf /var/lib/apt/lists/*
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

构建与运行：

docker build -t ai-agent .
docker run -d -p 8000:8000 --name ai-agent ai-agent

10.2 监控

使用 Prometheus + Grafana 组合监控服务指标。

# Prometheus
docker run -d -p 9090:9090 -v /path/to/prometheus.yml:/etc/prometheus/prometheus.yml prom/prometheus
# Grafana
docker run -d -p 3000:3000 grafana/grafana

prometheus.yml

global:
  scrape_interval: 15s
scrape_configs:
  - job_name: 'ai-agent'
    static_configs:
      - targets: ['ai-agent:8000']

11. 优化与迭代

11.1 性能优化

• 模型优化：尝试量化模型，调整 chunk 大小与 overlap 比例。
• 向量库优化：开启 GPU 加速，定期清理过期索引。
• API 优化：引入 Redis 缓存热点查询结果。

11.2 功能迭代

• 扩展支持的文档格式（如 PPT、Markdown）。
• 增加多语言翻译能力。
• 实现文档自动摘要与分类。

12. 常见问题排查

• 文档加载失败：检查文件编码及格式是否被支持。
• 向量连接失败：确认环境变量中的 API Key 及网络连通性。
• 响应延迟高：检查 Embedding 模型推理速度，考虑降级为本地小模型。

13. 参考资源

• LangChain 官方文档：https://python.langchain.com/
• FastAPI 文档：https://fastapi.tiangolo.com/
• React 文档：https://react.dev/
• ChromaDB GitHub：https://github.com/chroma-core/chroma