LangChain开发实战零基础入门教程：7天从RAG应用到Ollama本地大模型部署

大家好！作为一名长期关注AI应用开发的技术博主，我发现很多开发者在学习LangChain时面临资料零散、环境配置复杂、实战案例不完整等问题。本文将带你用7天时间系统掌握LangChain核心技能，从基础的RAG应用构建到Ollama本地大模型部署，最后实现Agent智能体开发。无论你是AI初学者还是有经验的开发者，都能通过本教程获得完整的实战能力。

1. LangChain与相关技术核心概念解析

1.1 什么是LangChain及其生态体系

LangChain是一个用于开发大语言模型(LLM)应用程序的框架，它提供了一套标准化的接口和组件，让开发者能够更轻松地构建基于LLM的应用。LangChain的核心价值在于它将复杂的LLM应用开发过程模块化，提供了链(Chains)、代理(Agents)、记忆(Memory)等核心概念。

在实际项目中，LangChain能够帮助开发者解决以下关键问题：

• 统一不同LLM供应商的API调用接口
• 管理对话历史和上下文记忆
• 构建复杂的多步骤推理流程
• 集成外部工具和数据源

与传统的直接调用API相比，LangChain提供了更高层次的抽象，让开发者能够专注于业务逻辑而不是底层实现细节。

1.2 RAG技术原理与应用场景

RAG（Retrieval-Augmented Generation）即检索增强生成，是一种将检索技术与大语言模型生成能力相结合的技术框架。其核心思想是在生成答案之前，先从知识库中检索相关信息，然后将检索到的信息与用户问题一起提供给LLM，从而生成更准确、更有依据的答案。

RAG系统的工作流程通常包括：

1. 文档处理 ：将原始文档进行分块、向量化处理
2. 检索阶段 ：根据用户查询从向量数据库中检索相关文档片段
3. 生成阶段 ：将检索结果与用户问题组合，交给LLM生成最终答案

RAG技术特别适用于以下场景：

• 企业知识库问答系统
• 专业领域文档分析
• 需要实时信息的应用
• 减少LLM幻觉现象

1.3 Ollama本地大模型部署的价值

Ollama是一个简化本地大模型部署和管理的工具，它让开发者能够在个人电脑或服务器上快速部署和运行各种开源大模型。与云端API调用相比，Ollama部署本地模型具有以下优势：

• 数据隐私安全 ：所有数据处理都在本地完成，避免敏感数据外泄
• 成本可控 ：一次部署后使用成本固定，不受API调用次数限制
• 网络要求低 ：不依赖互联网连接，响应速度更快
• 定制灵活 ：可以针对特定需求对模型进行微调

Ollama支持的主流模型包括Llama系列、Mistral、Gemma等，能够满足不同场景下的需求。

1.4 Agent智能体的核心能力

Agent（智能体）是LangChain中的高级概念，它赋予LLM使用工具、执行动作、进行推理的能力。与简单的问答系统不同，Agent能够：

• 自主规划任务执行步骤
• 使用外部工具获取信息
• 进行多轮对话和复杂推理
• 根据环境反馈调整策略

常见的Agent类型包括：

• Zero-shot React Agent ：基于ReAct模式的通用智能体
• Conversational Agent ：专为对话场景优化的智能体
• Self-ask with Search Agent ：能够自主搜索信息的智能体

2. 环境准备与版本兼容性说明

2.1 基础环境要求

在开始LangChain开发之前，需要确保你的开发环境满足以下要求：

操作系统要求：

• Windows 10/11, macOS 10.15+, 或 Ubuntu 18.04+
• 至少8GB内存（推荐16GB以上）
• 至少20GB可用磁盘空间

Python环境：

# 检查Python版本
python --version
# 应该显示Python 3.8或更高版本

# 创建虚拟环境
python -m venv langchain_env
source langchain_env/bin/activate  # Linux/macOS
# 或
langchain_env\Scripts\activate  # Windows

2.2 核心依赖安装与版本匹配

LangChain生态包含多个包，版本兼容性非常重要。以下是经过测试的稳定版本组合：

# 安装核心LangChain包
pip install langchain==0.1.0

# 安装社区贡献的组件
pip install langchain-community==0.0.11

# 安装文本嵌入相关
pip install langchain-text-splitters==0.0.1

# 安装Ollama Python客户端
pip install ollama

# 向量数据库支持（以Chroma为例）
pip install chromadb

