从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覆盖主流文档格式文本分块RecursiveCharacterTextSplitterLangChain内置支持按段落/句子递归切分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_pathos.path.join(directory_path,file)iffile.endswith(.pdf):loaderPyPDFLoader(file_path)eliffile.endswith(.docx):loaderDocx2txtLoader(file_path)eliffile.endswith(.txt)orfile.endswith(.md):loaderTextLoader(file_path,encodingutf-8)else:continuedocsloader.load()# 给每个文档块打上来源标签fordocindocs:doc.metadata[source]filedocuments.extend(docs)returndocuments# src/text_splitter.pyfromlangchain.text_splitterimportRecursiveCharacterTextSplitterdefsplit_documents(documents:List,chunk_size:int500,chunk_overlap:int50): 递归切分文档 chunk_size: 每块最大字符数 chunk_overlap: 块间重叠字符数保证上下文连贯 text_splitterRecursiveCharacterTextSplitter(chunk_sizechunk_size,chunk_overlapchunk_overlap,separators[\n\n,\n,。,,,,, ,],length_functionlen,)chunkstext_splitter.split_documents(documents)returnchunks为什么选择递归切分因为它会优先按段落\n\n切分再按句子。切分最后按字符切分。这样能最大程度保证语义完整性——你不会把一个句子的前半部分放在块A、后半部分放在块B。4.2 Embedding与向量存储# src/embedding.pyfromsentence_transformersimportSentenceTransformerdefload_embedding_model(model_name:strBAAI/bge-m3):加载本地Embedding模型modelSentenceTransformer(model_name,devicecuda)# 有GPU就用cudareturnmodel# src/vector_store.pyimportchromadbfromchromadb.utilsimportembedding_functionsclassVectorStoreManager:def__init__(self,persist_directory:str./chroma_db):self.clientchromadb.PersistentClient(pathpersist_directory)self.collection_nameknowledge_base# 使用BGE-M3的Embedding函数self.embedding_fnembedding_functions.SentenceTransformerEmbeddingFunction(model_nameBAAI/bge-m3)defcreate_or_get_collection(self):创建或获取已有集合returnself.client.get_or_create_collection(nameself.collection_name,embedding_functionself.embedding_fn)defadd_documents(self,chunks:List,batch_size:int100):批量添加文档块到向量库collectionself.create_or_get_collection()foriinrange(0,len(chunks),batch_size):batchchunks[i:ibatch_size]ids[fdoc_{i}_{j}forjinrange(len(batch))]texts[chunk.page_contentforchunkinbatch]metadatas[chunk.metadataforchunkinbatch]collection.add(documentstexts,metadatasmetadatas,idsids)print(f成功添加{len(chunks)}个文档块到向量库)defsearch(self,query:str,top_k:int3):检索最相关的Top-K文档块collectionself.create_or_get_collection()resultscollection.query(query_texts[query],n_resultstop_k)returnresultsChroma的优势它是一个轻量级向量数据库数据持久化在本地文件中不需要单独部署服务。对于企业内部5000人使用的场景Chroma完全够用不需要上Milvus这种重型方案。4.3 RAG链检索 生成# src/rag_chain.pyfromtransformersimportAutoModelForCausalLM,AutoTokenizerimporttorchclassRAGChain:def__init__(self,vector_store_manager,model_path:str):self.vsmvector_store_manager# 加载本地大模型self.tokenizerAutoTokenizer.from_pretrained(model_path,trust_remote_codeTrue)self.modelAutoModelForCausalLM.from_pretrained(model_path,torch_dtypetorch.float16,device_mapauto,trust_remote_codeTrue)def_build_prompt(self,query:str,contexts:List[str])-str:构建带上下文的Promptcontext_text\n\n.join([f【参考资料{i1}】{ctx}fori,ctxinenumerate(contexts)])promptf你是一个企业内部知识助手。请根据以下参考资料回答用户问题。 如果参考资料中没有相关信息请明确告知用户当前知识库中暂未找到相关信息。 【参考资料】{context_text}【用户问题】{query}【回答】returnpromptdefquery(self,query:str,top_k:int3)-dict:完整查询流程检索 → 生成# 1. 检索相关文档search_resultsself.vsm.search(query,top_ktop_k)contextssearch_results[documents][0]ifsearch_results[documents]else[]sourcessearch_results[metadatas][0]ifsearch_results[metadatas]else[]ifnotcontexts:return{answer:知识库中暂未找到与您问题相关的信息请联系管理员更新文档。,sources:[]}# 2. 构建Promptpromptself._build_prompt(query,contexts)# 3. 调用大模型生成inputsself.tokenizer(prompt,return_tensorspt,truncationTrue,max_length4096)withtorch.no_grad():outputsself.model.generate(**inputs,max_new_tokens512,temperature0.7,top_p0.9,do_sampleTrue,)answerself.tokenizer.decode(outputs[0][inputs[input_ids].shape[1]:],skip_special_tokensTrue)# 4. 提取来源文件名source_nameslist(set([s.get(source,未知来源)forsinsources]))return{answer:answer,sources:source_names,contexts:contexts# 可选用于调试}4.4 FastAPI服务接口# api/main.pyfromfastapiimportFastAPI,HTTPExceptionfrompydanticimportBaseModelfromtypingimportOptionalfromsrc.vector_storeimportVectorStoreManagerfromsrc.rag_chainimportRAGChain appFastAPI(title企业知识问答智能体)# 全局初始化服务启动时加载vsmVectorStoreManager()ragRAGChain(vsm,model_path/path/to/Qwen2.5-7B)classQueryRequest(BaseModel):question:strtop_k:Optional[int]3classQueryResponse(BaseModel):answer:strsources:listsuccess:boolapp.post(/api/query,response_modelQueryResponse)asyncdefquery(request:QueryRequest):知识问答接口try:resultrag.query(request.question,top_krequest.top_k)returnQueryResponse(answerresult[answer],sourcesresult[sources],successTrue)exceptExceptionase:raiseHTTPException(status_code500,detailstr(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企业智能知识助手,layoutwide)st.title( 企业知识问答智能体)# 初始化对话历史ifmessagesnotinst.session_state:st.session_state.messages[]# 显示历史对话formsginst.session_state.messages:withst.chat_message(msg[role]):st.write(msg[content])ifsourcesinmsgandmsg[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:resprequests.post(http://localhost:8000/api/query,json{question:prompt,top_k:3})ifresp.status_code200:dataresp.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(正在加载文档...)docsload_documents(./data/raw)print(f加载了{len(docs)}个文档)# 2. 分块print(正在切分文档...)chunkssplit_documents(docs,chunk_size500,chunk_overlap50)print(f切分为{len(chunks)}个块)# 3. 存入向量库print(正在存入向量数据库...)vsmVectorStoreManager()vsm.add_documents(chunks)print(导入完成)if__name____main__:main()六、上线与运维别忘了这些事代码写完只是第一步上线之后才是真正的开始。根据我的项目经验这三个点最容易踩坑6.1 模型推理显存优化Qwen2.5-7B在FP16下需要约14GB显存。如果你的服务器显存不够可以启用4-bit量化fromtransformersimportBitsAndBytesConfig quantization_configBitsAndBytesConfig(load_in_4bitTrue,bnb_4bit_compute_dtypetorch.float16)modelAutoModelForCausalLM.from_pretrained(model_path,quantization_configquantization_config,device_mapauto)4-bit量化后显存占用降到约6GB推理速度损失约20%对于企业内部使用完全可以接受。6.2 日志与反馈闭环在api/main.py中添加日志中间件记录每次查询的耗时和Token消耗。同时用户反馈的点赞/点踩数据需要定期复盘-- 查询点踩最多的类型SELECTSUBSTR(question,1,20)asquestion_sample,COUNT(*)asnegative_countFROMfeedbackWHEREfeedback_typenegativeGROUPBYSUBSTR(question,1,20)ORDERBYnegative_countDESCLIMIT10;这些数据会告诉你哪些问题是模型经常答错的是知识库缺失还是Prompt需要优化反馈闭环是AI应用持续迭代的核心驱动力。6.3 知识库更新机制文档更新后管理员需要重新导入数据。建议设计增量更新策略# 增量更新只删除指定文档的旧向量再插入新向量defupdate_document(file_name:str,new_chunks:List):collectionvsm.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优化是否有效。有问题欢迎在评论区交流也欢迎关注后续更新
从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覆盖主流文档格式文本分块RecursiveCharacterTextSplitterLangChain内置支持按段落/句子递归切分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_pathos.path.join(directory_path,file)iffile.endswith(.pdf):loaderPyPDFLoader(file_path)eliffile.endswith(.docx):loaderDocx2txtLoader(file_path)eliffile.endswith(.txt)orfile.endswith(.md):loaderTextLoader(file_path,encodingutf-8)else:continuedocsloader.load()# 给每个文档块打上来源标签fordocindocs:doc.metadata[source]filedocuments.extend(docs)returndocuments# src/text_splitter.pyfromlangchain.text_splitterimportRecursiveCharacterTextSplitterdefsplit_documents(documents:List,chunk_size:int500,chunk_overlap:int50): 递归切分文档 chunk_size: 每块最大字符数 chunk_overlap: 块间重叠字符数保证上下文连贯 text_splitterRecursiveCharacterTextSplitter(chunk_sizechunk_size,chunk_overlapchunk_overlap,separators[\n\n,\n,。,,,,, ,],length_functionlen,)chunkstext_splitter.split_documents(documents)returnchunks为什么选择递归切分因为它会优先按段落\n\n切分再按句子。切分最后按字符切分。这样能最大程度保证语义完整性——你不会把一个句子的前半部分放在块A、后半部分放在块B。4.2 Embedding与向量存储# src/embedding.pyfromsentence_transformersimportSentenceTransformerdefload_embedding_model(model_name:strBAAI/bge-m3):加载本地Embedding模型modelSentenceTransformer(model_name,devicecuda)# 有GPU就用cudareturnmodel# src/vector_store.pyimportchromadbfromchromadb.utilsimportembedding_functionsclassVectorStoreManager:def__init__(self,persist_directory:str./chroma_db):self.clientchromadb.PersistentClient(pathpersist_directory)self.collection_nameknowledge_base# 使用BGE-M3的Embedding函数self.embedding_fnembedding_functions.SentenceTransformerEmbeddingFunction(model_nameBAAI/bge-m3)defcreate_or_get_collection(self):创建或获取已有集合returnself.client.get_or_create_collection(nameself.collection_name,embedding_functionself.embedding_fn)defadd_documents(self,chunks:List,batch_size:int100):批量添加文档块到向量库collectionself.create_or_get_collection()foriinrange(0,len(chunks),batch_size):batchchunks[i:ibatch_size]ids[fdoc_{i}_{j}forjinrange(len(batch))]texts[chunk.page_contentforchunkinbatch]metadatas[chunk.metadataforchunkinbatch]collection.add(documentstexts,metadatasmetadatas,idsids)print(f成功添加{len(chunks)}个文档块到向量库)defsearch(self,query:str,top_k:int3):检索最相关的Top-K文档块collectionself.create_or_get_collection()resultscollection.query(query_texts[query],n_resultstop_k)returnresultsChroma的优势它是一个轻量级向量数据库数据持久化在本地文件中不需要单独部署服务。对于企业内部5000人使用的场景Chroma完全够用不需要上Milvus这种重型方案。4.3 RAG链检索 生成# src/rag_chain.pyfromtransformersimportAutoModelForCausalLM,AutoTokenizerimporttorchclassRAGChain:def__init__(self,vector_store_manager,model_path:str):self.vsmvector_store_manager# 加载本地大模型self.tokenizerAutoTokenizer.from_pretrained(model_path,trust_remote_codeTrue)self.modelAutoModelForCausalLM.from_pretrained(model_path,torch_dtypetorch.float16,device_mapauto,trust_remote_codeTrue)def_build_prompt(self,query:str,contexts:List[str])-str:构建带上下文的Promptcontext_text\n\n.join([f【参考资料{i1}】{ctx}fori,ctxinenumerate(contexts)])promptf你是一个企业内部知识助手。请根据以下参考资料回答用户问题。 如果参考资料中没有相关信息请明确告知用户当前知识库中暂未找到相关信息。 【参考资料】{context_text}【用户问题】{query}【回答】returnpromptdefquery(self,query:str,top_k:int3)-dict:完整查询流程检索 → 生成# 1. 检索相关文档search_resultsself.vsm.search(query,top_ktop_k)contextssearch_results[documents][0]ifsearch_results[documents]else[]sourcessearch_results[metadatas][0]ifsearch_results[metadatas]else[]ifnotcontexts:return{answer:知识库中暂未找到与您问题相关的信息请联系管理员更新文档。,sources:[]}# 2. 构建Promptpromptself._build_prompt(query,contexts)# 3. 调用大模型生成inputsself.tokenizer(prompt,return_tensorspt,truncationTrue,max_length4096)withtorch.no_grad():outputsself.model.generate(**inputs,max_new_tokens512,temperature0.7,top_p0.9,do_sampleTrue,)answerself.tokenizer.decode(outputs[0][inputs[input_ids].shape[1]:],skip_special_tokensTrue)# 4. 提取来源文件名source_nameslist(set([s.get(source,未知来源)forsinsources]))return{answer:answer,sources:source_names,contexts:contexts# 可选用于调试}4.4 FastAPI服务接口# api/main.pyfromfastapiimportFastAPI,HTTPExceptionfrompydanticimportBaseModelfromtypingimportOptionalfromsrc.vector_storeimportVectorStoreManagerfromsrc.rag_chainimportRAGChain appFastAPI(title企业知识问答智能体)# 全局初始化服务启动时加载vsmVectorStoreManager()ragRAGChain(vsm,model_path/path/to/Qwen2.5-7B)classQueryRequest(BaseModel):question:strtop_k:Optional[int]3classQueryResponse(BaseModel):answer:strsources:listsuccess:boolapp.post(/api/query,response_modelQueryResponse)asyncdefquery(request:QueryRequest):知识问答接口try:resultrag.query(request.question,top_krequest.top_k)returnQueryResponse(answerresult[answer],sourcesresult[sources],successTrue)exceptExceptionase:raiseHTTPException(status_code500,detailstr(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企业智能知识助手,layoutwide)st.title( 企业知识问答智能体)# 初始化对话历史ifmessagesnotinst.session_state:st.session_state.messages[]# 显示历史对话formsginst.session_state.messages:withst.chat_message(msg[role]):st.write(msg[content])ifsourcesinmsgandmsg[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:resprequests.post(http://localhost:8000/api/query,json{question:prompt,top_k:3})ifresp.status_code200:dataresp.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(正在加载文档...)docsload_documents(./data/raw)print(f加载了{len(docs)}个文档)# 2. 分块print(正在切分文档...)chunkssplit_documents(docs,chunk_size500,chunk_overlap50)print(f切分为{len(chunks)}个块)# 3. 存入向量库print(正在存入向量数据库...)vsmVectorStoreManager()vsm.add_documents(chunks)print(导入完成)if__name____main__:main()六、上线与运维别忘了这些事代码写完只是第一步上线之后才是真正的开始。根据我的项目经验这三个点最容易踩坑6.1 模型推理显存优化Qwen2.5-7B在FP16下需要约14GB显存。如果你的服务器显存不够可以启用4-bit量化fromtransformersimportBitsAndBytesConfig quantization_configBitsAndBytesConfig(load_in_4bitTrue,bnb_4bit_compute_dtypetorch.float16)modelAutoModelForCausalLM.from_pretrained(model_path,quantization_configquantization_config,device_mapauto)4-bit量化后显存占用降到约6GB推理速度损失约20%对于企业内部使用完全可以接受。6.2 日志与反馈闭环在api/main.py中添加日志中间件记录每次查询的耗时和Token消耗。同时用户反馈的点赞/点踩数据需要定期复盘-- 查询点踩最多的类型SELECTSUBSTR(question,1,20)asquestion_sample,COUNT(*)asnegative_countFROMfeedbackWHEREfeedback_typenegativeGROUPBYSUBSTR(question,1,20)ORDERBYnegative_countDESCLIMIT10;这些数据会告诉你哪些问题是模型经常答错的是知识库缺失还是Prompt需要优化反馈闭环是AI应用持续迭代的核心驱动力。6.3 知识库更新机制文档更新后管理员需要重新导入数据。建议设计增量更新策略# 增量更新只删除指定文档的旧向量再插入新向量defupdate_document(file_name:str,new_chunks:List):collectionvsm.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优化是否有效。有问题欢迎在评论区交流也欢迎关注后续更新
![[C++]内存管理：串顺序存储的内存回收](http://pic.xiahunao.cn/yaotu/[C++]内存管理：串顺序存储的内存回收)
