AI_Agent_搭建超级详细流程

优质文章学习记录

06 Apr 2026 — 15 min read

AI Agent 搭建超级详细流程

1. 环境准备

1.1 安装 Python

Windows

访问 Python 官网
下载最新的 Python 3.10 或 3.11 版本
运行安装程序，勾选 “Add Python to PATH”
点击 “Install Now”

验证安装

python --`version` pip --version

1.2 创建虚拟环境

# 创建虚拟环境 python -m venv ai-agent-env # 激活虚拟环境# Windows ai-agent-env\Scripts\activate # macOS/Linuxsource ai-agent-env/bin/activate # 更新 pip pip install --upgrade pip

1.3 安装必要工具

# 安装 Git# Windows: 从 https://git-scm.com/ 下载安装# macOS: brew install git# Linux: sudo apt install git# 验证 Git 安装git --version

2. 核心组件安装

2.1 安装 LangChain

pip install langchain langchain-core langchain-community langchain-openai

2.2 安装模型相关库

# 安装 OpenAI 客户端 pip install openai # 安装 Hugging Face 相关库（用于本地模型） pip install transformers torch torchvision torchaudio # 安装 Ollama（用于本地大模型）# Windows: 从 https://ollama.com/download 下载安装# macOS: brew install ollama# Linux: curl -fsSL https://ollama.com/install.sh | sh

2.3 安装文档处理库

# PDF 处理 pip install pypdf pdfplumber # OCR 处理 pip install pytesseract pillow # 安装 Tesseract OCR 引擎# Windows: 从 https://github.com/UB-Mannheim/tesseract/wiki 下载安装# macOS: brew install tesseract# Linux: sudo apt install tesseract-ocr# Word 文档处理 pip install python-docx # HTML 处理 pip install beautifulsoup4

2.4 安装向量数据库

# 安装 ChromaDB（轻量级向量数据库） pip install chromadb # 安装 Pinecone 客户端（云端向量数据库） pip install pinecone-client # 安装 FAISS（Facebook AI Similarity Search） pip install faiss-cpu # 或 GPU 版本（如果有 CUDA） pip install faiss-gpu

2.5 安装 Web 框架

# 安装 FastAPI pip install fastapi uvicorn # 安装 Flask（可选） pip install flask

3. 项目结构设计

ai-agent/ ├── app/ │ ├── main.py # FastAPI 应用入口 │ ├── config/ # 配置文件 │ │ └── settings.py │ ├── components/ # 核心组件 │ │ ├── document_loader.py # 文档加载器 │ │ ├── text_splitter.py # 文本分割器 │ │ ├── vector_store.py # 向量存储 │ │ └── llm_handler.py # LLM 处理 │ ├── api/ # API 路由 │ │ ├── endpoints/ # 具体 API 端点 │ │ │ ├── docs.py # 文档相关 API │ │ │ └── chat.py # 聊天相关 API │ │ └── routers.py # 路由注册 │ └── services/ # 业务逻辑服务 │ ├── document_service.py # 文档处理服务 │ └── chat_service.py # 聊天服务 ├── data/ # 数据目录 │ ├── raw/ # 原始文档 │ └── processed/ # 处理后的数据 ├── models/ # 模型目录 ├── tests/ # 测试代码 ├── requirements.txt # 依赖列表 ├── .env # 环境变量 ├── Dockerfile # Docker 配置 └── README.md # 项目文档

4. 配置文件设置

4.1 创建 .env 文件

touch .env

4.2 配置环境变量

# OpenAI API 配置 OPENAI_API_KEY=your-openai-api-key OPENAI_MODEL_NAME=gpt-3.5-turbo # 向量数据库配置 # ChromaDB 配置（本地） CHROMA_DB_PATH=./data/chroma_db # Pinecone 配置（云端，可选） 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 创建配置加载文件

# app/config/settings.pyfrom pydantic_settings import BaseSettings from typing import Optional classSettings(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=TrueclassConfig: env_file =".env" env_file_encoding ="utf-8"# 创建配置实例 settings = Settings()

5. 核心组件开发

5.1 文档加载器

# app/components/document_loader.pyfrom langchain_community.document_loaders import PyPDFLoader, Docx2txtLoader, TextLoader, BSHTMLLoader from langchain_community.document_loaders import UnstructuredFileLoader from typing import List from langchain_core.documents import Document classDocumentLoader:@staticmethoddefload_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()@staticmethoddefload_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 文本分割器

# app/components/text_splitter.pyfrom langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_core.documents import Document from typing import List classTextSplitter: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"," ",""])defsplit_documents(self, documents: List[Document])-> List[Document]:"""分割文档为小块"""return self.splitter.split_documents(documents)defsplit_text(self, text:str)-> List[str]:"""分割文本为小块"""return self.splitter.split_text(text)