# 其他实用工具
pip install python-dotenv requests beautifulsoup4

版本兼容性说明：

• langchain 0.1.0 与 langchain-community 0.0.11 完全兼容
• 避免混合使用不同大版本的包，以免出现接口不匹配
• 如果遇到兼容性问题，可以尝试使用较新的稳定版本

2.3 开发工具配置

推荐使用VS Code作为开发环境，并安装以下扩展：

• Python扩展（提供代码补全和调试功能）
• Jupyter扩展（方便进行实验性代码编写）
• GitLens（版本控制管理）

项目结构建议：

langchain-project/
├── src/
│   ├── agents/          # Agent相关代码
│   ├── chains/          # 链定义
│   ├── tools/           # 自定义工具
│   └── utils/           # 工具函数
├── data/                # 数据文件
├── tests/               # 测试代码
├── requirements.txt     # 依赖列表
└── README.md           # 项目说明

3. LangChain核心组件深度解析

3.1 模型组件（Models）详解

LangChain中的模型组件是与各种LLM交互的桥梁，主要包括三大类型：

LLM模型调用示例：

from langchain.llms import Ollama
from langchain.chat_models import ChatOllama

# 初始化Ollama本地模型
llm = Ollama(model="llama2")
chat_model = ChatOllama(model="llama2")

# 基本文本生成
response = llm.invoke("请介绍Python编程语言")
print(response)

# 聊天模型使用
from langchain.schema import HumanMessage
message = [HumanMessage(content="你好，请帮忙写一个Python函数")]
chat_response = chat_model.invoke(message)
print(chat_response.content)

嵌入模型（Embeddings）配置：

from langchain.embeddings import OllamaEmbeddings

# 初始化嵌入模型
embeddings = OllamaEmbeddings(model="nomic-embed-text")

# 生成文本向量
text = "这是一个示例文本"
vector = embeddings.embed_query(text)
print(f"向量维度: {len(vector)}")

3.2 提示模板（Prompt Templates）设计

提示模板是LangChain中管理LLM输入的重要组件，能够提高提示的重用性和可维护性。

from langchain.prompts import PromptTemplate
from langchain.prompts.chat import ChatPromptTemplate, HumanMessagePromptTemplate

# 基础提示模板
template = """你是一个专业的{role}。请根据以下问题提供详细的回答。

问题: {question}
回答:"""

prompt = PromptTemplate(
    input_variables=["role", "question"],
    template=template
)

# 使用模板
formatted_prompt = prompt.format(role="Python专家", question="如何优化Python代码性能？")
print(formatted_prompt)

# 聊天提示模板
human_template = HumanMessagePromptTemplate.from_template(
    "请以{style}的风格回答这个问题: {query}"
)
chat_prompt = ChatPromptTemplate.from_messages([human_template])

3.3 记忆（Memory）管理机制

记忆组件让LLM能够记住对话历史，实现多轮对话的连贯性。

from langchain.memory import ConversationBufferMemory, ConversationSummaryMemory

# 对话缓冲区记忆
memory = ConversationBufferMemory(
    return_messages=True,
    memory_key="chat_history"
)

# 使用记忆的示例
memory.save_context(
    {"input": "你好，我是张三"},
    {"output": "你好张三，很高兴认识你！"}
)

# 检索记忆
history = memory.load_memory_variables({})
print(history)

# 对话摘要记忆（适合长对话）
summary_memory = ConversationSummaryMemory(llm=llm)

3.4 索引（Indexes）与检索器（Retrievers）

索引和检索器是RAG系统的核心，负责文档的存储和检索。

from langchain.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.vectorstores import Chroma

# 文档加载和分割
loader = TextLoader("data/sample.txt")
documents = loader.load()

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)
split_docs = text_splitter.split_documents(documents)

# 创建向量存储
vectorstore = Chroma.from_documents(
    documents=split_docs,
    embedding=embeddings
)

# 创建检索器
retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 3}
)

4. RAG知识库构建实战

4.1 文档预处理与向量化

构建高质量的RAG系统始于文档的妥善处理，以下是完整的预处理流程：

import os
from langchain.document_loaders import (
    PyPDFLoader, 
    Docx2txtLoader,
    WebBaseLoader
)
from langchain.text_splitter import RecursiveCharacterTextSplitter

