从0到1搭建企业级AI智能体：需求分析、架构设计、代码实现全流程

前言

你有没有遇到过这种情况：业务方兴冲冲跑过来，说"我们要做个AI智能体"，然后你问具体做什么，对方说"就是像ChatGPT那样，能回答我们公司自己的问题"。

需求模糊、边界不清、技术栈散乱——这是每一个AI应用工程师都绕不开的坎。

这篇文章不是教你调一个现成的API，而是带你把从需求到上线的完整链路走一遍。用一个真实的"企业知识问答智能体"项目为例，从需求分析开始，到架构设计，再到核心代码实现，最后聊聊上线运维那些事。

---

一、需求分析：把模糊需求翻译成技术方案

业务方的原始需求往往只有一句话：“我们想要一个AI，能回答员工关于公司制度的问题。”

第一步：拆解业务场景

你需要追问几个关键问题：

追问方向	业务方回答（示例）	技术翻译
数据从哪里来？	有HR文档、财务制度、IT手册，都是Word/PDF	文档解析 + 数据预处理
问题类型是什么？	“年假怎么休？”“差旅报销流程是什么？”	FAQ问答为主，偶尔需要多轮追问
需要多高的准确性？	回答错了会被吐槽，但不会造成安全事故	需要引用来源，支持人工反馈纠正
用户量多大？	公司内部5000人，并发不高	不需要大规模集群，单机服务即可
谁来维护数据更新？	HR/财务部门每月更新文档	需要管理员后台进行数据更新

第二步：明确技术边界

经过需求梳理，技术方案逐渐清晰：

• 场景定位：企业内部FAQ知识问答，非任务型对话（不需要订机票、发邮件）
• 技术选型：RAG架构（检索增强生成），不需要Agent Tool Calling
• 数据更新：每月增量更新，不需要实时同步
• 部署方式：内网部署，API服务 + 简易Web界面

第三步：画出用户旅程

这是需求分析中最容易被忽视的一步——你需要在写代码之前，把用户怎么用这个系统想清楚：

1. 员工打开Web页面，在输入框提问
2. 系统检索相关文档，返回Top-3匹配段落
3. 大模型基于检索结果生成回答，并标注引用来源
4. 员工对回答点赞/点踩，反馈进入数据库
5. 管理员定期查看反馈，修正数据或调整Prompt

用户旅程画出来后，你会发现很多设计问题会提前暴露。比如"引用来源"这个功能，如果不提前设计，你可能会在后期推翻数据结构重写。

---

二、架构设计：选型决策与系统蓝图

基于上面的需求分析，我们定下技术栈：

模块	技术选型	选型理由
文档解析	PyPDF2 + python-docx + markdown	覆盖主流文档格式
文本分块	RecursiveCharacterTextSplitter	LangChain内置，支持按段落/句子递归切分
Embedding模型	BGE-M3（本地部署）	中文效果好，数据不出内网
向量数据库	Chroma（本地持久化）	轻量级，无需额外部署服务，适合中小企业
大模型	Qwen2.5-7B（本地部署）	开源免费，中文能力强，单卡可跑
服务框架	FastAPI	高性能异步框架，自动生成API文档
前端界面	Streamlit	快速搭建演示界面，无需前端开发经验
反馈收集	SQLite	轻量级数据库，存储用户反馈日志

系统架构图（逻辑流向）：

用户提问（Web界面）
    ↓
FastAPI接收请求
    ↓
向量检索（Chroma）→ 返回Top-3相关文档片段
    ↓
Prompt组装（检索结果 + 系统指令 + 用户问题）
    ↓
调用Qwen2.5-7B生成回答
    ↓
返回回答 + 引用来源
    ↓
（可选）用户反馈 → 存入SQLite

---

三、环境准备与项目结构

创建项目目录，建议按下面这个结构组织代码：