5.3 向量存储

# app/components/vector_store.pyfrom langchain_community.vectorstores import Chroma, Pinecone from langchain_openai import OpenAIEmbeddings from langchain_community.embeddings import HuggingFaceEmbeddings from langchain_core.documents import Document from typing import List, Optional from app.config.settings import settings classVectorStoreManager:def__init__(self):# 初始化嵌入模型 self.embeddings = OpenAIEmbeddings( api_key=settings.openai_api_key )# 可选：使用本地嵌入模型# self.embeddings = HuggingFaceEmbeddings(# model_name="sentence-transformers/all-MiniLM-L6-v2"# )defcreate_chroma_vector_store(self, documents: List[Document], persist_directory: Optional[str]=None)-> Chroma:"""创建 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 defload_chroma_vector_store(self, persist_directory: Optional[str]=None)-> Chroma:"""加载已存在的 Chroma 向量存储""" persist_dir = persist_directory or settings.chroma_db_path return Chroma( embedding_function=self.embeddings, persist_directory=persist_dir )defcreate_pinecone_vector_store(self, documents: List[Document])-> Pinecone:"""创建 Pinecone 向量存储"""import pinecone # 初始化 Pinecone pinecone.init( api_key=settings.pinecone_api_key, environment=settings.pinecone_environment )# 创建或连接索引 index_name = settings.pinecone_index_name if index_name notin pinecone.list_indexes():# 创建索引 pinecone.create_index( name=index_name, dimension=1536,# OpenAI 嵌入维度 metric="cosine")# 向 Pinecone 添加文档 vector_store = Pinecone.from_documents( documents=documents, embedding=self.embeddings, index_name=index_name )return vector_store defload_pinecone_vector_store(self)-> Pinecone:"""加载已存在的 Pinecone 向量存储"""import pinecone # 初始化 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 处理器

# app/components/llm_handler.pyfrom 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 classLLMHandler:def__init__(self):# 初始化 OpenAI LLM 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# )defgenerate_response(self, prompt:str, system_prompt: Optional[str]=None)->str:"""生成 LLM 响应""" messages =[]if system_prompt: messages.append(SystemMessage(content=system_prompt)) messages.append(HumanMessage(content=prompt)) response = self.llm.invoke(messages)return response.content defgenerate_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.pyfrom 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 classDocumentService:def__init__(self): self.loader = DocumentLoader() self.splitter = TextSplitter() self.vector_store_manager = VectorStoreManager()defprocess_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)returnTrueexcept Exception as e:print(f"处理文档时出错: {e}")returnFalsedefprocess_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 defadd_folder_documents(self, folder_path:str)->dict:"""添加文件夹中的所有文档""" supported_extensions =['.pdf','.docx','.txt','.html'] file_paths =[]# 遍历文件夹获取所有支持的文件for root, dirs, files in os.walk(folder_path):forfilein files:ifany(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 聊天服务

# app/services/chat_service.pyfrom app.components.vector_store import VectorStoreManager from app.components.llm_handler import LLMHandler from langchain.chains import RetrievalQA from langchain_core.prompts import ChatPromptTemplate classChatService: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})# 初始化 QA 链 self.qa_chain = RetrievalQA.from_chain_type( llm=self.llm_handler.llm, chain_type="stuff", retriever=self.retriever, return_source_documents=True)defget_answer(self, question:str)->dict:"""根据问题获取答案""" result = self.qa_chain.invoke({"query": question})# 格式化源文档信息 sources =[]for doc in result["source_documents"]: sources.append({"page_content": doc.page_content[:100]+"..."iflen(doc.page_content)>100else doc.page_content,"metadata": doc.metadata })return{"answer": result["result"],"sources": sources }defchat_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.pyfrom fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from app.api.routers import router from app.config.settings import settings # 创建 FastAPI 应用 app = FastAPI( title=settings.app_name, version=settings.app_version, debug=settings.debug )# 配置 CORS app.add_middleware( CORSMiddleware, allow_origins=["*"],# 在生产环境中应替换为具体的域名 allow_credentials=True, allow_methods=["*"], allow_headers=["*"],)# 注册路由 app.include_router(router, prefix="/api")# 根路径@app.get("/")defroot():return{"message":f"Welcome to {settings.app_name}","version": settings.app_version }# 健康检查@app.get("/health")defhealth_check():return{"status":"healthy"}

7.2 路由注册