class DocumentProcessor:
    def __init__(self, chunk_size=1000, chunk_overlap=200):
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
            length_function=len,
        )
    
    def load_documents(self, file_paths):
        """加载多种格式的文档"""
        all_docs = []
        
        for file_path in file_paths:
            if file_path.endswith('.pdf'):
                loader = PyPDFLoader(file_path)
            elif file_path.endswith('.docx'):
                loader = Docx2txtLoader(file_path)
            elif file_path.startswith('http'):
                loader = WebBaseLoader(file_path)
            else:
                loader = TextLoader(file_path)
            
            documents = loader.load()
            all_docs.extend(documents)
        
        return all_docs
    
    def process_documents(self, file_paths):
        """完整的文档处理流程"""
        # 加载文档
        raw_documents = self.load_documents(file_paths)
        print(f"加载了 {len(raw_documents)} 个文档")
        
        # 分割文档
        split_documents = self.text_splitter.split_documents(raw_documents)
        print(f"分割为 {len(split_documents)} 个文本块")
        
        return split_documents

# 使用示例
processor = DocumentProcessor()
documents = processor.process_documents([
    "data/technical_manual.pdf",
    "data/product_spec.docx"
])

4.2 向量数据库配置与优化

选择合适的向量数据库并进行优化配置对RAG系统性能至关重要：

from langchain.vectorstores import Chroma
import chromadb

class VectorStoreManager:
    def __init__(self, embedding_model, persist_directory="./chroma_db"):
        self.embedding_model = embedding_model
        self.persist_directory = persist_directory
        self.vectorstore = None
    
    def create_vectorstore(self, documents):
        """创建向量数据库"""
        self.vectorstore = Chroma.from_documents(
            documents=documents,
            embedding=self.embedding_model,
            persist_directory=self.persist_directory
        )
        return self.vectorstore
    
    def load_existing_vectorstore(self):
        """加载已有的向量数据库"""
        self.vectorstore = Chroma(
            persist_directory=self.persist_directory,
            embedding_function=self.embedding_model
        )
        return self.vectorstore
    
    def get_retriever(self, search_type="mmr", k=4, fetch_k=20):
        """配置检索器"""
        if not self.vectorstore:
            raise ValueError("向量数据库未初始化")
        
        return self.vectorstore.as_retriever(
            search_type=search_type,
            search_kwargs={"k": k, "fetch_k": fetch_k}
        )
    
    def similarity_search_with_score(self, query, k=3):
        """带相似度得分的搜索"""
        return self.vectorstore.similarity_search_with_score(query, k=k)

# 使用示例
embeddings = OllamaEmbeddings(model="nomic-embed-text")
vs_manager = VectorStoreManager(embeddings)
vectorstore = vs_manager.create_vectorstore(documents)
retriever = vs_manager.get_retriever()

4.3 RAG链的完整实现

将各个组件组合成完整的RAG流水线：

from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate
from langchain.llms import Ollama

class RAGSystem:
    def __init__(self, retriever, llm_model="llama2"):
        self.retriever = retriever
        self.llm = Ollama(model=llm_model)
        self.qa_chain = self._setup_qa_chain()
    
    def _setup_qa_chain(self):
        """设置QA链"""
        prompt_template = """基于以下上下文信息，请回答问题。如果你不知道答案，请直接说不知道，不要编造信息。

        上下文:
        {context}

        问题: {question}
        答案:"""
        
        PROMPT = PromptTemplate(
            template=prompt_template,
            input_variables=["context", "question"]
        )
        
        qa_chain = RetrievalQA.from_chain_type(
            llm=self.llm,
            chain_type="stuff",
            retriever=self.retriever,
            return_source_documents=True,
            chain_type_kwargs={"prompt": PROMPT}
        )
        
        return qa_chain
    
    def query(self, question):
        """执行查询"""
        result = self.qa_chain.invoke({"query": question})
        return {
            "answer": result["result"],
            "source_documents": result["source_documents"]
        }
    
    def batch_query(self, questions):
        """批量查询"""
        results = []
        for question in questions:
            results.append(self.query(question))
        return results

# 使用示例
rag_system = RAGSystem(retriever)
result = rag_system.query("什么是机器学习？")
print(f"答案: {result['answer']}")
print(f"参考文档: {len(result['source_documents'])} 个")

5. Ollama本地大模型部署实战

5.1 Ollama安装与模型管理

Ollama的安装和配置是本地大模型部署的关键步骤：

安装Ollama：

# Linux/macOS 安装命令
curl -fsSL https://ollama.ai/install.sh | sh

