当前位置：首页 > news >正文

基于LLM的智能数据查询与分析系统：实现思路与完整方案

news 2025/11/2 9:04:50

基于LLM的智能数据查询与分析系统：实现思路与完整方案

1. 引言：当传统系统遇见AI革命

在当今数据驱动的商业环境中，企业存储的数据量呈指数级增长。然而，传统的数据查询系统面临着严峻的挑战：用户需要掌握复杂的查询语言（如SQL）、理解数据库结构，并具备一定的技术背景才能有效地获取所需信息。这种技术门槛导致大量业务人员无法自主进行数据探索和分析，严重影响了企业的决策效率。

大型语言模型（LLM）的出现为解决这一困境提供了革命性的解决方案。通过将自然语言处理技术集成到现有系统中，我们可以让用户使用日常语言提问，系统自动将其转换为专业的查询语句，并返回易于理解的分析结果。本文将深入探讨如何在原有系统中集成LLM查询功能，提供从架构设计到具体实现的完整解决方案。

在这里插入图片描述

2. 系统架构设计

2.1 整体架构概览

我们的目标是构建一个既能保留原有系统稳定性，又能提供智能查询功能的混合架构。系统整体设计遵循分层原则，确保各模块之间的松散耦合和高内聚。

2.2 核心模块设计

2.2.1 自然语言处理模块

此模块负责理解用户的自然语言查询，识别查询意图，并提取关键参数。

class NaturalLanguageProcessor:def __init__(self, llm_service, intent_classifier):self.llm_service = llm_serviceself.intent_classifier = intent_classifierdef process_query(self, user_query: str, user_context: dict) -> ProcessedQuery:"""处理用户自然语言查询"""# 步骤1: 意图识别intent = self.intent_classifier.classify(user_query)# 步骤2: 实体提取entities = self.extract_entities(user_query, intent)# 步骤3: 查询类型判断query_type = self.determine_query_type(user_query, intent)# 步骤4: 参数验证和补全validated_params = self.validate_and_complete_params(entities, user_context)return ProcessedQuery(original_query=user_query,intent=intent,query_type=query_type,parameters=validated_params,confidence_score=self.calculate_confidence(intent, entities))def extract_entities(self, query: str, intent: str) -> List[Entity]:"""从查询中提取关键实体，如时间范围、指标、维度等"""# 使用LLM进行实体提取prompt = f"""从以下查询中提取关键实体信息：查询: "{query}"意图: {intent}请按JSON格式返回提取的实体，包括：- time_range: 时间范围- metrics: 指标列表- dimensions: 维度列表- filters: 过滤条件- aggregation: 聚合方式"""response = self.llm_service.generate(prompt)return self.parse_entities(response)

2.2.2 查询转换引擎

此模块将处理后的自然语言转换为系统可执行的查询语句。

class QueryTranslator:def __init__(self, schema_manager, template_engine):self.schema_manager = schema_managerself.template_engine = template_engineself.query_templates = self.load_query_templates()def translate_to_sql(self, processed_query: ProcessedQuery) -> TranslatedQuery:"""将处理后的查询转换为SQL"""# 根据查询类型选择模板template = self.select_template(processed_query.query_type)# 获取数据库schema信息schema_info = self.schema_manager.get_relevant_schema(processed_query.parameters)# 填充模板sql_query = self.template_engine.fill_template(template, processed_query.parameters,schema_info)# 优化查询optimized_sql = self.optimize_query(sql_query, processed_query)return TranslatedQuery(original_query=processed_query.original_query,sql_query=optimized_sql,query_type=processed_query.query_type,parameters=processed_query.parameters)def select_template(self, query_type: str) -> str:"""根据查询类型选择合适的SQL模板"""template_map = {"data_retrieval": self.query_templates["retrieval"],"statistical_analysis": self.query_templates["statistical"],"trend_analysis": self.query_templates["trend"],"comparative_analysis": self.query_templates["comparative"],"predictive_analysis": self.query_templates["predictive"]}return template_map.get(query_type, self.query_templates["default"])

2.2.3 结果解释器

此模块将数据库返回的原始数据转换为自然语言描述，为用户提供直观的分析结果。

