AI大模型应用开发实战：从提示词工程到RAG与Agent系统构建

最近在AI大模型应用开发中，经常遇到这样的困惑：学了很多零散知识点，但实际项目落地时却无从下手。特别是大模型、提示词、Agent、RAG、LangChain这些热门技术，网上资料虽多但缺乏系统性整合。本文将基于700集完整教程的核心内容，手把手带你构建完整的AI大模型知识体系，从零基础到项目实战全覆盖。

无论你是刚接触AI的初学者，还是希望系统提升的开发者，本文都将提供可落地的实操方案。我们将从基础概念讲起，逐步深入到企业级项目实战，每个环节都配有完整代码示例和避坑指南。

1. AI大模型技术全景解析

1.1 什么是AI大模型及其发展现状

AI大模型是指参数量达到亿级甚至千亿级的大型预训练语言模型。这类模型通过海量数据训练，具备了强大的语言理解和生成能力。当前全球AI大模型十强中，既有国际巨头产品，也有国内优秀模型如通义千问等。

大模型的核心价值在于其通用性——同一个基础模型经过微调后，可以应用于对话、写作、编程、分析等多种场景。这大大降低了AI应用开发的门槛，开发者无需从零开始训练模型，只需掌握如何有效利用现有大模型。

1.2 大模型应用开发技术栈组成

完整的大模型应用开发涉及多个技术层面：

• 基础大模型：提供核心的AI能力，如GPT系列、通义千问、文心一言等
• 提示词工程：如何设计有效的输入指令，让大模型输出期望的结果
• RAG框架：检索增强生成技术，扩展模型的知识边界
• Agent智能体：让大模型具备规划和执行复杂任务的能力
• LangChain：大模型应用开发框架，提供标准化组件和工具链

这些技术不是孤立的，而是相互配合构成完整解决方案。比如一个智能客服系统可能同时用到提示词优化、RAG知识库、多Agent协作等技术。

1.3 学习路线规划建议

对于零基础学习者，建议按以下顺序循序渐进：

1. 第一阶段：掌握大模型基础API调用和提示词设计
2. 第二阶段：学习RAG技术构建外部知识库
3. 第三阶段：开发Agent实现复杂任务自动化
4. 第四阶段：使用LangChain框架整合完整应用
5. 第五阶段：模型微调定制化满足特定需求

每个阶段都需要理论学习和实践项目相结合，下面我们将逐一深入各个技术模块。

2. 环境准备与基础工具配置

2.1 开发环境搭建

首先需要准备Python开发环境，建议使用Python 3.8+版本：

# 检查Python版本 python --version # 安装必要的包管理工具 pip install --upgrade pip

创建独立的虚拟环境以避免依赖冲突：

# 创建虚拟环境 python -m venv ai_tutorial # 激活虚拟环境 # Windows: ai_tutorial\Scripts\activate # Linux/Mac: source ai_tutorial/bin/activate

2.2 核心依赖库安装

大模型开发需要安装以下关键库：

# 基础AI开发库 pip install openai langchain langchain-community # 向量数据库和检索相关 pip install faiss-cpu chromadb # 文档处理工具 pip install pypdf2 python-docx beautifulsoup4 # 可选：国内大模型SDK pip install dashscope qianfan

2.3 API密钥配置

使用大模型服务需要配置相应的API密钥：

# config.py - 配置文件示例 import os # OpenAI配置（国际模型） os.environ["OPENAI_API_KEY"] = "your-openai-key" # 国内模型配置（如通义千问） os.environ["DASHSCOPE_API_KEY"] = "your-dashscope-key" # 其他服务配置 os.environ["SERPAPI_API_KEY"] = "your-serpapi-key" # 搜索API

重要提示：在实际项目中，不要将API密钥硬编码在代码中，应该使用环境变量或专业的密钥管理服务。

3. 提示词工程深度解析

3.1 提示词设计基本原则

提示词工程是大模型应用的基础，好的提示词能显著提升模型输出质量。核心原则包括：

