随着AI应用开发的快速发展LangChain作为构建智能代理的核心框架已经成为开发者必须掌握的关键技术。在实际项目中很多团队在集成LangChain时都会遇到环境配置复杂、调试困难、部署不稳定等问题。本文将系统性地讲解LangChain从基础概念到生产部署的全流程包含完整的代码示例和实战案例帮助开发者快速掌握这一重要技术。1. LangChain核心概念解析1.1 什么是LangChainLangChain是一个开源的AI应用开发框架专门用于构建、测试和部署可靠的AI代理。它提供了一套完整的工具链让开发者能够快速搭建基于大语言模型的智能应用系统。从技术架构来看LangChain主要包含以下几个核心组件链Chains将多个LLM调用组合成工作流代理Agents能够根据目标自主选择工具和执行动作的智能体记忆Memory维护对话历史和上下文信息工具Tools代理可以调用的外部功能接口1.2 LangChain与LangGraph的区别很多开发者容易混淆LangChain和LangGraph其实它们是同一生态下的不同工具LangChain更适合快速原型开发提供了大量预构建的模板和组件让开发者能够快速搭建基础的AI应用。它抽象了很多底层细节适合初学者和需要快速验证想法的场景。LangGraph则提供了更低级别的控制能力适合构建需要精确控制流程的生产级代理。它允许开发者定义复杂的状态机和执行逻辑在处理需要严格确定性的业务场景时更有优势。1.3 LangChain的应用场景LangChain在实际项目中有着广泛的应用智能客服系统构建能够理解上下文、调用知识库的对话机器人数据分析助手创建能够执行SQL查询、生成可视化图表的分析代理文档处理流水线开发自动化的文档摘要、分类和提取系统代码生成工具搭建能够理解需求、生成和调试代码的编程助手2. 环境准备与安装配置2.1 系统要求与前置条件在开始学习LangChain之前需要确保开发环境满足以下要求Python版本3.8或更高版本操作系统Windows 10/11, macOS 10.15, 或Linux Ubuntu 18.04内存至少8GB RAM处理大模型时推荐16GB网络能够稳定访问PyPI和模型API2.2 安装LangChain核心包使用pip安装LangChain的最新稳定版本# 安装核心包 pip install langchain # 安装常用的扩展组件 pip install langchain-community langchain-core langchain-text-splitters # 安装OpenAI接口如果需要使用GPT模型 pip install openai # 安装本地模型支持可选 pip install transformers torch2.3 配置开发环境推荐使用VS Code或PyCharm作为开发IDE并安装以下有用的扩展// VS Code推荐配置在settings.json中 { python.defaultInterpreterPath: ./venv/bin/python, python.analysis.extraPaths: [./src], editor.formatOnSave: true, python.linting.enabled: true }创建项目目录结构my-langchain-project/ ├── src/ │ ├── agents/ # 代理实现 │ ├── chains/ # 链组件 │ ├── tools/ # 自定义工具 │ └── utils/ # 工具函数 ├── tests/ # 测试代码 ├── config/ # 配置文件 ├── requirements.txt # 依赖列表 └── main.py # 入口文件3. LangChain核心组件详解3.1 模型Models组件模型是LangChain的核心支持多种类型的LLM提供商from langchain.llms import OpenAI from langchain.chat_models import ChatOpenAI from langchain.embeddings import OpenAIEmbeddings # 基础文本生成模型 llm OpenAI( model_namegpt-3.5-turbo-instruct, temperature0.7, max_tokens1000 ) # 对话优化模型 chat_model ChatOpenAI( model_namegpt-4, temperature0.5 ) # 嵌入模型用于文本向量化 embeddings OpenAIEmbeddings() # 使用本地模型降低成本 from langchain.llms import LlamaCpp llm_local LlamaCpp( model_path./models/llama-7b.gguf, temperature0.7, n_ctx2048 )3.2 提示模板Prompt Templates提示模板帮助标准化LLM输入提高代码可维护性from langchain.prompts import PromptTemplate, ChatPromptTemplate # 基础提示模板 template 你是一个专业的{role}。请根据以下问题提供详细的回答。 问题{question} 回答 prompt PromptTemplate( input_variables[role, question], templatetemplate ) # 使用模板 formatted_prompt prompt.format( role软件工程师, question如何优化Python代码的性能 ) # 对话提示模板更结构化 chat_template ChatPromptTemplate.from_messages([ (system, 你是一个有帮助的AI助手), (human, 请帮我解释一下{concept}) ]) # 复杂模板示例包含示例 few_shot_template 请根据示例回答问题。 示例 输入什么是Python的列表推导式 输出列表推导式是Python中创建列表的简洁方式... 现在请回答 输入{question} 输出3.3 记忆Memory管理记忆组件让AI代理能够记住对话历史from langchain.memory import ConversationBufferMemory, ConversationSummaryMemory # 基础对话记忆 memory ConversationBufferMemory( return_messagesTrue, memory_keychat_history ) # 摘要记忆处理长对话 summary_memory ConversationSummaryMemory( llmchat_model, return_messagesTrue ) # 实体记忆记住特定信息 from langchain.memory import EntityMemory entity_memory EntityMemory(llmchat_model) # 使用记忆的完整示例 from langchain.chains import ConversationChain conversation ConversationChain( llmchat_model, memorymemory, verboseTrue ) response conversation.predict(input你好我是张三) print(response) # 第一次对话 response conversation.predict(input你还记得我叫什么名字吗) print(response) # 会记得之前的对话3.4 链Chains的构建与使用链将多个组件连接成完整的工作流from langchain.chains import LLMChain, SequentialChain # 基础链 llm_chain LLMChain( llmllm, promptprompt, output_keyanswer ) # 顺序链多个步骤 overall_chain SequentialChain( chains[chain1, chain2, chain3], input_variables[input], output_variables[final_output], verboseTrue ) # 转换链数据预处理 from langchain.chains import TransformChain def transform_func(inputs): text inputs[text] return {processed_text: text.upper()} transform_chain TransformChain( input_variables[text], output_variables[processed_text], transformtransform_func )4. 代理Agents开发实战4.1 代理的基本概念代理是LangChain中最强大的功能它能够根据目标自主选择和执行动作from langchain.agents import initialize_agent, AgentType from langchain.tools import Tool # 定义自定义工具 def search_tool(query): 模拟搜索工具 return f搜索结果{query}的相关信息 def calculator_tool(expression): 计算工具 try: result eval(expression) return f计算结果{expression} {result} except: return 计算错误 # 创建工具列表 tools [ Tool( name搜索, funcsearch_tool, description用于搜索信息 ), Tool( name计算器, funccalculator_tool, description用于数学计算 ) ] # 初始化代理 agent initialize_agent( toolstools, llmchat_model, agentAgentType.ZERO_SHOT_REACT_DESCRIPTION, verboseTrue, handle_parsing_errorsTrue )4.2 代理类型详解LangChain提供了多种代理类型适应不同场景# 1. Zero-Shot React代理最常用 zero_shot_agent initialize_agent( toolstools, llmllm, agentAgentType.ZERO_SHOT_REACT_DESCRIPTION, verboseTrue ) # 2. Conversational React代理对话优化 conversational_agent initialize_agent( toolstools, llmchat_model, agentAgentType.CONVERSATIONAL_REACT_DESCRIPTION, memorymemory, verboseTrue ) # 3. Self-Ask with Search代理问答专用 self_ask_agent initialize_agent( tools[search_tool], llmllm, agentAgentType.SELF_ASK_WITH_SEARCH, verboseTrue ) # 使用代理执行复杂任务 result conversational_agent.run( 请先计算(15 25) * 2的结果然后搜索人工智能的最新发展 )4.3 自定义工具开发创建更专业的工具来扩展代理能力import requests from langchain.tools import BaseTool from typing import Type class WeatherTool(BaseTool): name 天气查询 description 用于查询城市天气信息 def _run(self, city: str) - str: 实际执行天气查询 # 这里使用模拟数据实际项目中可以接入真实API weather_data { 北京: 晴25°C, 上海: 多云23°C, 深圳: 雨28°C } return weather_data.get(city, 城市不存在) def _arun(self, city: str): raise NotImplementedError(不支持异步) class StockTool(BaseTool): name 股票查询 description 查询股票实时价格 def _run(self, symbol: str) - str: # 模拟股票数据 stock_prices { AAPL: 172.35 USD, TSLA: 245.18 USD, MSFT: 328.79 USD } return stock_prices.get(symbol, 股票代码不存在) # 使用自定义工具 custom_tools [WeatherTool(), StockTool()] advanced_agent initialize_agent( toolscustom_tools, llmchat_model, agentAgentType.ZERO_SHOT_REACT_DESCRIPTION, verboseTrue )5. 完整项目实战智能个人助理5.1 项目需求分析我们将构建一个功能完整的智能个人助理具备以下能力天气查询和提醒日程管理信息搜索计算功能文件操作对话记忆5.2 项目结构设计smart-assistant/ ├── src/ │ ├── agents/ │ │ └── personal_assistant.py │ ├── tools/ │ │ ├── weather_tool.py │ │ ├── calendar_tool.py │ │ ├── file_tool.py │ │ └── web_search_tool.py │ ├── memory/ │ │ └── enhanced_memory.py │ └── utils/ │ └── config_loader.py ├── data/ │ └── schedules.json ├── config/ │ └── settings.yaml └── main.py5.3 核心代码实现工具类实现# src/tools/weather_tool.py import json from datetime import datetime from langchain.tools import BaseTool from typing import Type class AdvancedWeatherTool(BaseTool): name advanced_weather description 获取详细天气信息包括温度、湿度、风速等 def _run(self, city: str) - str: # 模拟详细的天气数据 weather_info { 北京: { temperature: 25°C, humidity: 45%, wind_speed: 15 km/h, condition: 晴, update_time: datetime.now().strftime(%Y-%m-%d %H:%M) }, 上海: { temperature: 23°C, humidity: 65%, wind_speed: 12 km/h, condition: 多云, update_time: datetime.now().strftime(%Y-%m-%d %H:%M) } } if city in weather_info: info weather_info[city] return f{city}天气信息 温度{info[temperature]} 湿度{info[humidity]} 风速{info[wind_speed]} 天气状况{info[condition]} 更新时间{info[update_time]} else: return f找不到{city}的天气信息 # src/tools/calendar_tool.py import json from datetime import datetime, timedelta from langchain.tools import BaseTool class CalendarTool(BaseTool): name calendar description 管理个人日程包括添加、查看和删除日程 def __init__(self, data_filedata/schedules.json): super().__init__() self.data_file data_file self._load_schedules() def _load_schedules(self): try: with open(self.data_file, r, encodingutf-8) as f: self.schedules json.load(f) except: self.schedules [] def _save_schedules(self): with open(self.data_file, w, encodingutf-8) as f: json.dump(self.schedules, f, ensure_asciiFalse, indent2) def _run(self, action: str, **kwargs) - str: if action add: return self._add_schedule(**kwargs) elif action list: return self._list_schedules() elif action delete: return self._delete_schedule(**kwargs) else: return 未知操作 def _add_schedule(self, title: str, time: str, description: str ) - str: schedule { id: len(self.schedules) 1, title: title, time: time, description: description, created_at: datetime.now().isoformat() } self.schedules.append(schedule) self._save_schedules() return f日程添加成功{title} - {time} def _list_schedules(self) - str: if not self.schedules: return 当前没有日程安排 result 当前日程安排\n for schedule in self.schedules: result f{schedule[id]}. {schedule[title]} - {schedule[time]}\n return result智能助理主程序# src/agents/personal_assistant.py from langchain.agents import initialize_agent, AgentType from langchain.chat_models import ChatOpenAI from langchain.memory import ConversationBufferWindowMemory from .tools.weather_tool import AdvancedWeatherTool from .tools.calendar_tool import CalendarTool from .tools.file_tool import FileOperationTool from .tools.web_search_tool import WebSearchTool class PersonalAssistant: def __init__(self, api_key: str None): # 初始化模型 self.llm ChatOpenAI( model_namegpt-3.5-turbo, temperature0.3, openai_api_keyapi_key ) # 初始化记忆 self.memory ConversationBufferWindowMemory( memory_keychat_history, k10, return_messagesTrue ) # 初始化工具 self.tools [ AdvancedWeatherTool(), CalendarTool(), FileOperationTool(), WebSearchTool() ] # 创建代理 self.agent initialize_agent( toolsself.tools, llmself.llm, agentAgentType.CONVERSATIONAL_REACT_DESCRIPTION, memoryself.memory, verboseTrue, handle_parsing_errorsTrue ) def chat(self, message: str) - str: 与助理对话 try: response self.agent.run(message) return response except Exception as e: return f处理请求时出现错误{str(e)} def get_memory_summary(self) - str: 获取对话记忆摘要 return self.memory.load_memory_variables({}) # main.py from src.agents.personal_assistant import PersonalAssistant def main(): # 初始化助理 assistant PersonalAssistant() print(智能个人助理已启动输入退出结束对话) print( * 50) while True: try: user_input input(\n您) if user_input.lower() in [退出, exit, quit]: print(助理再见期待下次为您服务) break response assistant.chat(user_input) print(f助理{response}) except KeyboardInterrupt: print(\n\n对话已中断) break except Exception as e: print(f发生错误{e}) if __name__ __main__: main()5.4 运行与测试启动程序并进行功能测试# 运行程序 python main.py # 测试对话示例 用户今天北京的天气怎么样 助理北京天气信息 温度25°C 湿度45% 风速15 km/h 天气状况晴 更新时间2024-01-15 14:30 用户帮我添加一个明天下午3点的会议 助理日程添加成功会议 - 明天下午3点 用户查看我所有的日程 助理当前日程安排 1. 会议 - 明天下午3点6. 高级特性与优化技巧6.1 检查点Checkpoint机制检查点是LangChain中的重要概念用于保存和恢复代理状态from langchain.agents import AgentExecutor from langchain.schema import AgentAction, AgentFinish class CheckpointManager: def __init__(self): self.checkpoints {} def save_checkpoint(self, agent_executor: AgentExecutor, checkpoint_id: str): 保存代理状态 state { memory: agent_executor.memory.load_memory_variables({}), intermediate_steps: getattr(agent_executor, intermediate_steps, []) } self.checkpoints[checkpoint_id] state return checkpoint_id def load_checkpoint(self, agent_executor: AgentExecutor, checkpoint_id: str): 恢复代理状态 if checkpoint_id in self.checkpoints: state self.checkpoints[checkpoint_id] agent_executor.memory.save_context( state[memory][input], state[memory][output] ) setattr(agent_executor, intermediate_steps, state[intermediate_steps]) return True return False # 使用检查点 checkpoint_manager CheckpointManager() agent_executor AgentExecutor.from_agent_and_tools( agentagent, toolstools, memorymemory, verboseTrue ) # 保存状态 checkpoint_id checkpoint_manager.save_checkpoint(agent_executor, conversation_1) # 恢复状态如程序重启后 checkpoint_manager.load_checkpoint(agent_executor, conversation_1)6.2 性能优化策略批量处理优化from langchain.callbacks import StreamingStdOutCallbackHandler from langchain.schema import BaseOutputParser import asyncio class BatchProcessor: def __init__(self, llm, max_concurrent5): self.llm llm self.semaphore asyncio.Semaphore(max_concurrent) async def process_batch(self, prompts: list) - list: 批量处理提示词 async def process_single(prompt): async with self.semaphore: return await self.llm.apredict(prompt) tasks [process_single(prompt) for prompt in prompts] return await asyncio.gather(*tasks) # 缓存优化 from langchain.cache import InMemoryCache from langchain.globals import set_llm_cache # 启用缓存减少API调用 set_llm_cache(InMemoryCache()) # 流式输出改善用户体验 class CustomCallbackHandler(StreamingStdOutCallbackHandler): def on_llm_new_token(self, token: str, **kwargs) - None: print(token, end, flushTrue)6.3 错误处理与重试机制from tenacity import retry, stop_after_attempt, wait_exponential from langchain.schema import OutputParserException class RobustAgent: def __init__(self, agent, max_retries3): self.agent agent self.max_retries max_retries retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10) ) def run_with_retry(self, input_text: str) - str: 带重试机制的运行方法 try: return self.agent.run(input_text) except OutputParserException as e: # 解析错误时的处理 return f解析响应时出错{str(e)} except Exception as e: # 其他错误处理 raise e def safe_run(self, input_text: str) - str: 安全运行捕获所有异常 for attempt in range(self.max_retries): try: return self.run_with_retry(input_text) except Exception as e: if attempt self.max_retries - 1: return f经过{self.max_retries}次尝试后仍失败{str(e)} print(f第{attempt 1}次尝试失败重试中...)7. 部署与生产环境配置7.1 本地部署方案使用FastAPI构建Web服务# api/main.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from src.agents.personal_assistant import PersonalAssistant import uvicorn app FastAPI(title智能助理API, version1.0.0) # 全局助理实例 assistant PersonalAssistant() class ChatRequest(BaseModel): message: str user_id: str class ChatResponse(BaseModel): response: str status: str app.post(/chat, response_modelChatResponse) async def chat_endpoint(request: ChatRequest): try: response assistant.chat(request.message) return ChatResponse(responseresponse, statussuccess) except Exception as e: raise HTTPException(status_code500, detailstr(e)) app.get(/health) async def health_check(): return {status: healthy, service: smart-assistant} if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8000)Docker化部署# Dockerfile FROM python:3.9-slim WORKDIR /app # 复制依赖文件 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # 复制应用代码 COPY . . # 创建非root用户 RUN useradd -m -u1000 appuser chown -R appuser:appuser /app USER appuser # 暴露端口 EXPOSE 8000 # 启动命令 CMD [python, api/main.py]7.2 云平台部署使用Docker Compose编排服务# docker-compose.yml version: 3.8 services: smart-assistant: build: . ports: - 8000:8000 environment: - OPENAI_API_KEY${OPENAI_API_KEY} - LOG_LEVELINFO volumes: - ./data:/app/data restart: unless-stopped nginx: image: nginx:alpine ports: - 80:80 volumes: - ./nginx.conf:/etc/nginx/nginx.conf depends_on: - smart-assistant7.3 监控与日志配置# src/utils/logger.py import logging import json from datetime import datetime def setup_logging(): 配置结构化日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(logs/app.log), logging.StreamHandler() ] ) class UsageTracker: def __init__(self): self.usage_data [] def track_usage(self, user_id: str, action: str, tokens_used: int): 跟踪使用情况 record { timestamp: datetime.now().isoformat(), user_id: user_id, action: action, tokens_used: tokens_used } self.usage_data.append(record) # 定期保存到文件 if len(self.usage_data) % 100 0: self._save_usage_data() def _save_usage_data(self): with open(logs/usage.json, w) as f: json.dump(self.usage_data, f, indent2)8. 常见问题与解决方案8.1 安装与环境问题问题1依赖冲突导致安装失败解决方案# 创建干净的虚拟环境 python -m venv langchain-env source langchain-env/bin/activate # Linux/Mac # 或 langchain-env\Scripts\activate # Windows # 使用conda管理环境 conda create -n langchain python3.9 conda activate langchain # 优先安装基础包 pip install langchain-core pip install langchain-community问题2API密钥配置错误解决方案# 正确的API密钥管理方式 import os from dotenv import load_dotenv load_dotenv() # 加载.env文件 # 从环境变量读取 api_key os.getenv(OPENAI_API_KEY) if not api_key: raise ValueError(请设置OPENAI_API_KEY环境变量) llm ChatOpenAI(openai_api_keyapi_key)8.2 运行时常见错误问题3代理解析错误解决方案# 增强错误处理 from langchain.agents import AgentExecutor agent_executor AgentExecutor.from_agent_and_tools( agentagent, toolstools, handle_parsing_errorsTrue, # 关键参数 max_iterations5, # 限制迭代次数 early_stopping_methodgenerate # 提前停止策略 ) # 自定义错误处理 def safe_agent_run(agent, input_text): try: return agent.run(input_text) except ValueError as e: if Could not parse LLM output in str(e): return 抱歉我无法理解这个请求请换种方式提问。 else: raise e问题4记忆丢失或混乱解决方案# 使用更稳定的记忆实现 from langchain.memory import CombinedMemory, ConversationBufferMemory, VectorStoreRetrieverMemory # 组合多种记忆类型 memory CombinedMemory(memories[ ConversationBufferMemory(memory_keychat_history), VectorStoreRetrieverMemory(retrievervectorstore.as_retriever()) ]) # 定期清理和优化记忆 def optimize_memory(memory, max_entries100): 优化记忆存储 if hasattr(memory, chat_memory) and len(memory.chat_memory.messages) max_entries: # 保留最近的消息 memory.chat_memory.messages memory.chat_memory.messages[-max_entries:]8.3 性能优化问题问题5响应速度慢优化策略# 1. 使用更小的模型 fast_llm ChatOpenAI(model_namegpt-3.5-turbo, temperature0.3) # 2. 启用流式输出 from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler llm ChatOpenAI( streamingTrue, callbacks[StreamingStdOutCallbackHandler()], temperature0.3 ) # 3. 实现缓存机制 from langchain.cache import SQLiteCache import langchain langchain.llm_cache SQLiteCache(database_path.langchain.db)问题6Token消耗过高成本控制方案# Token使用监控 from langchain.callbacks import get_openai_callback def track_token_usage(func): 装饰器跟踪Token使用 def wrapper(*args, **kwargs): with get_openai_callback() as cb: result func(*args, **kwargs) print(f本次调用消耗Tokens: {cb.total_tokens}) return result return wrapper # 使用摘要减少上下文长度 from langchain.memory import ConversationSummaryMemory summary_memory ConversationSummaryMemory( llmChatOpenAI(temperature0), return_messagesTrue )9. 最佳实践与工程建议9.1 代码组织规范模块化设计# 推荐的项目结构 project/ ├── agents/ # 代理实现 │ ├── __init__.py │ ├── base_agent.py │ └── specialized_agents/ ├── chains/ # 链组件 │ ├── __init__.py │ └── custom_chains.py ├── tools/ # 工具库 │ ├── __init__.py │ ├── external_apis/ │ └── utilities/ ├── memory/ # 记忆管理 │ ├── __init__.py │ └── enhanced_memory.py ├── config/ # 配置管理 │ ├── __init__.py │ └── settings.py └── utils/ # 工具函数 ├── __init__.py └── helpers.py配置管理最佳实践# config/settings.py import os from typing import Optional from pydantic import BaseSettings class Settings(BaseSettings): # API配置 openai_api_key: Optional[str] None model_name: str gpt-3.5-turbo temperature: float 0.3 # 应用配置 max_iterations: int 5 memory_window_size: int 10 # 日志配置 log_level: str INFO class Config: env_file .env case_sensitive False # 单例配置实例 settings Settings()9.2 安全与权限控制API安全实践# 安全的API调用封装 import secrets from fastapi import Security, HTTPException from fastapi.security import APIKeyHeader api_key_header APIKeyHeader(nameX-API-Key) def validate_api_key(api_key: str Security(api_key_header)): 验证API密钥 valid_keys get_valid_api_keys() # 从数据库或配置获取 if api_key not in valid_keys: raise HTTPException(status_code403, detail无效的API密钥) return api_key # 输入验证和清理 from langchain.schema import BaseOutputParser import html class SafeOutputParser(BaseOutputParser): 安全的输出解析器 def parse(self, text: str) - str: # 清理HTML标签 cleaned_text html.escape(text) # 其他安全处理 return cleaned_text9.3 测试与质量保证单元测试示例# tests/test_agents.py import unittest from unittest.mock import Mock, patch from src.agents.personal_assistant import PersonalAssistant class TestPersonalAssistant(unittest.TestCase): def setUp(self): 测试前置设置 self.assistant PersonalAssistant() patch(src.agents.personal_assistant.ChatOpenAI) def test_chat_response(self, mock_llm): 测试对话响应 # 模拟LLM响应 mock_instance mock_llm.return_value mock_instance.predict.return_value 这是一个测试响应 response self.assistant.chat(你好) self.assertEqual(response, 这是一个测试响应) def test_memory_management(self): 测试记忆管理 # 测试记忆保存和恢复 initial_memory self.assistant.get_memory_summary() self.assistant.chat(记住我喜欢编程) updated_memory self.assistant.get_memory_summary() self.assertNotEqual(initial_memory, updated_memory) if __name__ __main__: unittest.main()集成测试策略# tests/integration/test_full_workflow.py import pytest from langchain.agents import AgentType class TestFullWorkflow: 完整工作流测试 pytest.fixture def setup_assistant(self): 测试夹具 assistant PersonalAssistant() yield assistant def test_complex_conversation(self, setup_assistant): 测试复杂对话流程 assistant setup_assistant # 多轮对话测试 responses [] questions [ 今天天气怎么样, 帮我计算15*20的结果, 还记得我问的第一个问题吗 ] for question in questions: response assistant.chat(question) responses.append(response) # 验证响应完整性 assert len(responses) 3 assert all(responses) # 所有响应都不为空9.4 性能监控与优化监控指标收集# src/utils/monitoring.py import time from dataclasses import dataclass from typing import Dict, List import statistics dataclass class PerformanceMetrics: response_time: float token_usage: int success: bool class PerformanceMonitor: def __init__(self): self.metrics: List[Performance