class ResultInterpreter:def __init__(self, llm_service, visualization_generator):self.llm_service = llm_serviceself.visualization_generator = visualization_generatordef interpret_results(self, data: pd.DataFrame, original_query: str,translated_query: TranslatedQuery) -> InterpretationResult:"""解释查询结果并生成自然语言总结"""# 生成数据洞察insights = self.generate_insights(data, translated_query)# 生成自然语言总结summary = self.generate_summary(insights, original_query, data)# 生成可视化建议visualization_suggestion = self.suggest_visualization(data, insights)# 生成可视化图表visualization = self.visualization_generator.generate(data, visualization_suggestion)return InterpretationResult(original_data=data,insights=insights,textual_summary=summary,visualization=visualization,suggested_follow_up_questions=self.generate_follow_up_questions(insights))def generate_insights(self, data: pd.DataFrame, query: TranslatedQuery) -> List[Insight]:"""使用LLM从数据中提取关键洞察"""prompt = f"""基于以下数据和查询类型，提取关键的业务洞察：查询类型: {query.query_type}查询参数: {query.parameters}数据摘要:{data.describe().to_string()}数据样例:{data.head(10).to_string()}请关注：1. 显著的趋势和模式2. 异常值或离群点3. 关键统计数据4. 业务意义以JSON格式返回洞察列表，每个洞察包含：- type: 洞察类型- description: 描述- significance: 重要性评分(1-10)- supporting_data: 支持数据"""response = self.llm_service.generate(prompt)return self.parse_insights(response)

3. 技术选型与比较

3.1 LLM模型选型比较

表1：主流LLM模型对比分析

模型名称	主要特点	适用场景	部署方式	成本考量	集成复杂度
GPT-4	强大的推理能力，丰富的知识库，支持复杂指令	复杂分析、多步骤推理、创意生成	API调用/本地部署	使用成本较高，按token计费	低，有完善的SDK
Claude 2	上下文窗口大(100K)，逻辑推理强	长文档分析、复杂逻辑推理	API调用	中等，按使用量计费	低，API接口简单
Llama 2	开源可定制，支持商业使用	企业内部部署、数据隐私要求高	本地部署	一次性硬件投入，运行成本可控	中高，需要自建服务
ChatGLM	中英双语优化，对中文支持好	中文场景、国内业务	本地/云端部署	开源免费，部署成本中等	中等，有中文社区支持
文心一言	百度开发，中文理解强	中文业务场景、国内生态集成	API调用	按调用量计费	低，中文文档完善

3.2 向量数据库选型

表2：向量数据库技术对比

数据库	存储引擎	索引算法	分布式支持	查询性能	学习曲线
Pinecone	专有云服务	多种ANN算法	自动扩展	极高	低
Chroma	轻量级嵌入	HNSW, IVF	有限支持	高	低
Weaviate	图数据库+向量	HNSW,自定义	完全支持	高	中
Milvus	列式存储	FAISS, HNSW等	完全支持	极高	高
Qdrant	Rust编写	HNSW自定义	完全支持	极高	中

3.3 系统组件技术栈

表3：系统技术栈选择

系统层级	技术选项	推荐选择	理由
前端框架	React, Vue, Angular	React	生态丰富，组件化成熟
后端框架	Spring Boot, Django, Express	Spring Boot	企业级特性，稳定性强
缓存层	Redis, Memcached	Redis	数据结构丰富，持久化支持
消息队列	Kafka, RabbitMQ, Redis Stream	Kafka	高吞吐，分布式特性强
任务调度	Celery, Quartz, Airflow	Airflow	工作流管理，监控完善
监控日志	ELK, Prometheus, Grafana	Prometheus+Grafana	云原生，集成度高

4. 具体实现方案

4.1 环境准备与依赖安装

首先，我们需要设置项目环境和安装必要的依赖。

# requirements.txt
# LLM相关
openai>=1.0.0
anthropic
transformers>=4.30.0
torch>=2.0.0
sentence-transformers# 数据处理
pandas>=1.5.0
numpy>=1.21.0
scikit-learn>=1.0.0# 数据库连接
sqlalchemy>=2.0.0
psycopg2-binary
pymysql# 向量数据库
chromadb
pymilvus# Web框架
fastapi>=0.100.0
uvicorn
pydantic>=2.0.0# 工具类
python-dotenv
loguru
tqdm# 异步处理
celery
redis# 测试
pytest
pytest-asyncio