• 明确性：清晰表达任务要求和期望输出格式
• 上下文充足：提供足够的背景信息帮助模型理解
• 结构化：使用标记、示例等结构化元素引导模型
• 迭代优化：基于测试结果持续改进提示词

3.2 基础提示词模式实战

下面通过具体示例展示不同场景的提示词设计：

# 基础分类任务提示词 classification_prompt = """ 请对以下文本进行情感分类（正面/负面/中性）： 文本：{text} 请以JSON格式返回结果： {{ "sentiment": "分类结果", "confidence": "置信度", "reason": "分类理由" }} """ # 代码生成提示词 coding_prompt = """ 你是一个资深的Python开发工程师。请编写一个函数来解决以下问题： 问题：{problem_description} 要求： 1. 函数名：{function_name} 2. 输入参数：{input_params} 3. 返回类型：{return_type} 4. 添加适当的注释和异常处理 请只返回Python代码，不要额外解释。 """

3.3 高级提示词技巧

3.3.1 思维链提示（Chain-of-Thought）

通过让模型展示推理过程，提升复杂问题的解决能力：

cot_prompt = """ 请逐步推理解决以下数学问题： 问题：如果一个篮子里有5个苹果，小明拿走了2个，然后又放进去3个，现在篮子里有多少个苹果？ 请按步骤思考： 1. 首先，篮子里最初有5个苹果 2. 然后，小明拿走了2个，剩下5-2=3个 3. 接着，放进去3个，现在有3+3=6个 4. 所以，最终答案是6个 现在请解决这个问题：{user_question} """

3.3.2 少样本学习提示（Few-Shot Learning）

提供示例帮助模型理解任务模式：

few_shot_prompt = """ 请将中文翻译成英文，并保持专业术语准确： 示例1： 输入：这个神经网络模型采用了注意力机制 输出：This neural network model employs an attention mechanism 示例2： 输入：我们需要优化损失函数来提升模型性能 输出：We need to optimize the loss function to improve model performance 现在请翻译： 输入：{user_input} 输出： """

4. RAG技术实战详解

4.1 RAG核心原理与架构

RAG（检索增强生成）通过结合检索器和生成器，让大模型能够访问外部知识库。其工作流程包括：

1. 文档处理：将原始文档分割成小块（chunk）
2. 向量化：将文本转换为向量表示
3. 检索：根据查询找到最相关的文档片段
4. 生成：将检索结果作为上下文，生成最终答案

4.2 本地PDF文档接入实战

下面演示如何将本地PDF文件接入RAG系统：

# rag_pdf_processor.py from langchain.document_loaders import PyPDFLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.embeddings import OpenAIEmbeddings from langchain.vectorstores import Chroma class PDFRAGSystem: def __init__(self, pdf_path, chunk_size=1000, chunk_overlap=200): self.pdf_path = pdf_path self.chunk_size = chunk_size self.chunk_overlap = chunk_overlap self.vectorstore = None def process_document(self): """处理PDF文档并构建向量数据库""" # 1. 加载PDF文档 loader = PyPDFLoader(self.pdf_path) documents = loader.load() # 2. 文档分割 text_splitter = RecursiveCharacterTextSplitter( chunk_size=self.chunk_size, chunk_overlap=self.chunk_overlap ) chunks = text_splitter.split_documents(documents) # 3. 创建向量存储 embeddings = OpenAIEmbeddings() self.vectorstore = Chroma.from_documents( documents=chunks, embedding=embeddings ) return len(chunks) def query(self, question, k=3): """查询RAG系统""" if not self.vectorstore: raise ValueError("请先调用process_document方法处理文档") # 检索相关文档 docs = self.vectorstore.similarity_search(question, k=k) # 构建提示词 context = "\n\n".join([doc.page_content for doc in docs]) prompt = f"""基于以下上下文信息回答问题： 上下文： {context} 问题：{question} 要求：如果上下文信息不足以回答问题，请明确说明。""" return prompt, docs # 使用示例 if __name__ == "__main__": rag_system = PDFRAGSystem("technical_manual.pdf") chunk_count = rag_system.process_document() print(f"文档处理完成，共分割为{chunk_count}个文本块") question = "什么是注意力机制？" prompt, relevant_docs = rag_system.query(question) print("生成的提示词：", prompt)

