AI应用落地爆发期：如何从零搭建企业级AI应用服务

ai-customer-service/
├── app/
│   ├── main.py              # FastAPI入口
│   ├── config.py            # 配置管理
│   ├── models/              # 数据模型
│   │   ├── chat.py          # 对话相关模型
│   │   └── user.py          # 用户模型
│   ├── services/            # 业务服务
│   │   ├── chat_service.py  # 对话服务
│   │   ├── rag_service.py   # RAG服务
│   │   └── agent_service.py # Agent服务
│   ├── api/                 # API路由
│   │   ├── chat.py          # 对话接口
│   │   └── knowledge.py     # 知识库接口
│   └── utils/               # 工具函数
│       ├── vector_store.py  # 向量数据库封装
│       └── llm_client.py    # LLM客户端封装
├── data/                    # 数据目录
│   └── knowledge/           # 知识库文档
├── requirements.txt         # 依赖清单
└── .env.example             # 环境变量示例

fastapi==0.115.0
uvicorn==0.30.6
langchain==0.3.0
langchain-openai==0.2.0
langchain-community==0.3.0
langchain-chroma==0.1.4
pydantic==2.9.2
pydantic-settings==2.5.2
python-dotenv==1.0.1
python-multipart==0.0.12
sqlalchemy==2.0.35
redis==5.0.8
httpx==0.27.2

# 服务配置
APP_NAME=AI Customer Service
DEBUG=true
PORT=8000

# 大模型配置
OPENAI_API_KEY=your-api-key
OPENAI_BASE_URL=https://api.openai.com/v1
LLM_MODEL=gpt-3.5-turbo
EMBEDDING_MODEL=text-embedding-3-small

# 向量数据库
CHROMA_PERSIST_DIR=./data/chroma
COLLECTION_NAME=knowledge_base

# Redis配置
REDIS_URL=redis://localhost:6379/0

# 数据库配置
DATABASE_URL=sqlite:///./data/app.db

from pydantic_settings import BaseSettings
from functools import lru_cache


class Settings(BaseSettings):
    """应用配置类"""
    
    # 服务配置
    app_name: str = "AI Customer Service"
    debug: bool = False
    port: int = 8000
    
    # 大模型配置
    openai_api_key: str
    openai_base_url: str = "https://api.openai.com/v1"
    llm_model: str = "gpt-3.5-turbo"
    embedding_model: str = "text-embedding-3-small"
    temperature: float = 0.1
    max_tokens: int = 2048
    
    # 向量数据库
    chroma_persist_dir: str = "./data/chroma"
    collection_name: str = "knowledge_base"
    
    # Redis
    redis_url: str = "redis://localhost:6379/0"
    
    # 数据库
    database_url: str = "sqlite:///./data/app.db"
    
    # RAG配置
    chunk_size: int = 500
    chunk_overlap: int = 50
    top_k: int = 4
    
    class Config:
        env_file = ".env"


@lru_cache()
def get_settings() -> Settings:
    """获取配置单例"""
    return Settings()

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from app.config import get_settings

settings = get_settings()


def get_llm(temperature: float = None, max_tokens: int = None) -> ChatOpenAI:
    """获取LLM实例"""
    return ChatOpenAI(
        model=settings.llm_model,
        temperature=temperature if temperature is not None else settings.temperature,
        max_tokens=max_tokens if max_tokens is not None else settings.max_tokens,
        api_key=settings.openai_api_key,
        base_url=settings.openai_base_url,
    )


def get_embeddings() -> OpenAIEmbeddings:
    """获取Embedding实例"""
    return OpenAIEmbeddings(
        model=settings.embedding_model,
        api_key=settings.openai_api_key,
        base_url=settings.openai_base_url,
    )

from langchain_chroma import Chroma
from langchain_community.document_loaders import (
    PyPDFLoader,
    TextLoader,
    Docx2txtLoader,
    UnstructuredMarkdownLoader,
)
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.documents import Document
from typing import List, Optional
import os

from app.config import get_settings
from app.utils.llm_client import get_embeddings

settings = get_settings()


