AI_Agent_搭建超级详细流程
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 | sh2.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 80009.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 的搭建是一个持续优化的过程,您可以根据实际使用情况不断调整和改进。祝您搭建成功!