4.3 RAG系统优化策略

4.3.1 文本分块优化

分块策略直接影响检索效果，需要根据文档类型调整：

# 优化后的文本分割器 def create_optimized_splitter(doc_type): if doc_type == "technical": # 技术文档：按章节和段落分割 return RecursiveCharacterTextSplitter( chunk_size=800, chunk_overlap=100, separators=["\n## ", "\n### ", "\n\n", "\n", " "] ) elif doc_type == "legal": # 法律文档：保持条款完整性 return RecursiveCharacterTextSplitter( chunk_size=1200, chunk_overlap=50, separators=["\n条款", "\n第", "\n\n", "\n", " "] ) else: # 通用文档 return RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=200 )

4.3.2 标题信息嵌入策略

在RAG系统中，是否嵌入标题信息取决于具体需求：

# 包含标题信息的文档处理 class TitleAwareProcessor: def process_with_titles(self, documents): processed_chunks = [] for doc in documents: # 提取标题信息 title = self.extract_title(doc.metadata) # 将标题信息融入内容 if title: enhanced_content = f"标题：{title}\n内容：{doc.page_content}" else: enhanced_content = doc.page_content processed_chunks.append(enhanced_content) return processed_chunks def extract_title(self, metadata): # 从元数据中提取标题 return metadata.get('title', '') or metadata.get('heading', '')

一般来说，技术文档和结构化内容建议保留标题信息，这有助于提升检索准确性。但对于创意写作等场景，可能不需要特别强调标题。

5. Agent智能体开发实战

5.1 Agent核心概念与架构

Agent是能够感知环境、制定计划并执行动作的智能系统。在大模型语境下，Agent通常包含以下组件：

• 规划器：分解复杂任务为可执行步骤
• 工具集：Agent可以调用的外部工具和API
• 记忆模块：保存对话历史和任务上下文
• 执行器：协调各个组件完成目标任务

5.2 基础Agent开发示例

下面创建一个能够使用搜索工具和计算工具的简单Agent：

# simple_agent.py from langchain.agents import AgentType, initialize_agent from langchain.tools import Tool from langchain.llms import OpenAI import math class CalculatorTool: """自定义计算工具""" def calculate(self, expression): try: # 安全评估数学表达式 allowed_names = {**math.__dict__} result = eval(expression, {"__builtins__": {}}, allowed_names) return f"计算结果：{result}" except Exception as e: return f"计算错误：{str(e)}" def create_agent(): # 初始化大模型 llm = OpenAI(temperature=0) # 创建工具集 calculator = CalculatorTool() tools = [ Tool( name="Calculator", func=calculator.calculate, description="用于数学计算，输入数学表达式如'3 + 5 * 2'" ), # 可以添加更多工具如搜索工具、API调用工具等 ] # 创建Agent agent = initialize_agent( tools=tools, llm=llm, agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION, verbose=True ) return agent # 使用示例 if __name__ == "__main__": agent = create_agent() # 测试复杂任务 result = agent.run("请计算圆的面积，半径为5厘米，然后告诉我结果是多少平方毫米") print("Agent执行结果：", result)

5.3 多Agent协作系统

对于复杂任务，可以构建多个专业Agent协作的系统：