# app/api/routers.pyfrom 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.pyfrom 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")asyncdefupload_document(file: UploadFile = File(...)):"""上传并处理单个文档"""try:# 保存文件 file_path = os.path.join(UPLOAD_DIR,file.filename)withopen(file_path,"wb")asbuffer: 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")asyncdefupload_multiple_documents(files: List[UploadFile]= File(...)):"""上传并处理多个文档""" file_paths =[]try:# 保存所有文件forfilein files: file_path = os.path.join(UPLOAD_DIR,file.filename)withopen(file_path,"wb")asbuffer: 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")asyncdefadd_folder(folder_path:str):"""添加文件夹中的所有文档"""try:ifnot 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.pyfrom fastapi import APIRouter, HTTPException from pydantic import BaseModel from app.services.chat_service import ChatService router = APIRouter() chat_service = ChatService()# 请求模型classChatRequest(BaseModel): question:str chat_history:[email protected]("/query")asyncdefquery_agent(request: ChatRequest):"""查询 AI Agent"""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. 前端界面开发（可选）

8.1 创建前端项目

# 安装 Node.js 和 npm# 从 https://nodejs.org/ 下载安装# 验证安装node --version npm --version # 创建 React 项目npm create vite@latest ai-agent-frontend -- --template react # 进入项目目录cd ai-agent-frontend # 安装依赖npminstall# 安装必要的库npminstall 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) const errorMessage = { role: 'assistant', content: '抱歉，我现在无法回答您的问题。请稍后再试。' } setMessages(prev => [...prev, errorMessage]) } 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> </div> </div> ) } 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 className="success-list"> <h4>成功文件:</h4> <ul> {result.results.success.map((file, index) => ( <li key={index}>{file}</li> ))} </ul> </div> )} {result.results.failed.length > 0 && ( <div className="failed-list"> <h4>失败文件:</h4> <ul> {result.results.failed.map((file, index) => ( <li key={index}>{file}</li> ))} </ul> </div> )} </> )} </div> )} </div> ) } export default DocumentUpload

8.3 启动前端开发服务器

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，查看自动生成的 API 文档。

9.3 测试 API 端点

使用 Postman 或 curl 测试 API 端点：

# 测试聊天 APIcurl -X POST -H "Content-Type: application/json" -d '{"question": "你好"}' http://localhost:8000/api/chat/query

9.4 单元测试

# 安装测试库 pip install pytest # 创建测试文件# tests/test_chat_service.py# 运行测试 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 . # 安装 Python 依赖 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 镜像

# 构建 Docker 镜像docker build -t ai-agent .# 运行 Docker 容器docker run -d -p 8000:8000 --name ai-agent ai-agent

10.2 监控

安装 Prometheus 和 Grafana

# 使用 Docker 运行 Prometheusdocker run -d -p 9090:9090 -v /path/to/prometheus.yml:/etc/prometheus/prometheus.yml prom/prometheus # 使用 Docker 运行 Grafanadocker run -d -p 3000:3000 grafana/grafana

配置 Prometheus

# prometheus.ymlglobal:scrape_interval: 15s scrape_configs:-job_name:'ai-agent'static_configs:-targets:['ai-agent:8000']

11. 优化与迭代

11.1 性能优化

模型优化：
- 使用模型量化
- 调整 chunk 大小和 overlap
- 使用更高效的嵌入模型
向量数据库优化：
- 调整索引参数
- 使用 GPU 加速（如果可用）
- 定期清理无用数据
API 优化：
- 添加缓存机制
- 使用异步处理
- 优化查询逻辑

11.2 功能迭代

添加更多文档格式支持
实现多语言支持
添加文档摘要功能
实现文档问答功能
添加文档分类功能
实现文档搜索功能

12. 常见问题与解决方案

12.1 文档加载失败

检查文件格式是否支持
检查文件编码是否正确
检查文件是否损坏

12.2 向量存储连接失败

检查环境变量配置
检查数据库服务是否运行
检查网络连接

12.3 API 调用失败

检查 API 端点是否正确
检查请求格式是否正确
检查服务器日志

12.4 响应速度慢

调整 chunk 大小
调整检索参数
考虑使用更强大的模型或硬件

13. 资源推荐

13.1 学习资源

LangChain 文档：https://python.langchain.com/
LlamaIndex 文档：https://gpt-index.readthedocs.io/
FastAPI 文档：https://fastapi.tiangolo.com/
React 文档：https://react.dev/

13.2 开源项目

LangChain：https://github.com/langchain-ai/langchain
LlamaIndex：https://github.com/run-llama/llama_index
Haystack：https://github.com/deepset-ai/haystack
ChromaDB：https://github.com/chroma-core/chroma

13.3 工具

Postman：API 测试工具
Docker：容器化工具
Prometheus + Grafana：监控工具
VS Code：代码编辑器

14. 总结

本教程提供了一个完整的 AI Agent 搭建流程，从环境准备到部署监控，涵盖了所有必要的步骤。您可以根据自己的需求和资源情况，选择适合的组件和配置。

AI Agent 的搭建是一个持续优化的过程，您可以根据实际使用情况不断调整和改进。祝您搭建成功！