4.2 核心系统实现

4.2.1 系统配置管理

import os
from typing import Dict, Any, Optional
from pydantic import BaseSettings, Field
from dotenv import load_dotenvload_dotenv()class SystemConfig(BaseSettings):"""系统配置类"""# 数据库配置database_url: str = Field(..., env="DATABASE_URL")database_pool_size: int = Field(20, env="DATABASE_POOL_SIZE")database_max_overflow: int = Field(30, env="DATABASE_MAX_OVERFLOW")# LLM配置openai_api_key: str = Field(..., env="OPENAI_API_KEY")openai_base_url: Optional[str] = Field(None, env="OPENAI_BASE_URL")anthropic_api_key: str = Field(..., env="ANTHROPIC_API_KEY")# 向量数据库配置chroma_persist_path: str = Field("./chroma_db", env="CHROMA_PERSIST_PATH")milvus_host: str = Field("localhost", env="MILVUS_HOST")milvus_port: int = Field(19530, env="MILVUS_PORT")# 缓存配置redis_url: str = Field("redis://localhost:6379/0", env="REDIS_URL")cache_ttl: int = Field(3600, env="CACHE_TTL")  # 1小时# 系统行为配置max_query_length: int = Field(1000, env="MAX_QUERY_LENGTH")query_timeout: int = Field(30, env="QUERY_TIMEOUT")max_results: int = Field(1000, env="MAX_RESULTS")class Config:env_file = ".env"case_sensitive = Falseclass DatabaseConfig:"""数据库连接配置"""@staticmethoddef create_engine(config: SystemConfig):from sqlalchemy import create_enginefrom sqlalchemy.pool import QueuePoolreturn create_engine(config.database_url,poolclass=QueuePool,pool_size=config.database_pool_size,max_overflow=config.database_max_overflow,pool_pre_ping=True,echo=False)

4.2.2 LLM服务抽象层

from abc import ABC, abstractmethod
from typing import List, Dict, Any, Optional
import asyncio
import logging
from openai import OpenAI
import anthropiclogger = logging.getLogger(__name__)class LLMService(ABC):"""LLM服务抽象基类"""@abstractmethodasync def generate(self, prompt: str, system_message: Optional[str] = None,**kwargs) -> str:pass@abstractmethodasync def generate_batch(self, prompts: List[str],**kwargs) -> List[str]:passclass OpenAIService(LLMService):"""OpenAI GPT服务实现"""def __init__(self, api_key: str, base_url: Optional[str] = None, model: str = "gpt-4"):self.client = OpenAI(api_key=api_key, base_url=base_url)self.model = modelself.default_params = {"temperature": 0.1,"max_tokens": 2000,"top_p": 0.9}async def generate(self, prompt: str, system_message: Optional[str] = None,**kwargs) -> str:"""生成单个回复"""messages = []if system_message:messages.append({"role": "system", "content": system_message})messages.append({"role": "user", "content": prompt})params = {**self.default_params, **kwargs}try:response = await asyncio.get_event_loop().run_in_executor(None,lambda: self.client.chat.completions.create(model=self.model,messages=messages,**params))return response.choices[0].message.contentexcept Exception as e:logger.error(f"OpenAI API调用失败: {e}")raiseasync def generate_batch(self, prompts: List[str],**kwargs) -> List[str]:"""批量生成回复"""tasks = [self.generate(prompt, **kwargs) for prompt in prompts]return await asyncio.gather(*tasks, return_exceptions=True)class ClaudeService(LLMService):"""Anthropic Claude服务实现"""def __init__(self, api_key: str, model: str = "claude-2"):self.client = anthropic.Anthropic(api_key=api_key)self.model = modelself.default_params = {"temperature": 0.1,"max_tokens_to_sample": 2000}async def generate(self, prompt: str, system_message: Optional[str] = None,**kwargs) -> str:"""生成单个回复"""params = {**self.default_params, **kwargs}try:if system_message:prompt = f"{system_message}\n\n{prompt}"response = await asyncio.get_event_loop().run_in_executor(None,lambda: self.client.completions.create(model=self.model,prompt=f"{anthropic.HUMAN_PROMPT} {prompt}{anthropic.AI_PROMPT}",**params))return response.completionexcept Exception as e:logger.error(f"Claude API调用失败: {e}")raiseclass LLMServiceFactory:"""LLM服务工厂"""@staticmethoddef create_service(provider: str, **kwargs) -> LLMService:if provider == "openai":return OpenAIService(**kwargs)elif provider == "claude":return ClaudeService(**kwargs)else:raise ValueError(f"不支持的LLM提供商: {provider}")