# multi_agent_system.py from typing import List, Dict from langchain.schema import BaseMessage, HumanMessage class SpecialistAgent: def __init__(self, role, expertise, llm): self.role = role self.expertise = expertise self.llm = llm def analyze(self, problem): prompt = f""" 你是一个{self.role}，擅长{self.expertise}。 请从你的专业角度分析以下问题： 问题：{problem} 请提供专业的分析建议：""" response = self.llm(prompt) return response class CoordinatorAgent: def __init__(self, specialist_agents: List[SpecialistAgent], llm): self.specialists = specialist_agents self.llm = llm def coordinate_task(self, complex_task): # 收集各专家的分析 analyses = [] for specialist in self.specialists: analysis = specialist.analyze(complex_task) analyses.append({ 'role': specialist.role, 'analysis': analysis }) # 综合各专家意见 summary_prompt = f""" 需要解决复杂任务：{complex_task} 各专家意见： {chr(10).join([f"{item['role']}：{item['analysis']}" for item in analyses])} 请综合以上意见，给出完整的解决方案：""" final_solution = self.llm(summary_prompt) return { 'specialist_analyses': analyses, 'final_solution': final_solution } # 使用示例 def create_software_project_team(): llm = OpenAI(temperature=0.7) # 创建专业Agent团队 backend_agent = SpecialistAgent("后端工程师", "系统架构和API设计", llm) frontend_agent = SpecialistAgent("前端工程师", "用户界面和交互设计", llm) dba_agent = SpecialistAgent("数据库管理员", "数据存储和优化", llm) coordinator = CoordinatorAgent([backend_agent, frontend_agent, dba_agent], llm) return coordinator # 测试复杂任务协调 coordinator = create_software_project_team() task = "设计一个在线文档协作系统，支持多人实时编辑和版本管理" result = coordinator.coordinate_task(task) print("多Agent协作结果：", result)

6. LangChain框架深度应用

6.1 LangChain核心组件详解

LangChain为大模型应用开发提供了标准化组件：

• Models：各种大模型的统一接口
• Prompts：提示词管理和模板化
• Chains：任务链式执行
• Agents：智能体工具调用
• Memory：对话状态管理
• Indexes：文档检索和向量化

6.2 完整应用开发实战

下面演示如何使用LangChain构建一个完整的文档问答系统：

# document_qa_system.py from langchain.chains import RetrievalQA from langchain.llms import OpenAI from langchain.embeddings import OpenAIEmbeddings from langchain.vectorstores import Chroma from langchain.text_splitter import CharacterTextSplitter from langchain.document_loaders import TextLoader from langchain.prompts import PromptTemplate class DocumentQASystem: def __init__(self, document_path): self.document_path = document_path self.qa_chain = None self.vectorstore = None def initialize_system(self): """初始化文档问答系统""" # 1. 加载文档 loader = TextLoader(self.document_path) documents = loader.load() # 2. 分割文档 text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0) texts = text_splitter.split_documents(documents) # 3. 创建向量存储 embeddings = OpenAIEmbeddings() self.vectorstore = Chroma.from_documents(texts, embeddings) # 4. 创建自定义提示词模板 custom_prompt = PromptTemplate( template="""基于以下上下文信息，请以专业、准确的方式回答问题。如果无法从上下文中找到答案，请说明信息不足。 上下文：{context} 问题：{question} 请提供详细、准确的回答：""", input_variables=["context", "question"] ) # 5. 创建检索式QA链 self.qa_chain = RetrievalQA.from_chain_type( llm=OpenAI(), chain_type="stuff", retriever=self.vectorstore.as_retriever(), chain_type_kwargs={"prompt": custom_prompt} ) def ask_question(self, question): """提问并获取答案""" if not self.qa_chain: self.initialize_system() result = self.qa_chain.run(question) return result def get_similar_documents(self, query, k=3): """获取相似文档片段""" if not self.vectorstore: self.initialize_system() docs = self.vectorstore.similarity_search(query, k=k) return docs # 使用示例 if __name__ == "__main__": # 初始化系统 qa_system = DocumentQASystem("knowledge_base.txt") # 提问测试 questions = [ "本文档的主要内容包括什么？", "如何配置开发环境？", "有哪些常见问题需要注意？" ] for question in questions: answer = qa_system.ask_question(question) print(f"问题：{question}") print(f"回答：{answer}") print("-" * 50)

6.3 LangChain高级特性

6.3.1 记忆管理实现多轮对话