ai_knowledge_bot/
├── config/
│   └── settings.py          # 配置文件（模型路径、API端口等）
├── data/
│   ├── raw/                  # 原始文档存放目录
│   └── processed/            # 处理后的文档
├── src/
│   ├── document_loader.py    # 文档解析与加载
│   ├── text_splitter.py      # 文本分块
│   ├── embedding.py          # Embedding模型加载
│   ├── vector_store.py       # 向量数据库操作
│   ├── llm.py               # 大模型调用
│   ├── rag_chain.py         # RAG完整流程
│   └── feedback.py          # 反馈收集
├── api/
│   └── main.py              # FastAPI服务入口
├── web/
│   └── streamlit_app.py     # Web界面
├── scripts/
│   └── ingest.py            # 文档导入脚本
├── requirements.txt
└── README.md

安装依赖：

pip install fastapi uvicorn streamlit chromadb langchain langchain-community sentence-transformers transformers torch pypdf2 python-docx sqlite3

---

四、核心代码实现

4.1 文档加载与分块

# src/document_loader.py
import os
from typing import List
from langchain_community.document_loaders import PyPDFLoader, Docx2txtLoader, TextLoader

def load_documents(directory_path: str) -> List:
    """加载指定目录下的所有文档"""
    documents = []
    for file in os.listdir(directory_path):
        file_path = os.path.join(directory_path, file)
        if file.endswith('.pdf'):
            loader = PyPDFLoader(file_path)
        elif file.endswith('.docx'):
            loader = Docx2txtLoader(file_path)
        elif file.endswith('.txt') or file.endswith('.md'):
            loader = TextLoader(file_path, encoding='utf-8')
        else:
            continue
        docs = loader.load()
        # 给每个文档块打上来源标签
        for doc in docs:
            doc.metadata['source'] = file
        documents.extend(docs)
    return documents

# src/text_splitter.py
from langchain.text_splitter import RecursiveCharacterTextSplitter

def split_documents(documents: List, chunk_size: int = 500, chunk_overlap: int = 50):
    """
    递归切分文档
    chunk_size: 每块最大字符数
    chunk_overlap: 块间重叠字符数，保证上下文连贯
    """
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""],
        length_function=len,
    )
    chunks = text_splitter.split_documents(documents)
    return chunks

为什么选择递归切分？ 因为它会优先按段落（\n\n）切分，再按句子（。）切分，最后按字符切分。这样能最大程度保证语义完整性——你不会把一个句子的前半部分放在块A、后半部分放在块B。

4.2 Embedding与向量存储

# src/embedding.py
from sentence_transformers import SentenceTransformer

def load_embedding_model(model_name: str = "BAAI/bge-m3"):
    """加载本地Embedding模型"""
    model = SentenceTransformer(model_name, device="cuda")  # 有GPU就用cuda
    return model

# src/vector_store.py
import chromadb
from chromadb.utils import embedding_functions

class VectorStoreManager:
    def __init__(self, persist_directory: str = "./chroma_db"):
        self.client = chromadb.PersistentClient(path=persist_directory)
        self.collection_name = "knowledge_base"
        # 使用BGE-M3的Embedding函数
        self.embedding_fn = embedding_functions.SentenceTransformerEmbeddingFunction(
            model_name="BAAI/bge-m3"
        )
    
    def create_or_get_collection(self):
        """创建或获取已有集合"""
        return self.client.get_or_create_collection(
            name=self.collection_name,
            embedding_function=self.embedding_fn
        )
    
    def add_documents(self, chunks: List, batch_size: int = 100):
        """批量添加文档块到向量库"""
        collection = self.create_or_get_collection()
        
        for i in range(0, len(chunks), batch_size):
            batch = chunks[i:i+batch_size]
            ids = [f"doc_{i}_{j}" for j in range(len(batch))]
            texts = [chunk.page_content for chunk in batch]
            metadatas = [chunk.metadata for chunk in batch]
            
            collection.add(
                documents=texts,
                metadatas=metadatas,
                ids=ids
            )
        print(f"成功添加 {len(chunks)} 个文档块到向量库")
    
    def search(self, query: str, top_k: int = 3):
        """检索最相关的Top-K文档块"""
        collection = self.create_or_get_collection()
        results = collection.query(
            query_texts=[query],
            n_results=top_k
        )
        return results

Chroma的优势：它是一个轻量级向量数据库，数据持久化在本地文件中，不需要单独部署服务。对于企业内部5000人使用的场景，Chroma完全够用，不需要上Milvus这种重型方案。