# Windows 用户可以从官网下载安装包
# 安装完成后验证
ollama --version

模型管理操作：

# 拉取模型（使用国内镜像加速）
OLLAMA_HOST=0.0.0.0 ollama pull llama2

# 查看已安装的模型
ollama list

# 运行模型
ollama run llama2

# 删除模型
ollama rm llama2

Python中集成Ollama：

import ollama
from langchain.llms import Ollama

class OllamaManager:
    def __init__(self, host="localhost", port=11434):
        self.client = ollama.Client(host=f"{host}:{port}")
        self.available_models = self._get_available_models()
    
    def _get_available_models(self):
        """获取可用的模型列表"""
        try:
            models = self.client.list()
            return [model['name'] for model in models['models']]
        except Exception as e:
            print(f"获取模型列表失败: {e}")
            return []
    
    def pull_model(self, model_name):
        """拉取模型"""
        try:
            response = self.client.pull(model_name)
            return response
        except Exception as e:
            print(f"拉取模型失败: {e}")
            return None
    
    def create_custom_model(self, base_model, custom_name, system_prompt):
        """创建自定义模型"""
        modelfile = f"""
        FROM {base_model}
        SYSTEM {system_prompt}
        """
        
        response = self.client.create(model=custom_name, modelfile=modelfile)
        return response

# 使用示例
ollama_mgr = OllamaManager()
print(f"可用模型: {ollama_mgr.available_models}")

# 初始化LangChain中的Ollama LLM
llm = Ollama(
    model="llama2",
    temperature=0.7,
    num_predict=1000
)

5.2 模型性能优化与配置

针对不同硬件配置进行模型优化：

class ModelOptimizer:
    def __init__(self, llm):
        self.llm = llm
    
    def optimize_for_cpu(self):
        """CPU优化配置"""
        # 适用于无GPU或GPU内存不足的情况
        optimized_llm = Ollama(
            model=self.llm.model,
            temperature=0.3,  # 降低随机性提高确定性
            num_thread=4,     # 根据CPU核心数调整
            num_predict=512   # 限制生成长度
        )
        return optimized_llm
    
    def optimize_for_gpu(self, gpu_layers=20):
        """GPU优化配置"""
        # 利用GPU加速推理
        optimized_llm = Ollama(
            model=self.llm.model,
            temperature=0.7,
            num_gpu=gpu_layers,  # 使用GPU层数
            num_predict=2048,
            main_gpu=0
        )
        return optimized_llm
    
    def benchmark_model(self, test_prompts, iterations=5):
        """模型性能基准测试"""
        import time
        
        results = []
        for prompt in test_prompts:
            times = []
            for i in range(iterations):
                start_time = time.time()
                response = self.llm.invoke(prompt)
                end_time = time.time()
                times.append(end_time - start_time)
            
            avg_time = sum(times) / len(times)
            results.append({
                "prompt": prompt[:50] + "...",  # 截断长提示
                "avg_response_time": avg_time,
                "min_time": min(times),
                "max_time": max(times)
            })
        
        return results

# 使用示例
optimizer = ModelOptimizer(llm)
cpu_optimized_llm = optimizer.optimize_for_cpu()

# 性能测试
test_prompts = ["请介绍人工智能", "写一个Python函数计算斐波那契数列"]
benchmark_results = optimizer.benchmark_model(test_prompts)
for result in benchmark_results:
    print(f"提示: {result['prompt']}, 平均响应时间: {result['avg_response_time']:.2f}s")

6. Agent智能体开发进阶

6.1 基础Agent构建与工具集成

Agent的核心在于能够使用工具进行推理和行动：

from langchain.agents import AgentType, initialize_agent, Tool
from langchain.utilities import WikipediaAPIWrapper
from langchain.tools import DuckDuckGoSearchRun

class BasicAgent:
    def __init__(self, llm, tools=None):
        self.llm = llm
        self.tools = tools or self._get_default_tools()
        self.agent = self._initialize_agent()
    
    def _get_default_tools(self):
        """获取默认工具集"""
        search = DuckDuckGoSearchRun()
        wikipedia = WikipediaAPIWrapper()
        
        tools = [
            Tool(
                name="搜索",
                func=search.run,
                description="用于搜索最新信息和事实"
            ),
            Tool(
                name="维基百科",
                func=wikipedia.run,
                description="用于查询百科知识和历史信息"
            )
        ]
        return tools
    
    def _initialize_agent(self):
        """初始化Agent"""
        agent = initialize_agent(
            tools=self.tools,
            llm=self.llm,
            agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
            verbose=True,
            handle_parsing_errors=True
        )
        return agent
    
    def run(self, query):
        """执行Agent任务"""
        try:
            result = self.agent.run(query)
            return result
        except Exception as e:
            return f"Agent执行出错: {e}"