4.2.3 智能查询处理器

import json
import re
from typing import Dict, List, Any, Optional, Tuple
from dataclasses import dataclass
from enum import Enum
import pandas as pd
from sqlalchemy import text, create_engine
from sqlalchemy.engine import Engineclass QueryType(Enum):DATA_RETRIEVAL = "data_retrieval"STATISTICAL_ANALYSIS = "statistical_analysis"TREND_ANALYSIS = "trend_analysis"COMPARATIVE_ANALYSIS = "comparative_analysis"PREDICTIVE_ANALYSIS = "predictive_analysis"@dataclass
class ProcessedQuery:original_query: strintent: strquery_type: QueryTypeparameters: Dict[str, Any]confidence_score: float@dataclass
class TranslatedQuery:original_query: strsql_query: strquery_type: QueryTypeparameters: Dict[str, Any]class IntelligentQueryProcessor:"""智能查询处理器"""def __init__(self, llm_service: LLMService,db_engine: Engine,schema_manager: 'DatabaseSchemaManager'):self.llm_service = llm_serviceself.db_engine = db_engineself.schema_manager = schema_managerself.intent_classifier = IntentClassifier(llm_service)self.query_translator = QueryTranslator(schema_manager)self.result_interpreter = ResultInterpreter(llm_service)async def process_user_query(self, user_query: str,user_context: Dict[str, Any]) -> Dict[str, Any]:"""处理用户查询的完整流程"""# 1. 自然语言处理processed_query = await self._process_natural_language(user_query, user_context)# 2. 查询转换translated_query = await self._translate_query(processed_query)# 3. 执行查询query_result = await self._execute_query(translated_query)# 4. 结果解释interpreted_result = await self._interpret_results(query_result, user_query, translated_query)return {"original_query": user_query,"processed_query": processed_query,"translated_query": translated_query,"query_result": query_result,"interpreted_result": interpreted_result,"success": True}async def _process_natural_language(self, user_query: str,user_context: Dict[str, Any]) -> ProcessedQuery:"""处理自然语言查询"""# 使用LLM进行意图分类和实体提取prompt = f"""分析以下用户查询，提取关键信息：用户查询: "{user_query}"用户上下文: {json.dumps(user_context, ensure_ascii=False)}请按以下JSON格式返回分析结果：{{"intent": "查询意图分类","query_type": "data_retrieval|statistical_analysis|trend_analysis|comparative_analysis|predictive_analysis","parameters": {{"time_range": {{"start": "开始时间", "end": "结束时间"}},"metrics": ["指标1", "指标2"],"dimensions": ["维度1", "维度2"],"filters": [{{"field": "字段名", "operator": "操作符", "value": "值"}}],"aggregation": "聚合方式"}},"confidence": 0.95}}注意事项：1. 时间范围要转换为标准格式2. 指标和维度要映射到数据库中的实际字段名3. 过滤条件要合理推断"""response = await self.llm_service.generate(prompt)analysis_result = self._parse_llm_response(response)return ProcessedQuery(original_query=user_query,intent=analysis_result["intent"],query_type=QueryType(analysis_result["query_type"]),parameters=analysis_result["parameters"],confidence_score=analysis_result["confidence"])async def _translate_query(self, processed_query: ProcessedQuery) -> TranslatedQuery:"""转换为SQL查询"""return await self.query_translator.translate(processed_query)async def _execute_query(self, translated_query: TranslatedQuery) -> pd.DataFrame:"""执行SQL查询"""try:with self.db_engine.connect() as conn:result = conn.execute(text(translated_query.sql_query))df = pd.DataFrame(result.fetchall(), columns=result.keys())return dfexcept Exception as e:logger.error(f"查询执行失败: {e}")# 这里可以添加查询重写逻辑raiseasync def _interpret_results(self,data: pd.DataFrame,original_query: str,translated_query: TranslatedQuery) -> Dict[str, Any]:"""解释查询结果"""return await self.result_interpreter.interpret(data, original_query, translated_query)def _parse_llm_response(self, response: str) -> Dict[str, Any]:"""解析LLM响应"""try:# 尝试提取JSON部分json_match = re.search(r'\{.*\}', response, re.DOTALL)if json_match:return json.loads(json_match.group())else:return json.loads(response)except json.JSONDecodeError:logger.error(f"LLM响应解析失败: {response}")# 返回默认结构return {"intent": "unknown","query_type": "data_retrieval","parameters": {},"confidence": 0.0}