4.3 RAG链：检索 + 生成

# src/rag_chain.py
from transformers import AutoModelForCausalLM, AutoTokenizer
import torch

class RAGChain:
    def __init__(self, vector_store_manager, model_path: str):
        self.vsm = vector_store_manager
        # 加载本地大模型
        self.tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True)
        self.model = AutoModelForCausalLM.from_pretrained(
            model_path,
            torch_dtype=torch.float16,
            device_map="auto",
            trust_remote_code=True
        )
    
    def _build_prompt(self, query: str, contexts: List[str]) -> str:
        """构建带上下文的Prompt"""
        context_text = "\n\n".join([f"【参考资料{i+1}】{ctx}" for i, ctx in enumerate(contexts)])
        
        prompt = f"""你是一个企业内部知识助手。请根据以下参考资料回答用户问题。
如果参考资料中没有相关信息，请明确告知用户"当前知识库中暂未找到相关信息"。

【参考资料】
{context_text}

【用户问题】
{query}

【回答】"""
        return prompt
    
    def query(self, query: str, top_k: int = 3) -> dict:
        """完整查询流程：检索 → 生成"""
        # 1. 检索相关文档
        search_results = self.vsm.search(query, top_k=top_k)
        
        contexts = search_results['documents'][0] if search_results['documents'] else []
        sources = search_results['metadatas'][0] if search_results['metadatas'] else []
        
        if not contexts:
            return {
                "answer": "知识库中暂未找到与您问题相关的信息，请联系管理员更新文档。",
                "sources": []
            }
        
        # 2. 构建Prompt
        prompt = self._build_prompt(query, contexts)
        
        # 3. 调用大模型生成
        inputs = self.tokenizer(prompt, return_tensors="pt", truncation=True, max_length=4096)
        with torch.no_grad():
            outputs = self.model.generate(
                **inputs,
                max_new_tokens=512,
                temperature=0.7,
                top_p=0.9,
                do_sample=True,
            )
        answer = self.tokenizer.decode(outputs[0][inputs['input_ids'].shape[1]:], skip_special_tokens=True)
        
        # 4. 提取来源文件名
        source_names = list(set([s.get('source', '未知来源') for s in sources]))
        
        return {
            "answer": answer,
            "sources": source_names,
            "contexts": contexts  # 可选，用于调试
        }

4.4 FastAPI服务接口

# api/main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional
from src.vector_store import VectorStoreManager
from src.rag_chain import RAGChain

app = FastAPI(title="企业知识问答智能体")

# 全局初始化（服务启动时加载）
vsm = VectorStoreManager()
rag = RAGChain(vsm, model_path="/path/to/Qwen2.5-7B")

class QueryRequest(BaseModel):
    question: str
    top_k: Optional[int] = 3

class QueryResponse(BaseModel):
    answer: str
    sources: list
    success: bool