# 使用示例
basic_agent = BasicAgent(llm)
result = basic_agent.run("请搜索2024年人工智能的最新发展，并在维基百科上查找相关背景知识")
print(result)

6.2 自定义工具开发

创建专门针对业务需求的定制工具：

from langchain.tools import BaseTool
from typing import Type
from pydantic import BaseModel, Field

class CalculatorInput(BaseModel):
    expression: str = Field(description="数学表达式，如 '2 + 3 * 4'")

class CalculatorTool(BaseTool):
    name = "计算器"
    description = "用于执行数学计算"
    args_schema: Type[BaseModel] = CalculatorInput
    
    def _run(self, expression: str) -> str:
        """执行计算"""
        try:
            # 安全评估数学表达式
            allowed_chars = set('0123456789+-*/(). ')
            if not all(c in allowed_chars for c in expression):
                return "表达式包含不安全字符"
            
            result = eval(expression)
            return f"{expression} = {result}"
        except Exception as e:
            return f"计算错误: {e}"
    
    def _arun(self, expression: str):
        raise NotImplementedError("异步执行未实现")

class WeatherTool(BaseTool):
    name = "天气预报"
    description = "获取指定城市的天气信息"
    
    def _run(self, city: str) -> str:
        """模拟天气查询"""
        # 实际项目中可以集成真实天气API
        weather_data = {
            "北京": "晴，15-25°C",
            "上海": "多云，18-26°C", 
            "深圳": "阵雨，22-30°C"
        }
        return weather_data.get(city, f"未找到{city}的天气信息")
    
    def _arun(self, city: str):
        raise NotImplementedError("异步执行未实现")

# 使用自定义工具的Agent
custom_tools = [CalculatorTool(), WeatherTool()]
custom_agent = BasicAgent(llm, custom_tools)

# 测试自定义工具
result1 = custom_agent.run("请计算(15 + 23) * 2的值")
result2 = custom_agent.run("查询北京的天气情况")
print(result1)
print(result2)

6.3 多Agent协作系统

构建能够协同工作的多Agent系统：

from langchain.agents import AgentExecutor
from langchain.schema import SystemMessage

class MultiAgentSystem:
    def __init__(self, llm):
        self.llm = llm
        self.agents = self._initialize_agents()
    
    def _initialize_agents(self):
        """初始化多个专业Agent"""
        # 研究Agent - 负责信息搜集
        research_tools = [Tool(
            name="网络搜索",
            func=DuckDuckGoSearchRun().run,
            description="搜索最新信息"
        )]
        research_agent = initialize_agent(
            research_tools, self.llm, 
            agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
            agent_kwargs={
                "system_message": SystemMessage(content="你是一个专业的研究员，负责搜集准确的信息")
            }
        )
        
        # 分析Agent - 负责数据分析
        analysis_tools = [CalculatorTool()]
        analysis_agent = initialize_agent(
            analysis_tools, self.llm,
            agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
            agent_kwargs={
                "system_message": SystemMessage(content="你是一个数据分析专家，擅长数值计算和逻辑分析")
            }
        )
        
        return {
            "research": research_agent,
            "analysis": analysis_agent
        }
    
    def collaborative_task(self, complex_query):
        """多Agent协作处理复杂任务"""
        # 第一阶段：研究Agent搜集信息
        research_prompt = f"请搜集关于'{complex_query}'的相关信息"
        research_result = self.agents["research"].run(research_prompt)
        
        # 第二阶段：分析Agent处理信息
        analysis_prompt = f"基于以下信息进行分析: {research_result}"
        analysis_result = self.agents["analysis"].run(analysis_prompt)
        
        return {
            "research_phase": research_result,
            "analysis_phase": analysis_result,
            "final_conclusion": f"基于研究和分析，关于'{complex_query}'的结论是..."
        }

# 使用示例
multi_agent_system = MultiAgentSystem(llm)
result = multi_agent_system.collaborative_task("人工智能在医疗领域的应用")
print("研究结果:", result["research_phase"])
print("分析结果:", result["analysis_phase"])
print("最终结论:", result["final_conclusion"])