# conversation_with_memory.py from langchain.memory import ConversationBufferMemory from langchain.chains import ConversationChain def create_conversational_agent(): # 创建带记忆的对话链 memory = ConversationBufferMemory() llm = OpenAI(temperature=0.7) conversation = ConversationChain( llm=llm, memory=memory, verbose=True ) return conversation # 测试多轮对话 agent = create_conversational_agent() # 第一轮对话 response1 = agent.predict(input="我叫张三，是一名软件工程师") print("AI:", response1) # 第二轮对话（能记住上下文） response2 = agent.predict(input="我刚才说我叫什么名字？") print("AI:", response2)

6.3.2 复杂工作流设计

# complex_workflow.py from langchain.chains import SimpleSequentialChain, TransformChain def create_data_processing_workflow(): """创建复杂的数据处理工作流""" # 定义数据转换链 def transform_function(inputs): text = inputs["text"] # 执行数据清洗和转换 cleaned_text = text.lower().strip() return {"cleaned_text": cleaned_text} transform_chain = TransformChain( input_variables=["text"], output_variables=["cleaned_text"], transform=transform_function ) # 定义分析链 def analysis_function(inputs): text = inputs["cleaned_text"] # 这里可以接入大模型进行分析 analysis = f"分析结果：文本长度为{len(text)}字符" return {"analysis_result": analysis} analysis_chain = TransformChain( input_variables=["cleaned_text"], output_variables=["analysis_result"], transform=analysis_function ) # 组合成顺序链 overall_chain = SimpleSequentialChain( chains=[transform_chain, analysis_chain], verbose=True ) return overall_chain

7. 大模型微调技术实战

7.1 微调的基本概念与适用场景

微调是指在预训练大模型的基础上，使用特定领域的数据继续训练，使模型适应特定任务。主要适用场景包括：

• 领域适应：让通用模型掌握专业领域知识
• 风格迁移：调整模型的输出风格和语气
• 任务优化：提升模型在特定任务上的性能
• 参数效率：相比从头训练，微调成本低效果好

7.2 数据准备与预处理

微调成功的关键在于高质量的数据准备：

# data_preparation.py import json from typing import List, Dict class FineTuningDataPreparer: def __init__(self): self.training_data = [] def prepare_conversation_data(self, conversations: List[Dict]): """准备对话式微调数据""" formatted_data = [] for conv in conversations: # 转换对话格式 messages = [] for turn in conv["turns"]: messages.append({ "role": turn["role"], "content": turn["content"] }) formatted_data.append({ "messages": messages }) return formatted_data def prepare_instruction_data(self, instructions: List[Dict]): """准备指令遵循微调数据""" formatted_data = [] for item in instructions: formatted_data.append({ "instruction": item["instruction"], "input": item.get("input", ""), "output": item["output"] }) return formatted_data def save_training_file(self, data, filename): """保存为训练文件""" with open(filename, 'w', encoding='utf-8') as f: for item in data: f.write(json.dumps(item, ensure_ascii=False) + '\n') print(f"训练数据已保存到：{filename}，共{len(data)}条样本") # 使用示例 preparer = FineTuningDataPreparer() # 示例对话数据 conversations = [ { "turns": [ {"role": "user", "content": "如何学习Python编程？"}, {"role": "assistant", "content": "学习Python可以从基础语法开始..."} ] } ] formatted_data = preparer.prepare_conversation_data(conversations) preparer.save_training_file(formatted_data, "training_data.jsonl")

7.3 微调实战流程

下面以OpenAI API为例展示微调流程：