class RAGService:
    """RAG知识库服务"""
    
    def __init__(self):
        self.embeddings = get_embeddings()
        self.vector_store = self._init_vector_store()
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=settings.chunk_size,
            chunk_overlap=settings.chunk_overlap,
            separators=["

", "
", ".", ",", " ", ""],
        )
    
    def _init_vector_store(self) -> Chroma:
        """初始化向量数据库"""
        os.makedirs(settings.chroma_persist_dir, exist_ok=True)
        return Chroma(
            collection_name=settings.collection_name,
            embedding_function=self.embeddings,
            persist_directory=settings.chroma_persist_dir,
        )
    
    def _load_document(self, file_path: str) -> List[Document]:
        """根据文件类型加载文档"""
        ext = os.path.splitext(file_path)[1].lower()
        loaders = {
            ".pdf": PyPDFLoader,
            ".txt": TextLoader,
            ".docx": Docx2txtLoader,
            ".md": UnstructuredMarkdownLoader,
        }
        loader_class = loaders.get(ext)
        if not loader_class:
            raise ValueError(f"不支持的文件格式: {ext}")
        return loader_class(file_path).load()
    
    def add_document(self, file_path: str, metadata: dict = None) -> int:
        """添加文档到知识库
        
        Args:
            file_path: 文件路径
            metadata: 元数据（如来源、分类等）
            
        Returns:
            添加的文档块数量
        """
        # 加载文档
        documents = self._load_document(file_path)
        
        # 添加元数据
        if metadata:
            for doc in documents:
                doc.metadata.update(metadata)
        
        # 文本分块
        chunks = self.text_splitter.split_documents(documents)
        
        # 添加到向量数据库
        if chunks:
            self.vector_store.add_documents(chunks)
        
        return len(chunks)
    
    def similarity_search(
        self, 
        query: str, 
        top_k: int = None,
        filter: Optional[dict] = None,
    ) -> List[Document]:
        """相似度检索
        
        Args:
            query: 查询文本
            top_k: 返回结果数量
            filter: 元数据过滤条件
            
        Returns:
            相关文档块列表
        """
        top_k = top_k or settings.top_k
        return self.vector_store.similarity_search(
            query, 
            k=top_k,
            filter=filter,
        )
    
    def delete_by_source(self, source: str) -> None:
        """根据来源删除文档"""
        # Chroma的delete操作需要获取ID后批量删除
        # 这里简化处理，实际项目中需要更完善的管理
        pass
    
    def get_stats(self) -> dict:
        """获取知识库统计信息"""
        collection = self.vector_store._collection
        return {
            "total_docs": collection.count(),
            "collection_name": settings.collection_name,
        }


# 全局单例
rag_service = RAGService()

from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.tools import tool
from langchain.memory import ConversationBufferMemory

from app.utils.llm_client import get_llm
from app.services.rag_service import rag_service