7. 完整项目实战：企业知识库问答系统

7.1 系统架构设计

构建一个完整的企业级RAG问答系统：

import os
from datetime import datetime
import json

class EnterpriseRAGSystem:
    def __init__(self, config_path="config.json"):
        self.config = self._load_config(config_path)
        self.llm = self._initialize_llm()
        self.vectorstore = None
        self.retriever = None
        self.qa_chain = None
        
    def _load_config(self, config_path):
        """加载配置文件"""
        default_config = {
            "model": "llama2",
            "embedding_model": "nomic-embed-text",
            "chunk_size": 1000,
            "chunk_overlap": 200,
            "search_k": 4,
            "temperature": 0.3
        }
        
        if os.path.exists(config_path):
            with open(config_path, 'r', encoding='utf-8') as f:
                user_config = json.load(f)
                default_config.update(user_config)
        
        return default_config
    
    def _initialize_llm(self):
        """初始化LLM"""
        return Ollama(
            model=self.config["model"],
            temperature=self.config["temperature"]
        )
    
    def setup_knowledge_base(self, document_directory):
        """设置知识库"""
        # 加载和处理文档
        processor = DocumentProcessor(
            chunk_size=self.config["chunk_size"],
            chunk_overlap=self.config["chunk_overlap"]
        )
        
        # 获取所有文档文件
        document_files = []
        for root, dirs, files in os.walk(document_directory):
            for file in files:
                if file.endswith(('.pdf', '.docx', '.txt')):
                    document_files.append(os.path.join(root, file))
        
        documents = processor.process_documents(document_files)
        
        # 创建向量存储
        embeddings = OllamaEmbeddings(model=self.config["embedding_model"])
        self.vectorstore = Chroma.from_documents(
            documents=documents,
            embedding=embeddings,
            persist_directory="./enterprise_kb"
        )
        
        self.retriever = self.vectorstore.as_retriever(
            search_kwargs={"k": self.config["search_k"]}
        )
        
        print(f"知识库设置完成，共处理 {len(documents)} 个文档块")
    
    def setup_qa_system(self):
        """设置问答系统"""
        # 定制化提示模板
        prompt_template = """你是一个专业的企业知识库助手。请基于提供的上下文信息回答问题。

        上下文信息:
        {context}

        用户问题: {question}

        要求:
        1. 如果上下文信息不足以回答问题，请明确说明
        2. 回答要专业、准确、简洁
        3. 如果涉及多个方面，请分点说明

        回答:"""
        
        prompt = PromptTemplate(
            template=prompt_template,
            input_variables=["context", "question"]
        )
        
        self.qa_chain = RetrievalQA.from_chain_type(
            llm=self.llm,
            chain_type="stuff",
            retriever=self.retriever,
            chain_type_kwargs={"prompt": prompt},
            return_source_documents=True
        )
    
    def query(self, question, user_id="anonymous"):
        """执行查询并记录日志"""
        start_time = datetime.now()
        
        try:
            result = self.qa_chain.invoke({"query": question})
            
            # 记录查询日志
            log_entry = {
                "timestamp": datetime.now().isoformat(),
                "user_id": user_id,
                "question": question,
                "response_time": (datetime.now() - start_time).total_seconds(),
                "sources_used": len(result["source_documents"])
            }
            
            self._log_query(log_entry)
            
            return {
                "answer": result["result"],
                "sources": result["source_documents"],
                "response_time": log_entry["response_time"]
            }
            
        except Exception as e:
            error_log = {
                "timestamp": datetime.now().isoformat(),
                "user_id": user_id,
                "question": question,
                "error": str(e)
            }
            self._log_error(error_log)
            return {"error": "系统处理问题时出现错误"}
    
    def _log_query(self, log_entry):
        """记录查询日志"""
        log_file = "query_logs.jsonl"
        with open(log_file, 'a', encoding='utf-8') as f:
            f.write(json.dumps(log_entry, ensure_ascii=False) + '\n')
    
    def _log_error(self, error_log):
        """记录错误日志"""
        error_file = "error_logs.jsonl"
        with open(error_file, 'a', encoding='utf-8') as f:
            f.write(json.dumps(error_log, ensure_ascii=False) + '\n')

# 使用示例
enterprise_system = EnterpriseRAGSystem()
enterprise_system.setup_knowledge_base("企业文档")
enterprise_system.setup_qa_system()

