从0到1搭建企业级AI智能体：需求分析、架构设计、代码实现全流程
前言
你有没有遇到过这种情况：业务方兴冲冲跑过来，说"我们要做个AI智能体"，然后你问具体做什么，对方说"就是像ChatGPT那样，能回答我们公司自己的问题"。
需求模糊、边界不清、技术栈散乱——这是每一个AI应用工程师都绕不开的坎。
这篇文章不是教你调一个现成的API，而是带你把从需求到上线的完整链路走一遍。用一个真实的"企业知识问答智能体"项目为例，从需求分析开始，到架构设计，再到核心代码实现，最后聊聊上线运维那些事。
一、需求分析：把模糊需求翻译成技术方案
业务方的原始需求往往只有一句话：“我们想要一个AI，能回答员工关于公司制度的问题。”
第一步：拆解业务场景
你需要追问几个关键问题：
| 追问方向 | 业务方回答（示例） | 技术翻译 |
|---|---|---|
| 数据从哪里来？ | 有HR文档、财务制度、IT手册，都是Word/PDF | 文档解析 + 数据预处理 |
| 问题类型是什么？ | “年假怎么休？”“差旅报销流程是什么？” | FAQ问答为主，偶尔需要多轮追问 |
| 需要多高的准确性？ | 回答错了会被吐槽，但不会造成安全事故 | 需要引用来源，支持人工反馈纠正 |
| 用户量多大？ | 公司内部5000人，并发不高 | 不需要大规模集群，单机服务即可 |
| 谁来维护数据更新？ | HR/财务部门每月更新文档 | 需要管理员后台进行数据更新 |
第二步：明确技术边界
经过需求梳理，技术方案逐渐清晰：
- 场景定位：企业内部FAQ知识问答，非任务型对话（不需要订机票、发邮件）
- 技术选型：RAG架构（检索增强生成），不需要Agent Tool Calling
- 数据更新：每月增量更新，不需要实时同步
- 部署方式：内网部署，API服务 + 简易Web界面
第三步：画出用户旅程
这是需求分析中最容易被忽视的一步——你需要在写代码之前，把用户怎么用这个系统想清楚：
- 员工打开Web页面，在输入框提问
- 系统检索相关文档，返回Top-3匹配段落
- 大模型基于检索结果生成回答，并标注引用来源
- 员工对回答点赞/点踩，反馈进入数据库
- 管理员定期查看反馈，修正数据或调整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安装依赖：
pipinstallfastapi uvicorn streamlit chromadb langchain langchain-community sentence-transformers transformers torch pypdf2 python-docx sqlite3四、核心代码实现
4.1 文档加载与分块
# src/document_loader.pyimportosfromtypingimportListfromlangchain_community.document_loadersimportPyPDFLoader,Docx2txtLoader,TextLoaderdefload_documents(directory_path:str)->List:"""加载指定目录下的所有文档"""documents=[]forfileinos.listdir(directory_path):file_path=os.path.join(directory_path,file)iffile.endswith('.pdf'):loader=PyPDFLoader(file_path)eliffile.endswith('.docx'):loader=Docx2txtLoader(file_path)eliffile.endswith('.txt')orfile.endswith('.md'):loader=TextLoader(file_path,encoding='utf-8')else:continuedocs=loader.load()# 给每个文档块打上来源标签fordocindocs:doc.metadata['source']=filedocuments.extend(docs)returndocuments# src/text_splitter.pyfromlangchain.text_splitterimportRecursiveCharacterTextSplitterdefsplit_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)returnchunks为什么选择递归切分？因为它会优先按段落（\n\n）切分，再按句子（。）切分，最后按字符切分。这样能最大程度保证语义完整性——你不会把一个句子的前半部分放在块A、后半部分放在块B。
4.2 Embedding与向量存储
# src/embedding.pyfromsentence_transformersimportSentenceTransformerdefload_embedding_model(model_name:str="BAAI/bge-m3"):"""加载本地Embedding模型"""model=SentenceTransformer(model_name,device="cuda")# 有GPU就用cudareturnmodel# src/vector_store.pyimportchromadbfromchromadb.utilsimportembedding_functionsclassVectorStoreManager: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")defcreate_or_get_collection(self):"""创建或获取已有集合"""returnself.client.get_or_create_collection(name=self.collection_name,embedding_function=self.embedding_fn)defadd_documents(self,chunks:List,batch_size:int=100):"""批量添加文档块到向量库"""collection=self.create_or_get_collection()foriinrange(0,len(chunks),batch_size):batch=chunks[i:i+batch_size]ids=[f"doc_{i}_{j}"forjinrange(len(batch))]texts=[chunk.page_contentforchunkinbatch]metadatas=[chunk.metadataforchunkinbatch]collection.add(documents=texts,metadatas=metadatas,ids=ids)print(f"成功添加{len(chunks)}个文档块到向量库")defsearch(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)returnresultsChroma的优势：它是一个轻量级向量数据库，数据持久化在本地文件中，不需要单独部署服务。对于企业内部5000人使用的场景，Chroma完全够用，不需要上Milvus这种重型方案。
4.3 RAG链：检索 + 生成
# src/rag_chain.pyfromtransformersimportAutoModelForCausalLM,AutoTokenizerimporttorchclassRAGChain: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}"fori,ctxinenumerate(contexts)])prompt=f"""你是一个企业内部知识助手。请根据以下参考资料回答用户问题。 如果参考资料中没有相关信息，请明确告知用户"当前知识库中暂未找到相关信息"。 【参考资料】{context_text}【用户问题】{query}【回答】"""returnpromptdefquery(self,query:str,top_k:int=3)->dict:"""完整查询流程：检索 → 生成"""# 1. 检索相关文档search_results=self.vsm.search(query,top_k=top_k)contexts=search_results['documents'][0]ifsearch_results['documents']else[]sources=search_results['metadatas'][0]ifsearch_results['metadatas']else[]ifnotcontexts:return{"answer":"知识库中暂未找到与您问题相关的信息，请联系管理员更新文档。","sources":[]}# 2. 构建Promptprompt=self._build_prompt(query,contexts)# 3. 调用大模型生成inputs=self.tokenizer(prompt,return_tensors="pt",truncation=True,max_length=4096)withtorch.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','未知来源')forsinsources]))return{"answer":answer,"sources":source_names,"contexts":contexts# 可选，用于调试}4.4 FastAPI服务接口
# api/main.pyfromfastapiimportFastAPI,HTTPExceptionfrompydanticimportBaseModelfromtypingimportOptionalfromsrc.vector_storeimportVectorStoreManagerfromsrc.rag_chainimportRAGChain app=FastAPI(title="企业知识问答智能体")# 全局初始化（服务启动时加载）vsm=VectorStoreManager()rag=RAGChain(vsm,model_path="/path/to/Qwen2.5-7B")classQueryRequest(BaseModel):question:strtop_k:Optional[int]=3classQueryResponse(BaseModel):answer:strsources:listsuccess:bool@app.post("/api/query",response_model=QueryResponse)asyncdefquery(request:QueryRequest):"""知识问答接口"""try:result=rag.query(request.question,top_k=request.top_k)returnQueryResponse(answer=result["answer"],sources=result["sources"],success=True)exceptExceptionase:raiseHTTPException(status_code=500,detail=str(e))@app.get("/api/health")asyncdefhealth():return{"status":"healthy"}# 启动命令：uvicorn api.main:app --host 0.0.0.0 --port 80004.5 Streamlit Web界面
# web/streamlit_app.pyimportstreamlitasstimportrequests st.set_page_config(page_title="企业智能知识助手",layout="wide")st.title("📚 企业知识问答智能体")# 初始化对话历史if"messages"notinst.session_state:st.session_state.messages=[]# 显示历史对话formsginst.session_state.messages:withst.chat_message(msg["role"]):st.write(msg["content"])if"sources"inmsgandmsg["sources"]:st.caption(f"📎 来源：{', '.join(msg['sources'])}")# 输入框ifprompt:=st.chat_input("请问你想了解什么？"):# 显示用户消息withst.chat_message("user"):st.write(prompt)st.session_state.messages.append({"role":"user","content":prompt})# 调用后端APIwithst.chat_message("assistant"):withst.spinner("思考中..."):try:resp=requests.post("http://localhost:8000/api/query",json={"question":prompt,"top_k":3})ifresp.status_code==200:data=resp.json()st.write(data["answer"])ifdata["sources"]:st.caption(f"📎 来源：{', '.join(data['sources'])}")# 保存到历史st.session_state.messages.append({"role":"assistant","content":data["answer"],"sources":data["sources"]})else:st.error("服务异常，请稍后重试")exceptrequests.exceptions.ConnectionError:st.error("无法连接到后端服务，请确认API服务已启动")五、数据导入脚本
# scripts/ingest.pyimportsysimportos sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))fromsrc.document_loaderimportload_documentsfromsrc.text_splitterimportsplit_documentsfromsrc.vector_storeimportVectorStoreManagerdefmain():# 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量化：
fromtransformersimportBitsAndBytesConfig 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消耗。同时，用户反馈的点赞/点踩数据需要定期复盘：
-- 查询点踩最多的类型SELECTSUBSTR(question,1,20)asquestion_sample,COUNT(*)asnegative_countFROMfeedbackWHEREfeedback_type='negative'GROUPBYSUBSTR(question,1,20)ORDERBYnegative_countDESCLIMIT10;这些数据会告诉你：哪些问题是模型经常答错的？是知识库缺失还是Prompt需要优化？反馈闭环是AI应用持续迭代的核心驱动力。
6.3 知识库更新机制
文档更新后，管理员需要重新导入数据。建议设计增量更新策略：
# 增量更新：只删除指定文档的旧向量，再插入新向量defupdate_document(file_name:str,new_chunks:List):collection=vsm.create_or_get_collection()# 删除旧数据collection.delete(where={"source":file_name})# 插入新数据collection.add(documents=[c.page_contentforcinnew_chunks],metadatas=[c.metadataforcinnew_chunks],ids=[f"{file_name}_{i}"foriinrange(len(new_chunks))])七、总结：从代码到项目，差的是什么？
这篇文章写到这里，你已经有了一个完整可运行的企业知识问答智能体。代码能跑起来只是及格线，真正让项目落地的是这三点：
- 需求分析能力：你能把业务方的"我想要个AI"翻译成具体的用户旅程和技术规格，这决定了项目方向对不对
- 工程化思维：代码写完只是30%的工作，剩下的70%是日志、监控、反馈闭环、增量更新
- 成本意识：本地部署BGE-M3和Qwen2.5-7B，省去了API调用的Token费用，长期来看对企业更友好
如果你正在面试JD里描述的那个岗位，这篇文章里的技术栈和代码片段足够你写进项目经验里了。下一篇文章，我会深入讲RAG的评估体系——怎么用Ragas量化你的检索效果，以及怎么通过AB测试验证Prompt优化是否有效。
有问题欢迎在评论区交流，也欢迎关注后续更新！