@app.post("/api/query", response_model=QueryResponse)
async def query(request: QueryRequest):
    """知识问答接口"""
    try:
        result = rag.query(request.question, top_k=request.top_k)
        return QueryResponse(
            answer=result["answer"],
            sources=result["sources"],
            success=True
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

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

# 启动命令：uvicorn api.main:app --host 0.0.0.0 --port 8000

4.5 Streamlit Web界面

# web/streamlit_app.py
import streamlit as st
import requests

st.set_page_config(page_title="企业智能知识助手", layout="wide")
st.title("📚 企业知识问答智能体")

# 初始化对话历史
if "messages" not in st.session_state:
    st.session_state.messages = []

# 显示历史对话
for msg in st.session_state.messages:
    with st.chat_message(msg["role"]):
        st.write(msg["content"])
        if "sources" in msg and msg["sources"]:
            st.caption(f"📎 来源：{', '.join(msg['sources'])}")

# 输入框
if prompt := st.chat_input("请问你想了解什么？"):
    # 显示用户消息
    with st.chat_message("user"):
        st.write(prompt)
    st.session_state.messages.append({"role": "user", "content": prompt})
    
    # 调用后端API
    with st.chat_message("assistant"):
        with st.spinner("思考中..."):
            try:
                resp = requests.post(
                    "http://localhost:8000/api/query",
                    json={"question": prompt, "top_k": 3}
                )
                if resp.status_code == 200:
                    data = resp.json()
                    st.write(data["answer"])
                    if data["sources"]:
                        st.caption(f"📎 来源：{', '.join(data['sources'])}")
                    # 保存到历史
                    st.session_state.messages.append({
                        "role": "assistant",
                        "content": data["answer"],
                        "sources": data["sources"]
                    })
                else:
                    st.error("服务异常，请稍后重试")
            except requests.exceptions.ConnectionError:
                st.error("无法连接到后端服务，请确认API服务已启动")

---

五、数据导入脚本

# scripts/ingest.py
import sys
import os
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from src.document_loader import load_documents
from src.text_splitter import split_documents
from src.vector_store import VectorStoreManager

def main():
    # 1. 加载文档
    print("正在加载文档...")
    docs = load_documents("./data/raw")
    print(f"加载了 {len(docs)} 个文档")
    
    # 2. 分块
    print("正在切分文档...")
    chunks = split_documents(docs, chunk_size=500, chunk_overlap=50)
    print(f"切分为 {len(chunks)} 个块")
    
    # 3. 存入向量库
    print("正在存入向量数据库...")
    vsm = VectorStoreManager()
    vsm.add_documents(chunks)
    print("导入完成！")

if __name__ == "__main__":
    main()

---

六、上线与运维：别忘了这些事

代码写完只是第一步，上线之后才是真正的开始。根据我的项目经验，这三个点最容易踩坑：

6.1 模型推理显存优化

Qwen2.5-7B在FP16下需要约14GB显存。如果你的服务器显存不够，可以启用4-bit量化：

from transformers import BitsAndBytesConfig

quantization_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_compute_dtype=torch.float16
)

model = AutoModelForCausalLM.from_pretrained(
    model_path,
    quantization_config=quantization_config,
    device_map="auto"
)

4-bit量化后显存占用降到约6GB，推理速度损失约20%，对于企业内部使用完全可以接受。

6.2 日志与反馈闭环

在api/main.py中添加日志中间件，记录每次查询的耗时和Token消耗。同时，用户反馈的点赞/点踩数据需要定期复盘：

-- 查询点踩最多的类型
SELECT 
    SUBSTR(question, 1, 20) as question_sample,
    COUNT(*) as negative_count
FROM feedback
WHERE feedback_type = 'negative'
GROUP BY SUBSTR(question, 1, 20)
ORDER BY negative_count DESC
LIMIT 10;

这些数据会告诉你：哪些问题是模型经常答错的？是知识库缺失还是Prompt需要优化？反馈闭环是AI应用持续迭代的核心驱动力。

6.3 知识库更新机制

文档更新后，管理员需要重新导入数据。建议设计增量更新策略：

# 增量更新：只删除指定文档的旧向量，再插入新向量
def update_document(file_name: str, new_chunks: List):
    collection = vsm.create_or_get_collection()
    # 删除旧数据
    collection.delete(where={"source": file_name})
    # 插入新数据
    collection.add(
        documents=[c.page_content for c in new_chunks],
        metadatas=[c.metadata for c in new_chunks],
        ids=[f"{file_name}_{i}" for i in range(len(new_chunks))]
    )

---

七、总结：从代码到项目，差的是什么？

这篇文章写到这里，你已经有了一个完整可运行的企业知识问答智能体。代码能跑起来只是及格线，真正让项目落地的是这三点：

1. 需求分析能力：你能把业务方的"我想要个AI"翻译成具体的用户旅程和技术规格，这决定了项目方向对不对
2. 工程化思维：代码写完只是30%的工作，剩下的70%是日志、监控、反馈闭环、增量更新
3. 成本意识：本地部署BGE-M3和Qwen2.5-7B，省去了API调用的Token费用，长期来看对企业更友好

如果你正在面试JD里描述的那个岗位，这篇文章里的技术栈和代码片段足够你写进项目经验里了。下一篇文章，我会深入讲RAG的评估体系——怎么用Ragas量化你的检索效果，以及怎么通过AB测试验证Prompt优化是否有效。

有问题欢迎在评论区交流，也欢迎关注后续更新！