@tool
def search_knowledge_base(query: str) -> str:
    """在企业知识库中搜索相关信息。
    当用户询问产品信息、公司政策、操作指南等内部知识时，使用此工具。
    
    Args:
        query: 搜索查询，描述要查找的信息
    """
    docs = rag_service.similarity_search(query)
    if not docs:
        return "知识库中未找到相关信息。"
    
    result_parts = []
    for i, doc in enumerate(docs, 1):
        source = doc.metadata.get("source", "未知来源")
        result_parts.append(f"[文档{i}] 来源: {source}
内容: {doc.page_content}
")
    
    return "
".join(result_parts)


@tool
def get_order_status(order_id: str) -> str:
    """查询订单状态。当用户询问订单物流、发货情况时使用。
    
    Args:
        order_id: 订单编号，通常以ORD开头
    """
    # 模拟实现，实际项目中对接订单系统API
    mock_data = {
        "ORD001": "订单已发货，物流单号：SF123456789，预计明天送达",
        "ORD002": "订单正在仓库打包，预计今天下午发货",
        "ORD003": "订单已完成，签收时间：2024-01-15 14:30",
    }
    return mock_data.get(order_id, f"未找到订单 {order_id} 的信息，请确认订单号是否正确")


class AgentService:
    """智能体服务"""
    
    def __init__(self):
        self.llm = get_llm()
        self.tools = [search_knowledge_base, get_order_status]
        self.agent_executors = {}  # session_id -> executor
    
    def _create_agent_executor(self, session_id: str) -> AgentExecutor:
        """创建Agent执行器"""
        prompt = ChatPromptTemplate.from_messages([
            ("system", """你是一个专业的企业客服助手，名字叫小智。
你的职责是帮助用户解答产品相关问题、查询订单状态、提供操作指导。

回答规则：
1. 优先使用知识库中的信息回答问题
2. 如果知识库中没有答案，可以说"我需要更多信息"或转人工
3. 回答要简洁明了，专业友好
4. 涉及订单查询时，使用订单查询工具
5. 不要编造信息，不确定的内容要明确说明

当前对话ID：{session_id}"""),
            MessagesPlaceholder(variable_name="chat_history"),
            ("human", "{input}"),
            MessagesPlaceholder(variable_name="agent_scratchpad"),
        ])
        
        # 填充session_id变量
        prompt = prompt.partial(session_id=session_id)
        
        # 创建Agent
        agent = create_openai_tools_agent(self.llm, self.tools, prompt)
        
        # 对话记忆
        memory = ConversationBufferMemory(
            memory_key="chat_history",
            return_messages=True,
        )
        
        # 创建执行器
        executor = AgentExecutor(
            agent=agent,
            tools=self.tools,
            memory=memory,
            verbose=True,
            max_iterations=5,
            handle_parsing_errors=True,
        )
        
        return executor
    
    def get_executor(self, session_id: str) -> AgentExecutor:
        """获取或创建会话对应的Agent执行器"""
        if session_id not in self.agent_executors:
            self.agent_executors[session_id] = self._create_agent_executor(session_id)
        return self.agent_executors[session_id]
    
    async def chat(self, session_id: str, message: str) -> str:
        """发送消息给Agent并获取回复
        
        Args:
            session_id: 会话ID
            message: 用户消息
            
        Returns:
            Agent回复内容
        """
        executor = self.get_executor(session_id)
        result = await executor.ainvoke({"input": message})
        return result["output"]
    
    def clear_session(self, session_id: str) -> bool:
        """清除会话记忆"""
        if session_id in self.agent_executors:
            del self.agent_executors[session_id]
            return True
        return False


# 全局单例
agent_service = AgentService()

from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from typing import Optional
import uuid

from app.services.agent_service import agent_service

router = APIRouter(prefix="/api/chat", tags=["对话"])


class ChatRequest(BaseModel):
    session_id: Optional[str] = None
    message: str


class ChatResponse(BaseModel):
    session_id: str
    reply: str


@router.post("/send", response_model=ChatResponse)
async def send_message(request: ChatRequest):
    """发送消息并获取回复"""
    try:
        # 如果没有session_id，自动创建
        session_id = request.session_id or str(uuid.uuid4())
        
        # 调用Agent服务
        reply = await agent_service.chat(session_id, request.message)
        
        return ChatResponse(
            session_id=session_id,
            reply=reply,
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"对话服务异常: {str(e)}")


@router.post("/clear/{session_id}")
async def clear_session(session_id: str):
    """清除会话历史"""
    success = agent_service.clear_session(session_id)
    return {"success": success, "session_id": session_id}

from fastapi import APIRouter, UploadFile, File, HTTPException
from pydantic import BaseModel
import os
import tempfile

from app.services.rag_service import rag_service

router = APIRouter(prefix="/api/knowledge", tags=["知识库"])


class KnowledgeStatsResponse(BaseModel):
    total_docs: int
    collection_name: str


@router.get("/stats", response_model=KnowledgeStatsResponse)
async def get_knowledge_stats():
    """获取知识库统计信息"""
    try:
        stats = rag_service.get_stats()
        return KnowledgeStatsResponse(**stats)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@router.post("/upload")
async def upload_document(file: UploadFile = File(...)):
    """上传文档到知识库"""
    try:
        # 创建临时文件
        suffix = os.path.splitext(file.filename)[1]
        with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
            content = await file.read()
            tmp.write(content)
            tmp_path = tmp.name
        
        # 添加到知识库
        chunk_count = rag_service.add_document(
            tmp_path,
            metadata={"source": file.filename, "uploaded_by": "api"}
        )
        
        # 清理临时文件
        os.unlink(tmp_path)
        
        return {
            "success": True,
            "filename": file.filename,
            "chunks_added": chunk_count,
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"上传失败: {str(e)}")

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager

from app.config import get_settings
from app.api.chat import router as chat_router
from app.api.knowledge import router as knowledge_router

settings = get_settings()


@asynccontextmanager
async def lifespan(app: FastAPI):
    """应用生命周期管理"""
    # 启动时初始化
    print(f"{settings.app_name} 启动中...")
    print(f"模型: {settings.llm_model}")
    print(f"知识库: {settings.collection_name}")
    yield
    # 关闭时清理
    print("应用关闭中...")


app = FastAPI(
    title=settings.app_name,
    description="企业级AI客服系统API文档",
    version="1.0.0",
    lifespan=lifespan,
)

# CORS配置
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 生产环境需限制域名
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 注册路由
app.include_router(chat_router)
app.include_router(knowledge_router)


@app.get("/health")
async def health_check():
    """健康检查"""
    return {"status": "ok", "app": settings.app_name}

pip install -r requirements.txt
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

curl -X POST http://localhost:8000/api/chat/send \
  -H "Content-Type: application/json" \
  -d '{"message": "你们的产品支持定制吗？"}'

curl -X POST http://localhost:8000/api/knowledge/upload \
  -F "file=@product_manual.pdf"