4.2.4 数据库Schema管理器

class DatabaseSchemaManager:"""数据库Schema管理器"""def __init__(self, db_engine: Engine):self.db_engine = db_engineself.schema_cache = {}self.field_mappings = self._load_field_mappings()def get_table_schema(self, table_name: str) -> Dict[str, Any]:"""获取表结构信息"""if table_name in self.schema_cache:return self.schema_cache[table_name]schema = self._fetch_table_schema(table_name)self.schema_cache[table_name] = schemareturn schemadef _fetch_table_schema(self, table_name: str) -> Dict[str, Any]:"""从数据库获取表结构"""# 这里需要根据具体数据库类型实现# 以PostgreSQL为例query = """SELECT column_name,data_type,is_nullable,column_defaultFROM information_schema.columns WHERE table_name = :table_nameORDER BY ordinal_position"""with self.db_engine.connect() as conn:result = conn.execute(text(query), {"table_name": table_name})columns = []for row in result:columns.append({"name": row.column_name,"type": row.data_type,"nullable": row.is_nullable == 'YES',"default": row.column_default})return {"table_name": table_name,"columns": columns}def map_natural_language_to_fields(self, natural_terms: List[str]) -> List[str]:"""将自然语言术语映射到数据库字段"""mapped_fields = []for term in natural_terms:if term in self.field_mappings:mapped_fields.extend(self.field_mappings[term])else:# 使用模糊匹配查找相似字段similar_fields = self._find_similar_fields(term)mapped_fields.extend(similar_fields)return list(set(mapped_fields))  # 去重def _load_field_mappings(self) -> Dict[str, List[str]]:"""加载字段映射配置"""# 可以从配置文件或数据库加载return {"销售额": ["sales_amount", "revenue"],"用户数": ["user_count", "active_users"],"时间": ["create_time", "update_time", "timestamp"],"产品": ["product_name", "product_category"],"地区": ["region", "province", "city"]}def _find_similar_fields(self, term: str) -> List[str]:"""查找相似的数据库字段"""# 可以使用向量相似度搜索# 这里简化为关键词匹配similar_fields = []for table_name in self.get_all_tables():schema = self.get_table_schema(table_name)for column in schema["columns"]:if term.lower() in column["name"].lower():similar_fields.append(f"{table_name}.{column['name']}")return similar_fieldsdef get_all_tables(self) -> List[str]:"""获取所有表名"""query = """SELECT table_name FROM information_schema.tables WHERE table_schema = 'public'"""with self.db_engine.connect() as conn:result = conn.execute(text(query))return [row.table_name for row in result]

5. 高级功能实现

5.1 查询优化与缓存

