基于RAG与LangChain的AI Agent实战：构建企业级智能问答系统

1. 背景与核心概念AI Agent与工程化实践近期字节跳动CEO梁汝波在内部全员会Allhands上释放了关于AI战略的明确信号其中“豆包”升级为公司级入口以及“AI全面绑定绩效晋升”尤为引人注目。这不仅是单一产品的定位调整更标志着国内一线互联网公司正将AI从“技术探索”推向“全员工程化应用”的新阶段。对于广大开发者而言理解并实践AI工程化特别是AI Agent的开发与应用已成为提升个人竞争力的关键。什么是AI Agent简单来说AI Agent智能体是一个能够感知环境、进行决策并执行行动以实现特定目标的AI系统。它不同于传统的聊天机器人仅完成单轮问答而是具备自主性、规划能力和工具使用能力。例如一个数据分析Agent可以自动连接数据库、执行查询、分析趋势并生成报告全程无需人工逐步指导。为什么AI工程化突然变得如此重要从“豆包”升级和“AI绑定绩效”这两个信号可以看出企业的核心诉求已经从“拥有大模型”转变为“让大模型在业务中高效、可靠地创造价值”。这背后涉及一整套工程实践包括模型部署与优化如何将百亿、千亿参数的大模型低成本、高性能地部署到生产环境。应用开发框架如何快速、标准化地构建基于大模型的应用程序如AI Agent。效能度量与评估如何量化AI应用带来的业务提升并将其与团队和个人的绩效挂钩。对于开发者这意味着单纯调用API已经不够必须掌握从模型微调、服务部署到智能体编排、效果评估的全链路能力。接下来我们将从一个AI Agent的完整开发与部署实战出发拆解其中的关键技术环节。2. 环境准备与版本说明本次实战将构建一个“技术文档智能问答Agent”它能够基于给定的项目文档库自动回答用户的技术问题。我们选择当前主流且开源友好的技术栈。核心环境与工具操作系统Ubuntu 20.04 LTS 或 macOS Monterey (12.x) 及以上。Windows用户建议使用WSL2。Python: 3.9 或 3.10。这是大多数AI框架兼容性最好的版本。大模型服务我们将使用DeepSeek的开源模型并通过Ollama在本地运行以模拟企业私有化部署场景。也可替换为OpenAI GPT、通义千问等API。向量数据库ChromaDB轻量级、易于集成适合快速原型和中小规模知识库。应用开发框架LangChain当前构建AI应用和Agent最流行的框架之一。前端演示可选Gradio快速构建AI应用界面的Python库。版本说明请根据实际情况调整# 使用 conda 或 venv 创建虚拟环境 conda create -n ai-agent python3.10 conda activate ai-agent # 安装核心依赖 pip install langchain0.1.0 pip install langchain-community0.0.10 # 社区集成工具 pip install chromadb0.4.22 pip install sentence-transformers2.2.2 # 用于文本嵌入 pip install pypdf3.17.4 # 用于解析PDF文档 pip install gradio4.19.1 # 用于Web界面 pip install ollama0.1.9 # 用于本地运行大模型 # 安装并运行Ollama服务 (需要先安装Ollama本体请参考官网) # 拉取DeepSeek模型以DeepSeek-Coder为例约16B参数需确保有足够GPU/内存 ollama pull deepseek-coder:6.7b # 可选择更小的版本如1.3b进行测试项目结构预览tech_doc_agent/ ├── data/ # 存放原始技术文档 │ ├── api_documentation.pdf │ └── developer_guide.md ├── vector_store/ # 向量数据库持久化目录 ├── src/ │ ├── __init__.py │ ├── document_loader.py # 文档加载与处理 │ ├── embedding_handler.py # 向量化处理 │ ├── agent_orchestrator.py # Agent编排逻辑 │ └── app.py # 主应用入口Gradio ├── config.yaml # 配置文件 ├── requirements.txt └── README.md3. 核心原理与技术拆解要构建一个实用的问答Agent我们需要理解其背后的关键技术链条检索增强生成RAG。它解决了大模型“幻觉”胡编乱造和知识滞后的问题。3.1 RAGRetrieval-Augmented Generation工作流程索引阶段文档加载与分割将PDF、Markdown、Word等格式的文档加载进来并按语义分割成大小适中的片段如500字符。向量化Embedding使用嵌入模型如text-embedding-ada-002或开源的sentence-transformers模型将每个文本片段转换为一个高维向量。语义相似的文本其向量在空间中的距离也相近。存储将文本片段及其对应的向量存储到向量数据库如ChromaDB中。查询阶段用户提问用户提出一个问题例如“如何配置Spring Security的OAuth2客户端”。检索将用户问题同样转换为向量并在向量数据库中搜索与之最相似的K个文本片段例如最相似的3段。增强提示Prompt Augmentation将检索到的相关文本片段作为“上下文”与用户问题一起组合成一个新的、信息更丰富的提示Prompt提交给大模型。生成答案大模型基于这个包含了准确上下文的提示生成最终答案。由于答案依据来自提供的文档其准确性和可信度大幅提升。3.2 AI Agent的扩展在上述RAG基础上Agent赋予了系统“行动”能力。例如我们的问答Agent可以内嵌以下工具计算器工具当用户问题涉及计算时自动调用。代码执行工具对于代码相关问题可以安全地执行代码片段并返回结果。网络搜索工具需谨慎使用当本地知识库无法回答时可申请搜索最新信息。LangChain框架通过Tool、AgentExecutor等概念将这些能力封装起来让开发者可以像搭积木一样构建复杂的AI工作流。4. 完整实战构建技术文档智能问答Agent4.1 项目初始化与配置首先创建项目并编写配置文件将模型路径、向量库位置等参数化。config.yamlmodel: local_llm: ollama/deepseek-coder:6.7b # Ollama模型名称 # 若使用OpenAI API则配置如下 # api_type: openai # model_name: gpt-3.5-turbo # api_key: ${OPENAI_API_KEY} # 建议从环境变量读取 embedding: model: sentence-transformers/all-MiniLM-L6-v2 # 轻量级且效果不错的开源嵌入模型 vector_store: persist_directory: ./vector_store # 向量数据库持久化路径 collection_name: tech_docs # 集合名称 document: source_directory: ./data # 原始文档目录 chunk_size: 500 # 文本分割大小 chunk_overlap: 50 # 分割重叠部分避免语义断裂requirements.txtlangchain0.1.0 langchain-community0.0.10 chromadb0.4.22 sentence-transformers2.2.2 pypdf3.17.4 gradio4.19.1 ollama0.1.9 pyyaml6.0.14.2 文档加载与向量化存储这是构建知识库的核心步骤。src/document_loader.pyimport os from langchain_community.document_loaders import PyPDFLoader, TextLoader, UnstructuredMarkdownLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.schema import Document from typing import List import yaml def load_config(): with open(config.yaml, r) as f: return yaml.safe_load(f) def load_documents(source_dir: str) - List[Document]: 加载指定目录下的所有支持格式的文档 documents [] for filename in os.listdir(source_dir): file_path os.path.join(source_dir, filename) if filename.endswith(.pdf): loader PyPDFLoader(file_path) elif filename.endswith(.md): loader UnstructuredMarkdownLoader(file_path) elif filename.endswith(.txt): loader TextLoader(file_path) else: print(f跳过不支持的文件格式: {filename}) continue loaded_docs loader.load() documents.extend(loaded_docs) print(f已加载: {filename}, 共 {len(loaded_docs)} 页/段) return documents def split_documents(documents: List[Document], chunk_size: int, chunk_overlap: int) - List[Document]: 将文档分割成小块 text_splitter RecursiveCharacterTextSplitter( chunk_sizechunk_size, chunk_overlapchunk_overlap, length_functionlen, separators[\n\n, \n, 。, , , , , , ] ) return text_splitter.split_documents(documents) if __name__ __main__: config load_config() source_dir config[document][source_directory] chunk_size config[document][chunk_size] chunk_overlap config[document][chunk_overlap] print(开始加载文档...) raw_docs load_documents(source_dir) print(f原始文档加载完成共 {len(raw_docs)} 个文档对象。) print(开始分割文档...) split_docs split_documents(raw_docs, chunk_size, chunk_overlap) print(f文档分割完成共生成 {len(split_docs)} 个文本块。) # 保存分割后的文档供后续使用 # 在实际项目中可以序列化保存这里简单打印示例 for i, doc in enumerate(split_docs[:2]): # 打印前两个块 print(f\n--- 块 {i1} ---) print(doc.page_content[:200] ...)src/embedding_handler.pyfrom langchain_community.embeddings import HuggingFaceEmbeddings from langchain_community.vectorstores import Chroma from langchain.schema import Document from typing import List import yaml def load_config(): with open(config.yaml, r) as f: return yaml.safe_load(f) def create_vector_store(documents: List[Document], persist_directory: str, collection_name: str, embedding_model_name: str): 创建并持久化向量数据库 # 1. 初始化嵌入模型 print(f正在加载嵌入模型: {embedding_model_name}) embeddings HuggingFaceEmbeddings(model_nameembedding_model_name) # 2. 从文档创建向量存储并持久化到磁盘 print(正在生成向量并存入数据库...) vector_store Chroma.from_documents( documentsdocuments, embeddingembeddings, persist_directorypersist_directory, collection_namecollection_name ) # 确保数据写入磁盘 vector_store.persist() print(f向量数据库创建完成已保存至: {persist_directory}) return vector_store def load_existing_vector_store(persist_directory: str, collection_name: str, embedding_model_name: str): 加载已存在的向量数据库 embeddings HuggingFaceEmbeddings(model_nameembedding_model_name) vector_store Chroma( persist_directorypersist_directory, collection_namecollection_name, embedding_functionembeddings ) print(f已加载现有向量数据库集合: {collection_name}) return vector_store if __name__ __main__: # 此部分通常由主流程调用此处演示 config load_config() # 假设我们已经有了分割后的 documents # from document_loader import split_docs # vector_store create_vector_store(split_docs, ...)4.3 构建问答链与Agent现在我们将检索器、大模型和提示模板组合成一条问答链。src/agent_orchestrator.pyfrom langchain.chains import RetrievalQA from langchain_community.llms import Ollama from langchain.prompts import PromptTemplate from langchain_community.vectorstores import Chroma from langchain_community.embeddings import HuggingFaceEmbeddings from langchain.agents import initialize_agent, Tool, AgentType from langchain.tools import Tool as LangchainTool import yaml def load_config(): with open(config.yaml, r) as f: return yaml.safe_load(f) def create_qa_chain(vector_store): 创建基础的检索问答链 config load_config() # 1. 初始化本地大模型通过Ollama llm Ollama(modelconfig[model][local_llm], temperature0.1) # temperature低答案更确定 # 2. 定义自定义提示模板指导模型如何利用上下文 prompt_template 请严格根据以下提供的上下文信息来回答问题。如果上下文中的信息不足以回答问题请直接说“根据提供的资料我无法回答这个问题”不要编造信息。 上下文 {context} 问题{question} 基于上下文的答案 PROMPT PromptTemplate( templateprompt_template, input_variables[context, question] ) # 3. 创建检索问答链 qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, # 将检索到的所有文档“堆叠”后传入 retrievervector_store.as_retriever(search_kwargs{k: 3}), # 检索最相似的3个片段 chain_type_kwargs{prompt: PROMPT}, return_source_documentsTrue # 返回参考来源 ) return qa_chain def create_agent_with_tools(qa_chain): 创建一个具备更多工具如计算的Agent # 定义工具列表 from langchain.utilities import ArxivAPIWrapper, WikipediaAPIWrapper # 注意网络搜索工具在生产环境需严格管控此处仅作示例 # 定义知识库问答工具 def tech_doc_qa(input_text): 用于回答技术文档相关问题的工具。输入是一个问题。 result qa_chain({query: input_text}) return f答案{result[result]}\n\n参考来源{result[source_documents]} tools [ Tool( name技术文档知识库, functech_doc_qa, description当需要回答关于本项目技术文档、API、开发指南的具体问题时使用此工具。输入应是一个清晰的问题。 ), # 可以在此添加更多工具例如 # Tool(nameCalculator, func..., description用于数学计算), ] config load_config() llm Ollama(modelconfig[model][local_llm], temperature0.1) # 初始化Agent agent initialize_agent( toolstools, llmllm, agentAgentType.ZERO_SHOT_REACT_DESCRIPTION, # 一种通用的Agent类型 verboseTrue, # 打印Agent的思考过程便于调试 handle_parsing_errorsTrue # 处理解析错误 ) return agent if __name__ __main__: # 测试代码 config load_config() embeddings HuggingFaceEmbeddings(model_nameconfig[embedding][model]) vector_store Chroma( persist_directoryconfig[vector_store][persist_directory], collection_nameconfig[vector_store][collection_name], embedding_functionembeddings ) print(创建QA链...) qa_chain create_qa_chain(vector_store) test_question 本文档中提到了哪些关于安全的最佳实践 print(f提问: {test_question}) result qa_chain({query: test_question}) print(f答案: {result[result]}) print(--- 参考来源 ---) for doc in result[source_documents]: print(doc.page_content[:150])4.4 集成Web界面并运行使用Gradio快速构建一个用户友好的交互界面。src/app.pyimport gradio as gr from src.agent_orchestrator import create_qa_chain, create_agent_with_tools from src.embedding_handler import load_existing_vector_store import yaml import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) def load_config(): with open(config.yaml, r) as f: return yaml.safe_load(f) # 全局初始化在实际生产应用中应考虑更优雅的启动和资源管理 config load_config() logger.info(正在加载向量数据库...) embeddings HuggingFaceEmbeddings(model_nameconfig[embedding][model]) vector_store load_existing_vector_store( config[vector_store][persist_directory], config[vector_store][collection_name], config[embedding][model] ) logger.info(正在创建QA链...) qa_chain create_qa_chain(vector_store) # 如需使用功能更全的Agent取消下一行注释 # agent create_agent_with_tools(qa_chain) def answer_question(question, history): 处理用户提问的Gradio接口函数 if not question.strip(): return 请输入一个有效的问题。, history try: logger.info(f处理问题: {question}) # 使用基础的QA链 result qa_chain({query: question}) answer result[result] # 格式化历史记录 history.append((question, answer)) logger.info(问题处理完毕。) return , history # 返回空字符串清空输入框并更新历史 except Exception as e: logger.error(f处理问题时发生错误: {e}) return f系统出错: {str(e)}, history # 构建Gradio界面 with gr.Blocks(title技术文档智能问答助手, themegr.themes.Soft()) as demo: gr.Markdown(# 技术文档智能问答助手) gr.Markdown(基于本地知识库的RAG系统请输入关于技术文档的问题。) chatbot gr.Chatbot(label对话历史, height400) msg gr.Textbox(label您的问题, placeholder例如如何配置数据库连接池, lines2) clear gr.Button(清空对话) msg.submit(answer_question, [msg, chatbot], [msg, chatbot]) clear.click(lambda: None, None, chatbot, queueFalse) # 清空聊天记录 if __name__ __main__: logger.info(启动Gradio应用...) demo.launch(server_name0.0.0.0, server_port7860, shareFalse) # shareFalse仅本地访问4.5 运行与验证准备文档将你的技术文档PDF、MD等放入./data目录。构建知识库运行文档处理脚本可编写一个主脚本依次调用document_loader和embedding_handler中的函数生成向量数据库。启动应用在项目根目录下执行python src/app.py。访问界面打开浏览器访问http://localhost:7860。测试提问在界面中输入问题如“XX功能的API参数有哪些”观察系统是否能从文档中检索并生成准确答案。5. 常见问题与排查思路在开发与部署此类AI应用时你会遇到一些典型问题。问题现象可能原因排查思路与解决方案Ollama服务连接失败Ollama服务未启动模型未拉取网络端口冲突。1. 终端执行ollama serve检查服务状态。2. 执行ollama list确认模型已存在。3. 检查config.yaml中模型名称是否与ollama list中的一致。向量检索结果不相关文本分割策略不当嵌入模型不匹配检索参数K值不合适。1. 调整chunk_size和chunk_overlap如改为300/30。2. 尝试不同的嵌入模型如paraphrase-multilingual-MiniLM-L12-v2。3. 调整search_kwargs{k: 5}或尝试相似度分数阈值过滤。大模型回答质量差提示词Prompt设计不佳模型能力有限上下文长度超限。1. 优化prompt_template加入更明确的指令如“请分点列出”。2. 考虑更换或微调模型。3. 确保检索到的上下文总长度未超过模型令牌限制。应用响应速度慢嵌入模型首次加载慢向量检索未使用索引硬件资源不足。1. 嵌入模型加载是瓶颈可考虑使用更轻量模型或缓存嵌入结果。2. 确保ChromaDB使用了合适的索引默认已创建。3. 对于生产环境考虑将向量数据库和模型服务分离部署。答案出现“幻觉”检索到的上下文不包含答案但模型被强制生成。1. 强化提示词如增加“如果上下文没有明确信息请回答不知道”。2. 在最终答案前增加一个“验证”步骤判断检索片段是否真能支撑答案。Gradio界面无法访问防火墙设置端口被占用server_name设置错误。1. 检查demo.launch(server_port7860)端口是否被其他程序占用。2. 本地访问尝试http://127.0.0.1:7860。3. 确保在安全的内网环境如需外网访问需配置SSH隧道或安全组。6. 最佳实践与工程建议将AI Agent从Demo推向生产需要遵循严格的工程规范这也呼应了“AI绑定绩效”中对效能和可靠性的要求。1. 配置与密钥管理绝对禁止在代码中硬编码API Key、数据库密码等敏感信息。必须使用环境变量或专业的密钥管理服务如HashiCorp Vault、AWS Secrets Manager。配置文件如config.yaml应区分开发、测试、生产环境。2. 可观测性与日志记录完整的AI调用链用户输入、检索到的文档片段、发送给模型的完整Prompt、模型原始输出、最终答案。记录每次调用的耗时、令牌使用量、费用如果使用商用API。使用结构化日志如JSON格式便于接入ELK等日志分析系统。3. 性能与成本优化缓存对频繁出现的相同或相似查询缓存最终答案或中间嵌入结果。异步处理对于耗时的文档解析和向量化任务使用异步队列如Celery处理避免阻塞主请求。模型选型在效果和成本间权衡。简单的任务使用小模型如7B复杂任务再调度大模型。分级检索先使用简单的关键词匹配BM25进行粗筛再使用向量检索进行精排提升效率。4. 安全与合规输入输出过滤对用户输入进行严格的敏感词过滤和内容安全检测防止恶意Prompt攻击。对模型输出也需进行合规性审查。权限控制知识库应实现基于角色的访问控制RBAC确保用户只能访问被授权的文档范围。数据隐私涉及企业敏感数据的文档必须使用私有化部署的模型和向量数据库数据不出域。审计追踪记录所有问答记录满足合规审计要求。5. 评估与迭代建立评估集收集一批典型问题及其标准答案定期运行测试监控问答准确率、召回率等指标。A/B测试对比不同提示词、不同模型版本、不同检索策略的效果。反馈闭环提供“答案是否有用”的反馈按钮收集人工反馈数据用于持续优化模型和检索。7. 总结与学习路线通过本次实战我们完整走通了构建一个企业级AI问答Agent的核心路径从环境搭建、文档处理、向量检索到提示工程、链与Agent编排最后完成应用集成。这正是一个AI工程化项目的缩影。回顾关键点RAG是基石它有效结合了外部知识库与大模型生成能力是当前落地最广的AI应用范式。工程化是核心模型本身只是起点如何稳定、高效、安全地集成到现有系统才是创造价值的关键。LangChain是利器它提供了丰富的抽象和组件极大加速了AI应用的开发但深入理解其底层原理才能更好地驾驭和调试。下一步学习方向深入LangChain学习更复杂的Agent类型如Plan-and-Execute、工具调用Function Calling以及LangGraph用于编排工作流。探索模型微调当通用模型在特定领域表现不佳时学习使用LoRA、QLoRA等技术对开源大模型进行轻量级微调。研究高级RAG如查询重写、HyDE假设性文档嵌入、句子窗口检索等进一步提升检索质量。关注云原生部署学习使用Docker容器化你的AI应用并部署到Kubernetes实现弹性伸缩和高可用。建立评估体系学习使用RAGAS、TruLens等框架自动化评估你的AI应用效果。AI技术与绩效晋升的绑定本质上是将技术能力与业务价值直接挂钩。作为开发者你的目标不应仅是完成一个Demo而是交付一个可监控、可迭代、可解释、有商业价值的AI系统。从这个小项目开始不断深入每个环节你就能在AI工程化的浪潮中建立起自己的核心竞争力。