result = enterprise_system.query("我们公司的请假流程是什么？", user_id="employee_001")
print(f"答案: {result['answer']}")
print(f"响应时间: {result['response_time']:.2f}秒")

7.2 系统监控与性能优化

为企业系统添加监控和优化功能：

class SystemMonitor:
    def __init__(self, rag_system):
        self.rag_system = rag_system
        self.metrics = {
            "total_queries": 0,
            "successful_queries": 0,
            "average_response_time": 0,
            "error_count": 0
        }
    
    def track_query(self, question, result):
        """跟踪查询性能"""
        self.metrics["total_queries"] += 1
        
        if "error" not in result:
            self.metrics["successful_queries"] += 1
            
            # 更新平均响应时间
            current_avg = self.metrics["average_response_time"]
            new_response_time = result.get("response_time", 0)
            total_success = self.metrics["successful_queries"]
            
            self.metrics["average_response_time"] = (
                (current_avg * (total_success - 1) + new_response_time) / total_success
            )
        else:
            self.metrics["error_count"] += 1
    
    def get_system_health(self):
        """获取系统健康状态"""
        success_rate = (
            self.metrics["successful_queries"] / self.metrics["total_queries"] * 100
            if self.metrics["total_queries"] > 0 else 100
        )
        
        health_status = "健康" if success_rate > 95 else "警告" if success_rate > 80 else "异常"
        
        return {
            "健康状态": health_status,
            "总查询数": self.metrics["total_queries"],
            "成功率": f"{success_rate:.1f}%",
            "平均响应时间": f"{self.metrics['average_response_time']:.2f}秒",
            "错误数": self.metrics["error_count"]
        }
    
    def generate_performance_report(self):
        """生成性能报告"""
        health = self.get_system_health()
        report = {
            "生成时间": datetime.now().isoformat(),
            "系统指标": health,
            "建议优化措施": self._generate_optimization_suggestions(health)
        }
        return report
    
    def _generate_optimization_suggestions(self, health):
        """生成优化建议"""
        suggestions = []
        
        if health["平均响应时间"] > "5.00秒":
            suggestions.append("考虑优化向量检索算法或增加硬件资源")
        
        if health["成功率"] < "90.0%":
            suggestions.append("检查知识库文档质量和LLM配置参数")
        
        if health["错误数"] > 10:
            suggestions.append("详细分析错误日志，修复系统异常")
        
        return suggestions if suggestions else ["系统运行良好，继续保持"]

# 集成监控功能
monitor = SystemMonitor(enterprise_system)

# 模拟多次查询
test_questions = [
    "公司年假政策是什么？",
    "报销流程怎么走？", 
    "技术文档在哪里找？"
]

for question in test_questions:
    result = enterprise_system.query(question)
    monitor.track_query(question, result)

# 查看系统状态
health_report = monitor.generate_performance_report()
print("系统性能报告:")
print(json.dumps(health_report, indent=2, ensure_ascii=False))

8. 常见问题与解决方案

8.1 环境配置问题排查

问题1：Ollama连接失败

症状：无法连接到Ollama服务，出现连接超时错误
解决方案：
1. 检查Ollama服务是否启动：`ollama serve`
2. 验证端口设置：默认11434端口是否被占用
3. 检查防火墙设置：确保端口访问权限

问题2：LangChain版本冲突

症状：导入模块时出现AttributeError或ImportError
解决方案：
1. 检查版本兼容性：`pip show langchain langchain-community`
2. 清理环境：`pip uninstall langchain langchain-community` 后重新安装
3. 使用虚拟环境：避免全局包冲突

问题3：内存不足错误

症状：处理大文档时出现MemoryError或OOM错误
解决方案：
1. 减小chunk_size：从1000调整为500
2. 分批处理文档：避免一次性加载所有文档
3. 增加系统交换空间：临时缓解内存压力

8.2 模型推理问题处理

问题4：响应速度过慢

# 优化代码示例
def optimize_response_speed(llm, retriever):
    """优化响应速度的配置"""
    # 限制检索文档数量
    optimized_retriever = retriever.copy()
    optimized_retriever.search_kwargs["k"] = 2
    
    # 配置LLM参数
    optimized_llm = Ollama(
        model=llm.model,
        temperature=0.1,  # 降低随机性
        num_predict=200   # 限制输出长度
    )
    
    return optimized_llm,