import redis
import hashlib
import pickle
from typing import Any, Optionalclass QueryOptimizer:"""查询优化器"""def __init__(self, cache_client: redis.Redis):self.cache = cache_clientself.optimization_rules = self._load_optimization_rules()def optimize_sql(self, sql: str, query_params: Dict[str, Any]) -> str:"""优化SQL查询"""# 应用优化规则optimized_sql = sqlfor rule in self.optimization_rules:optimized_sql = rule.apply(optimized_sql, query_params)return optimized_sqldef get_cached_result(self, query_signature: str) -> Optional[pd.DataFrame]:"""获取缓存结果"""cached = self.cache.get(query_signature)if cached:return pickle.loads(cached)return Nonedef cache_result(self, query_signature: str, result: pd.DataFrame, ttl: int = 3600):"""缓存查询结果"""self.cache.setex(query_signature,ttl,pickle.dumps(result))def generate_query_signature(self, sql: str, params: Dict[str, Any]) -> str:"""生成查询签名用于缓存键"""query_str = f"{sql}_{json.dumps(params, sort_keys=True)}"return hashlib.md5(query_str.encode()).hexdigest()def _load_optimization_rules(self) -> List['OptimizationRule']:"""加载优化规则"""return [SelectColumnOptimizationRule(),WhereClauseOptimizationRule(),JoinOptimizationRule(),GroupByOptimizationRule()]class OptimizationRule(ABC):"""优化规则基类"""@abstractmethoddef apply(self, sql: str, params: Dict[str, Any]) -> str:passclass SelectColumnOptimizationRule(OptimizationRule):"""SELECT列优化规则"""def apply(self, sql: str, params: Dict[str, Any]) -> str:# 移除不必要的SELECT *if "SELECT *" in sql.upper():# 分析实际需要的列required_columns = self._analyze_required_columns(sql, params)if required_columns:sql = sql.replace("SELECT *", f"SELECT {', '.join(required_columns)}")return sqldef _analyze_required_columns(self, sql: str, params: Dict[str, Any]) -> List[str]:# 基于查询参数分析需要的列# 简化实现，实际需要更复杂的分析return []class QueryCacheManager:"""查询缓存管理器"""def __init__(self, redis_url: str):self.redis_client = redis.from_url(redis_url)self.optimizer = QueryOptimizer(self.redis_client)async def execute_with_cache(self, sql: str, params: Dict[str, Any],executor: callable) -> pd.DataFrame:"""带缓存的查询执行"""# 生成查询签名query_signature = self.optimizer.generate_query_signature(sql, params)# 尝试从缓存获取cached_result = self.optimizer.get_cached_result(query_signature)if cached_result is not None:logger.info("缓存命中")return cached_result# 执行查询result = await executor(sql, params)# 缓存结果self.optimizer.cache_result(query_signature, result)return result

5.2 错误处理与重试机制

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_typeclass QueryErrorHandler:"""查询错误处理器"""def __init__(self, max_retries: int = 3):self.max_retries = max_retries@retry(stop=stop_after_attempt(3),wait=wait_exponential(multiplier=1, min=4, max=10),retry=retry_if_exception_type((ConnectionError, TimeoutError)))async def execute_with_retry(self, executor: callable, *args, **kwargs):"""带重试的执行"""try:return await executor(*args, **kwargs)except Exception as e:logger.warning(f"查询执行失败，进行重试: {e}")raisedef handle_query_error(self, error: Exception, original_query: str) -> Dict[str, Any]:"""处理查询错误"""error_type = type(error).__name__if "syntax error" in str(error).lower():return self._handle_syntax_error(error, original_query)elif "timeout" in str(error).lower():return self._handle_timeout_error(error, original_query)elif "connection" in str(error).lower():return self._handle_connection_error(error, original_query)else:return self._handle_generic_error(error, original_query)def _handle_syntax_error(self, error: Exception, original_query: str) -> Dict[str, Any]:"""处理语法错误"""# 尝试重新生成查询return {"success": False,"error_type": "syntax_error","message": "查询语法错误，正在尝试重新生成...","suggestion": "请尝试重新表述您的问题","can_retry": True}def _handle_timeout_error(self, error: Exception, original_query: str) -> Dict[str, Any]:"""处理超时错误"""return {"success": False,"error_type": "timeout_error","message": "查询执行超时","suggestion": "请尝试缩小查询范围或简化查询条件","can_retry": True}def _handle_connection_error(self, error: Exception, original_query: str) -> Dict[str, Any]:"""处理连接错误"""return {"success": False,"error_type": "connection_error","message": "数据库连接失败","suggestion": "请稍后重试或联系系统管理员","can_retry": True}def _handle_generic_error(self, error: Exception, original_query: str) -> Dict[str, Any]:"""处理通用错误"""return {"success": False,"error_type": "generic_error","message": str(error),"suggestion": "请检查查询条件或联系技术支持","can_retry": False}

6. 系统部署与监控

6.1 Docker容器化部署