# fine_tuning_client.py import openai import time from typing import Optional class FineTuningManager: def __init__(self, api_key): openai.api_key = api_key def create_fine_tuning_job(self, training_file_id: str, model: str = "gpt-3.5-turbo"): """创建微调任务""" try: response = openai.FineTuningJob.create( training_file=training_file_id, model=model ) return response except Exception as e: print(f"创建微调任务失败：{e}") return None def check_job_status(self, job_id: str) -> Optional[dict]: """检查微调任务状态""" try: response = openai.FineTuningJob.retrieve(job_id) return response except Exception as e: print(f"检查任务状态失败：{e}") return None def wait_for_completion(self, job_id: str, check_interval: int = 60): """等待微调任务完成""" while True: status = self.check_job_status(job_id) if status: print(f"任务状态：{status.status}") if status.status == "succeeded": print("微调任务完成！") return status elif status.status in ["failed", "cancelled"]: print(f"任务失败或取消：{status}") return status time.sleep(check_interval) # 使用示例（需要先上传训练文件） def demo_fine_tuning_process(): manager = FineTuningManager("your-api-key") # 假设已经上传了训练文件并获得了文件ID training_file_id = "file-abc123" # 创建微调任务 job = manager.create_fine_tuning_job(training_file_id) if job: print(f"微调任务已创建，ID：{job.id}") # 等待任务完成 result = manager.wait_for_completion(job.id) if result and result.status == "succeeded": print(f"微调后的模型：{result.fine_tuned_model}")

8. 项目实战：构建智能技术文档助手

8.1 项目需求分析

我们将构建一个完整的智能技术文档助手，具备以下功能：

• 支持多种格式文档上传（PDF、TXT、DOCX）
• 基于RAG的智能问答系统
• 多轮对话记忆功能
• 文档内容摘要和关键点提取
• 相关文档推荐

8.2 系统架构设计

# smart_doc_assistant.py import os from typing import List, Dict, Optional from langchain.chains import RetrievalQA, ConversationChain from langchain.memory import ConversationBufferWindowMemory from langchain.embeddings import OpenAIEmbeddings from langchain.vectorstores import Chroma from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.document_loaders import PyPDFLoader, TextLoader, UnstructuredWordDocumentLoader class SmartDocumentAssistant: def __init__(self, persist_directory: str = "./chroma_db"): self.persist_directory = persist_directory self.vectorstore = None self.qa_chain = None self.conversation_chain = None self.memory = ConversationBufferWindowMemory(k=5) def load_documents(self, file_paths: List[str]): """加载多种格式的文档""" all_documents = [] for file_path in file_paths: if file_path.endswith('.pdf'): loader = PyPDFLoader(file_path) elif file_path.endswith('.txt'): loader = TextLoader(file_path) elif file_path.endswith('.docx'): loader = UnstructuredWordDocumentLoader(file_path) else: print(f"不支持的文件格式：{file_path}") continue documents = loader.load() all_documents.extend(documents) print(f"已加载文档：{file_path}，共{len(documents)}页") return all_documents def process_documents(self, documents): """处理文档并构建向量数据库""" # 文本分割 text_splitter = RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=200 ) chunks = text_splitter.split_documents(documents) # 创建向量存储 embeddings = OpenAIEmbeddings() self.vectorstore = Chroma.from_documents( documents=chunks, embedding=embeddings, persist_directory=self.persist_directory ) print(f"文档处理完成，共创建{len(chunks)}个文本块") return len(chunks) def initialize_qa_system(self): """初始化问答系统""" if not self.vectorstore: raise ValueError("请先加载和处理文档") from langchain.llms import OpenAI from langchain.prompts import PromptTemplate # 自定义提示词模板 qa_template = """你是一个专业的技术文档助手。请基于以下上下文信息回答问题。 如果上下文信息不足，请基于你的知识回答，但需要说明信息来源。 上下文：{context} 问题：{question} 请提供准确、详细的回答：""" QA_PROMPT = PromptTemplate( template=qa_template, input_variables=["context", "question"] ) # 创建检索式QA链 self.qa_chain = RetrievalQA.from_chain_type( llm=OpenAI(temperature=0.3), chain_type="stuff", retriever=self.vectorstore.as_retriever(search_kwargs={"k": 3}), chain_type_kwargs={"prompt": QA_PROMPT}, return_source_documents=True ) # 创建对话链 self.conversation_chain = ConversationChain( llm=OpenAI(temperature=0.7), memory=self.memory, verbose=False ) def ask_question(self, question: str) -> Dict: """提问并获取答案""" if not self.qa_chain: self.initialize_qa_system() # 使用QA链获取基于文档的答案 qa_result = self.qa_chain({"query": question}) # 使用对话链维护上下文 conversation_context = self.conversation_chain.predict(input=question) return { "answer": qa_result["result"], "source_documents": qa_result.get("source_documents", []), "conversation_context": conversation_context } def summarize_document(self, max_length: int = 500) -> str: """生成文档摘要""" # 获取所有文档内容 all_docs = self.vectorstore.get()["documents"] combined_content = "\n".join(all_docs[:10]) # 取前10个文档块 summary_prompt = f"""请为以下技术文档生成一个简洁的摘要，长度不超过{max_length}字： 文档内容： {combined_content[:3000]} # 限制输入长度 摘要：""" from langchain.llms import OpenAI llm = OpenAI(temperature=0.3) summary = llm(summary_prompt) return summary # 使用示例 def demo_smart_assistant(): assistant = SmartDocumentAssistant() # 加载文档 documents = assistant.load_documents([ "docs/api_manual.pdf", "docs/technical_guide.txt" ]) # 处理文档 assistant.process_documents(documents) # 初始化系统 assistant.initialize_qa_system() # 进行问答 questions = [ "这个系统的API认证如何实现？", "有哪些最佳实践建议？", "如何处理错误情况？" ] for question in questions: result = assistant.ask_question(question) print(f"问题：{question}") print(f"回答：{result['answer']}") print(f"来源文档数：{len(result['source_documents'])}") print("-" * 80) # 生成摘要 summary = assistant.summarize_document() print("文档摘要：", summary) if __name__ == "__main__": demo_smart_assistant()

