AI Agent 的搭建流程，涵盖环境配置（Python、虚拟环境）、核心组件安装（LangChain、向量数据库、LLM 客户端）、项目结构设计、配置文件设置、核心功能开发（文档加载、文本分割、向量存储、LLM 处理）、业务逻辑服务实现、API 接口开发、可选前端界面构建以及 Docker 部署与监控。通过 FastAPI 和 LangChain 框架，实现了文档上传、处理和智能问答功能，并提供了性能优化建议及常见问题解决方案，适合希望快速构建基于大模型的 AI 应用开发者参考。
python --version
pip --version
# 创建虚拟环境
python -m venv ai-agent-env
# 激活虚拟环境
# Windows
ai-agent-env\Scripts\activate
# macOS/Linux
source ai-agent-env/bin/activate
# 更新 pip
pip install --upgrade pip
# 安装 Git
# Windows: 从 https://git-scm.com/ 下载安装
# macOS: brew install git
# Linux: sudo apt install git
# 验证 Git 安装
git --version
pip install langchain langchain-core langchain-community langchain-openai
# 安装 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
# 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
# 安装 ChromaDB（轻量级向量数据库）
pip install chromadb
# 安装 Pinecone 客户端（云端向量数据库）
pip install pinecone-client
# 安装 FAISS（Facebook AI Similarity Search）
pip install faiss-cpu
# 或 GPU 版本（如果有 CUDA）
pip install faiss-gpu
# 安装 FastAPI
pip install fastapi uvicorn
# 安装 Flask（可选）
pip install flask
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 # 项目文档
touch .env
# 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
# 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()
# app/components/document_loader.py
from 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
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
# 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)
# app/components/vector_store.py
from 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
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:
"""创建 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 settings.chroma_db_path
Chroma(
embedding_function=.embeddings,
persist_directory=persist_dir
)
() -> Pinecone:
pinecone
pinecone.init(
api_key=settings.pinecone_api_key,
environment=settings.pinecone_environment
)
index_name = settings.pinecone_index_name
index_name pinecone.list_indexes():
pinecone.create_index(
name=index_name,
dimension=,
metric=
)
vector_store = Pinecone.from_documents(
documents=documents,
embedding=.embeddings,
index_name=index_name
)
vector_store
() -> Pinecone:
pinecone
pinecone.init(
api_key=settings.pinecone_api_key,
environment=settings.pinecone_environment
)
Pinecone.from_existing_index(
index_name=settings.pinecone_index_name,
embedding=.embeddings
)
# 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):
# 初始化 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
# )
def generate_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
def generate_with_template(self, template: str, input_variables: dict) -> :
prompt_template = ChatPromptTemplate.from_template(template)
chain = prompt_template | .llm | StrOutputParser()
chain.invoke(input_variables)
# 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: []) -> :
results = {: [], : []}
file_path file_paths:
.process_and_store_document(file_path):
results[].append(file_path)
:
results[].append(file_path)
results
() -> :
supported_extensions = [, , , ]
file_paths = []
root, dirs, files os.walk(folder_path):
file files:
(file.endswith(ext) ext supported_extensions):
file_paths.append(os.path.join(root, file))
.process_and_store_documents(file_paths)
# app/services/chat_service.py
from app.components.vector_store import VectorStoreManager
from app.components.llm_handler import LLMHandler
from langchain.chains import RetrievalQA
from langchain_core.prompts import ChatPromptTemplate
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}
)
# 初始化 QA 链
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"]:
sources.append({
: doc.page_content[:] + (doc.page_content) > doc.page_content,
: doc.metadata
})
{: result[], : sources}
() -> :
chat_history:
context = .join([ msg chat_history])
prompt =
:
prompt = question
.get_answer(prompt)
# 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
# 创建 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("/")
def root():
return {"message": f"Welcome to {settings.app_name}", "version": settings.app_version}
# 健康检查
@app.get("/health")
def health_check():
return {"status": "healthy"}
# 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"])
# 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=, detail=)
():
file_paths = []
:
file files:
file_path = os.path.join(UPLOAD_DIR, file.filename)
(file_path, ) buffer:
shutil.copyfileobj(file.file, buffer)
file_paths.append(file_path)
results = document_service.process_and_store_documents(file_paths)
JSONResponse(
status_code=,
content={: , : results}
)
Exception e:
HTTPException(status_code=, detail=)
():
:
os.path.exists(folder_path):
HTTPException(status_code=, detail=)
results = document_service.add_folder_documents(folder_path)
JSONResponse(
status_code=,
content={: , : results}
)
Exception e:
HTTPException(status_code=, detail=)
# 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):
"""查询 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)}")
# 安装 Node.js 和 npm
# 从 https://nodejs.org/ 下载安装
# 验证安装
node --version
npm --version
# 创建 React 项目
npm create vite@latest ai-agent-frontend -- --template react
# 进入项目目录
cd ai-agent-frontend
# 安装依赖
npm install
# 安装必要的库
npm install axios react-dropzone
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>
{activeTab === 'chat' && }
{activeTab === 'upload' && }
)
}
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. :
})).( msg.)
})
assistantMessage = { : , : response.. }
( [...prev, assistantMessage])
} (error) {
.(, error)
errorMessage = { : , : }
( [...prev, errorMessage])
} {
()
}
}
(
)
}
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 {
()
}
}, [])
{ getRootProps, getInputProps, isDragActive } = ({ onDrop })
(
)}
</div>
)
}
npm run dev
# 启动后端服务
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000
打开浏览器访问 http://localhost:8000/docs，查看自动生成的 API 文档。
使用 Postman 或 curl 测试 API 端点：
# 测试聊天 API
curl -X POST -H "Content-Type: application/json" -d '{"question": "你好"}' http://localhost:8000/api/chat/query
# 安装测试库
pip install pytest
# 创建测试文件
# tests/test_chat_service.py
# 运行测试
pytest tests/
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 build -t ai-agent .
# 运行 Docker 容器
docker run -d -p 8000:8000 --name ai-agent ai-agent
# 使用 Docker 运行 Prometheus
docker run -d -p 9090:9090 -v /path/to/prometheus.yml:/etc/prometheus/prometheus.yml prom/prometheus
# 使用 Docker 运行 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']
本教程提供了一个完整的 AI Agent 搭建流程，从环境准备到部署监控，涵盖了所有必要的步骤。您可以根据自己的需求和资源情况，选择适合的组件和配置。
AI Agent 的搭建是一个持续优化的过程，您可以根据实际使用情况不断调整和改进。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online