# Dockerfile
FROM python:3.9-slimWORKDIR /app# 安装系统依赖
RUN apt-get update && apt-get install -y \gcc \g++ \&& rm -rf /var/lib/apt/lists/*# 复制依赖文件
COPY requirements.txt .# 安装Python依赖
RUN pip install --no-cache-dir -r requirements.txt# 复制应用代码
COPY . .# 创建非root用户
RUN useradd --create-home --shell /bin/bash app
USER app# 暴露端口
EXPOSE 8000# 启动命令
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

# docker-compose.yml
version: '3.8'services:app:build: .ports:- "8000:8000"environment:- DATABASE_URL=postgresql://user:password@db:5432/llm_query- REDIS_URL=redis://redis:6379/0depends_on:- db- redisvolumes:- ./logs:/app/logsdb:image: postgres:13environment:- POSTGRES_DB=llm_query- POSTGRES_USER=user- POSTGRES_PASSWORD=passwordvolumes:- postgres_data:/var/lib/postgresql/dataredis:image: redis:6-alpinevolumes:- redis_data:/datavolumes:postgres_data:redis_data:

6.2 性能监控与日志

import time
import logging
from functools import wraps
from prometheus_client import Counter, Histogram, generate_latest
from typing import Callable, Any# Prometheus指标
QUERY_COUNTER = Counter('llm_query_total', 'Total LLM queries', ['query_type', 'status'])
QUERY_DURATION = Histogram('llm_query_duration_seconds', 'LLM query duration')
ERROR_COUNTER = Counter('llm_query_errors_total', 'Total query errors', ['error_type'])def monitor_performance(func: Callable) -> Callable:"""性能监控装饰器"""@wraps(func)async def wrapper(*args, **kwargs):start_time = time.time()try:result = await func(*args, **kwargs)duration = time.time() - start_time# 记录指标QUERY_DURATION.observe(duration)QUERY_COUNTER.labels(query_type=kwargs.get('query_type', 'unknown'),status='success').inc()return resultexcept Exception as e:duration = time.time() - start_timeQUERY_DURATION.observe(duration)ERROR_COUNTER.labels(error_type=type(e).__name__).inc()raisereturn wrapperclass QueryLogger:"""查询日志记录器"""def __init__(self):self.logger = logging.getLogger('llm_query')def log_query(self, original_query: str,translated_query: str,execution_time: float,success: bool,error_message: str = None):"""记录查询日志"""log_entry = {"timestamp": time.time(),"original_query": original_query,"translated_query": translated_query,"execution_time": execution_time,"success": success,"error_message": error_message}if success:self.logger.info(json.dumps(log_entry))else:self.logger.error(json.dumps(log_entry))

7. 测试与验证

7.1 单元测试

import pytest
import pytest_asyncio
from unittest.mock import Mock, AsyncMockclass TestIntelligentQueryProcessor:"""智能查询处理器测试"""@pytest_asyncio.fixtureasync def processor(self):"""创建测试用的处理器实例"""llm_service = Mock(spec=LLMService)db_engine = Mock()schema_manager = Mock(spec=DatabaseSchemaManager)return IntelligentQueryProcessor(llm_service=llm_service,db_engine=db_engine,schema_manager=schema_manager)@pytest.mark.asyncioasync def test_process_simple_query(self, processor):"""测试简单查询处理"""# 设置mockprocessor.llm_service.generate = AsyncMock(return_value=json.dumps({"intent": "数据查询","query_type": "data_retrieval","parameters": {"metrics": ["sales"],"time_range": {"start": "2023-01-01", "end": "2023-12-31"}},"confidence": 0.95}))# 执行测试result = await processor.process_user_query("显示2023年的销售额",{"user_id": "test_user"})# 验证结果assert result["success"] == Trueassert result["processed_query"].intent == "数据查询"@pytest.mark.asyncioasync def test_query_translation(self, processor):"""测试查询翻译"""processed_query = ProcessedQuery(original_query="测试查询",intent="数据查询",query_type=QueryType.DATA_RETRIEVAL,parameters={"metrics": ["sales"]},confidence_score=0.9)# 测试翻译逻辑translated = await processor._translate_query(processed_query)assert translated.sql_query is not Noneassert translated.query_type == QueryType.DATA_RETRIEVALclass TestErrorHandling:"""错误处理测试"""@pytest.mark.asyncioasync def test_network_error_retry(self):"""测试网络错误重试"""error_handler = QueryErrorHandler()mock_executor = AsyncMock(side_effect=[ConnectionError(), "success"])result = await error_handler.execute_with_retry(mock_executor, "test_query")assert result == "success"assert mock_executor.call_count == 2

7.2 集成测试

class TestEndToEnd:"""端到端测试"""@pytest.mark.asyncioasync def test_complete_workflow(self):"""测试完整工作流"""# 初始化真实组件（测试环境）config = SystemConfig()db_engine = DatabaseConfig.create_engine(config)llm_service = LLMServiceFactory.create_service("openai", api_key="test_key")processor = IntelligentQueryProcessor(llm_service=llm_service,db_engine=db_engine,schema_manager=DatabaseSchemaManager(db_engine))# 执行端到端测试result = await processor.process_user_query("查询最近一周的用户活跃度",{"user_id": "test_user", "department": "数据分析"})# 验证各个环节assert result["success"] == Trueassert "interpreted_result" in resultassert "visualization" in result["interpreted_result"]

8. 实际应用案例

8.1 电商数据分析

# 电商场景专用处理器
class EcommerceQueryProcessor(IntelligentQueryProcessor):"""电商领域专用查询处理器"""def __init__(self, *args, **kwargs):super().__init__(*args, **kwargs)self.domain_knowledge = self._load_ecommerce_knowledge()async def process_ecommerce_query(self, user_query: str, user_role: str) -> Dict[str, Any]:"""处理电商领域查询"""# 添加上下文信息context = {"user_role": user_role,"domain": "ecommerce","available_metrics": self.domain_knowledge["metrics"],"available_dimensions": self.domain_knowledge["dimensions"]}return await self.process_user_query(user_query, context)def _load_ecommerce_knowledge(self) -> Dict[str, Any]:"""加载电商领域知识"""return {"metrics": ["gmv", "order_count", "user_count", "conversion_rate","average_order_value", "refund_rate", "customer_lifetime_value"],"dimensions": ["product_category", "user_segment", "sales_region","time_period", "marketing_channel"],"common_queries": {"sales_analysis": "销售分析","user_behavior": "用户行为分析","inventory_management": "库存管理","marketing_effectiveness": "营销效果分析"}}# 使用示例
async def demo_ecommerce_analysis():"""电商分析演示"""processor = EcommerceQueryProcessor(...)queries = ["显示上周各个品类的销售额和增长率","分析新老用户的购买行为差异","预测下个季度的热门商品","比较不同营销渠道的转化率"]for query in queries:result = await processor.process_ecommerce_query(query, "business_analyst")print(f"查询: {query}")print(f"结果: {result['interpreted_result']['textual_summary']}")print("---")

9. 性能优化建议

9.1 查询性能优化

索引优化：为常用查询字段创建合适的索引
查询重写：自动优化生成的SQL查询
结果缓存：缓存频繁查询的结果
分页处理：大数据集的分页查询优化

9.2 LLM调用优化

提示词优化：设计高效的提示词模板
批量处理：合并多个LLM调用
缓存响应：缓存相似的LLM查询结果
降级策略：在LLM服务不可用时使用规则引擎

10. 安全与隐私考虑

10.1 数据安全

class SecurityManager:"""安全管理器"""def __init__(self):self.sensitive_patterns = self._load_sensitive_patterns()def sanitize_query(self, query: str) -> str:"""清理查询中的敏感信息"""sanitized = queryfor pattern in self.sensitive_patterns:sanitized = re.sub(pattern, "[REDACTED]", sanitized, flags=re.IGNORECASE)return sanitizeddef validate_query_access(self, user_context: Dict[str, Any], query: ProcessedQuery) -> bool:"""验证查询访问权限"""# 基于用户角色和数据权限验证user_role = user_context.get("role", "user")requested_data = self._extract_data_scope(query)return self._check_permissions(user_role, requested_data)def _load_sensitive_patterns(self) -> List[str]:"""加载敏感信息模式"""return [r'\b\d{4}-\d{4}-\d{4}-\d{4}\b',  # 信用卡号r'\b\d{3}-\d{2}-\d{4}\b',  # 社保号r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'  # 邮箱]