8.3 系统优化与部署

8.3.1 性能优化策略

# optimization_strategies.py class PerformanceOptimizer: @staticmethod def optimize_retrieval(vectorstore, query, strategy="hybrid"): """优化检索策略""" if strategy == "hybrid": # 混合检索：结合相似度搜索和MMR（最大边际相关性） docs = vectorstore.max_marginal_relevance_search( query, k=3, fetch_k=10 ) elif strategy == "filtered": # 带过滤的检索 docs = vectorstore.similarity_search( query, k=3, filter={"document_type": "technical"} # 假设有元数据过滤 ) else: # 默认相似度搜索 docs = vectorstore.similarity_search(query, k=3) return docs @staticmethod def cache_embeddings(documents, cache_file="embeddings_cache.pkl"): """缓存嵌入向量提升性能""" import pickle import hashlib # 创建文档内容的哈希值作为缓存键 content_hash = hashlib.md5( "".join([doc.page_content for doc in documents]).encode() ).hexdigest() cache_key = f"{content_hash}_{cache_file}" if os.path.exists(cache_key): with open(cache_key, 'rb') as f: return pickle.load(f) else: # 计算嵌入并缓存 embeddings = OpenAIEmbeddings() vectors = embeddings.embed_documents([doc.page_content for doc in documents]) with open(cache_key, 'wb') as f: pickle.dump(vectors, f) return vectors

8.3.2 生产环境部署考虑

# production_config.py class ProductionConfig: """生产环境配置类""" def __init__(self): self.config = { "api_timeout": 30, # API调用超时时间 "max_retries": 3, # 最大重试次数 "rate_limit": 10, # 速率限制（请求/秒） "cache_ttl": 3600, # 缓存生存时间（秒） "log_level": "INFO" # 日志级别 } def setup_monitoring(self): """设置监控和日志""" import logging logging.basicConfig( level=getattr(logging, self.config["log_level"]), format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('app.log'), logging.StreamHandler() ] ) def setup_error_handling(self): """设置错误处理机制""" def graceful_shutdown(signum, frame): print("收到关闭信号，正在保存状态...") # 保存向量数据库状态 # 关闭数据库连接 exit(0) import signal signal.signal(signal.SIGINT, graceful_shutdown) signal.signal(signal.SIGTERM, graceful_shutdown)

9. 常见问题与解决方案

9.1 环境配置问题

问题1：依赖版本冲突

错误现象：ModuleNotFoundError 或版本兼容性错误 解决方案：使用虚拟环境，固定依赖版本

# 生成requirements.txt pip freeze > requirements.txt # 安装指定版本 pip install -