在数据驱动的业务场景中如何让非技术背景的团队成员也能高效查询和分析领域特定的元数据一直是企业数据平台建设的痛点。传统 SQL 或 API 查询方式门槛较高而直接开放数据库权限又存在安全风险。本文基于大语言模型LLM技术设计一套可复用的自然语言查询生成框架将自然语言问题自动转换为结构化的元数据查询语句实现安全、便捷的数据自助服务。无论你是数据平台开发者、算法工程师还是业务数据分析师都能通过本文掌握从架构设计、核心模块实现到生产部署的全流程。我们将使用 Python 和主流 LLM API如 OpenAI GPT 或开源模型构建一个完整的领域元数据查询系统涵盖提示词工程、查询验证、错误处理等关键环节。1. 自然语言查询框架的核心概念1.1 什么是自然语言到元数据查询自然语言到元数据查询Natural Language to Metadata Query是指将用户用日常语言提出的数据问题自动转换为计算机可执行的元数据查询语句如 SQL、GraphQL 或特定 API 调用的技术。例如用户输入显示最近一个月销售额最高的产品系统应自动生成对应的 SQL 查询SELECT product_name, SUM(sales_amount) as total_sales FROM sales_table WHERE sale_date DATE_SUB(CURRENT_DATE, INTERVAL 1 MONTH) GROUP BY product_name ORDER BY total_sales DESC LIMIT 10;这种技术的核心价值在于降低数据查询门槛让业务人员可以直接用自然语言与数据系统交互而无需掌握复杂的技术语法。1.2 领域特定元数据的特点与挑战领域特定元数据Domain-Specific Metadata指的是在特定业务领域内描述数据特征的信息如电商领域的商品分类、库存状态或金融领域的交易类型、风险等级等。与通用元数据相比领域元数据具有以下特点业务语义丰富每个字段都承载具体的业务含义结构复杂多样可能涉及多个关联表和复杂约束变更频繁随着业务发展不断调整和扩展访问权限精细不同角色只能访问特定范围的数据这些特点给自然语言查询带来三大挑战语义理解难度大需要准确理解领域术语和业务逻辑查询生成准确性要求高错误的查询可能导致错误决策安全控制复杂需要确保生成的查询不越权访问数据1.3 LLM 在查询生成中的角色定位大语言模型在自然语言查询框架中扮演语义理解与转换引擎的角色其核心能力包括意图识别判断用户想要查询什么类型的数据实体提取识别查询中涉及的领域实体和属性逻辑推理理解时间范围、比较关系、聚合条件等复杂逻辑语法生成根据目标查询语言的语法规则生成正确语句但 LLM 并非万能需要框架在以下方面提供支持领域知识增强提供业务元数据上下文查询验证检查生成语句的语法和逻辑正确性安全约束确保查询符合权限控制要求错误处理对模糊或无法处理的请求提供友好反馈2. 框架架构设计与技术选型2.1 整体系统架构我们的自然语言查询框架采用分层架构设计确保各模块职责清晰、易于扩展用户界面层 → 自然语言处理层 → 查询生成层 → 执行验证层 → 数据源层各层核心职责用户界面层接收自然语言查询展示查询结果自然语言处理层使用 LLM 进行意图识别和语义解析查询生成层根据解析结果构建目标查询语句执行验证层验证查询安全性并执行处理异常数据源层提供领域元数据访问接口2.2 核心技术组件选型基于当前技术生态和实用性考虑我们选择以下技术栈LLM 服务选项OpenAI GPT 系列效果稳定API 成熟适合生产环境开源模型Llama、ChatGLM数据隐私性好可本地部署Azure OpenAI企业级安全和服务等级协议后端框架FastAPI高性能 Python Web 框架自动生成 API 文档SQLAlchemy数据库 ORM支持多种数据库方言Pydantic数据验证和设置管理元数据管理AmundsenLyft 开源的元数据发现平台自定义元数据存储基于业务需求设计的轻量级方案2.3 元数据模型设计有效的元数据模型是框架成功的基础。我们设计一个灵活的元数据模型来支持多种业务场景# 元数据核心模型定义 from typing import List, Dict, Optional from pydantic import BaseModel class ColumnMetadata(BaseModel): name: str data_type: str description: str business_meaning: str sensitivity_level: str # 敏感级别public, internal, confidential class TableMetadata(BaseModel): name: str description: str columns: List[ColumnMetadata] business_owner: str data_domain: str # 数据域sales, finance, hr等 class DataDomain(BaseModel): name: str description: str tables: List[str] allowed_query_patterns: List[str] # 允许的查询模式这种模型设计确保了元数据不仅包含技术信息还融入了业务语义和访问控制策略。3. 环境准备与依赖配置3.1 开发环境要求确保你的开发环境满足以下要求操作系统Windows 10/11, macOS 10.15, Ubuntu 18.04Python 版本3.8-3.11推荐 3.9内存至少 8GB RAM网络可访问 LLM API如需要3.2 项目依赖安装创建项目目录并安装核心依赖# 创建项目目录 mkdir nlq-framework cd nlq-framework # 创建虚拟环境 python -m venv venv source venv/bin/activate # Linux/macOS # venv\Scripts\activate # Windows # 安装核心依赖 pip install fastapi uvicorn sqlalchemy pydantic openai python-dotenv对于使用开源 LLM 的场景额外安装# 使用 transformers 库加载本地模型 pip install transformers torch accelerate # 或者使用 Llama.cpp 进行优化推理 pip install llama-cpp-python3.3 配置文件设置创建环境配置文件.env# LLM 配置 OPENAI_API_KEYyour_openai_api_key_here LLM_MODELgpt-3.5-turbo LLM_TEMPERATURE0.1 # 数据库配置 DATABASE_URLpostgresql://user:passwordlocalhost/metadata_db # 应用配置 MAX_QUERY_RESULTS1000 DEFAULT_TIMEOUT30 # 安全配置 ALLOWED_DATA_DOMAINSsales,finance,marketing BLACKLISTED_KEYWORDSdelete,drop,truncate创建配置管理模块config.pyimport os from pydantic import BaseSettings class Settings(BaseSettings): openai_api_key: str llm_model: str gpt-3.5-turbo database_url: str max_query_results: int 1000 class Config: env_file .env settings Settings()4. 核心模块实现详解4.1 元数据管理模块元数据管理器负责加载、缓存和提供领域元数据信息import json from typing import Dict, List from sqlalchemy import create_engine, text class MetadataManager: def __init__(self, database_url: str): self.engine create_engine(database_url) self.metadata_cache: Dict[str, Dict] {} def load_table_metadata(self, table_name: str) - Dict: 加载指定表的元数据 if table_name in self.metadata_cache: return self.metadata_cache[table_name] # 从数据库查询表结构信息 query text( SELECT column_name, data_type, description FROM information_schema.columns WHERE table_name :table_name ) with self.engine.connect() as conn: result conn.execute(query, {table_name: table_name}) columns [dict(row) for row in result] metadata { table_name: table_name, columns: columns, sample_data: self._get_sample_data(table_name) } self.metadata_cache[table_name] metadata return metadata def _get_sample_data(self, table_name: str, limit: int 5) - List[Dict]: 获取样本数据用于理解数据结构 query text(fSELECT * FROM {table_name} LIMIT {limit}) with self.engine.connect() as conn: result conn.execute(query) return [dict(row) for row in result] def get_domain_schema(self, domain: str) - str: 获取指定领域的模式描述用于提示词工程 tables self._get_domain_tables(domain) schema_description f数据域 {domain} 包含以下表\n for table in tables: metadata self.load_table_metadata(table) schema_description f\n表 {table}:\n for col in metadata[columns]: schema_description f - {col[column_name]} ({col[data_type]}): {col[description]}\n return schema_description4.2 LLM 查询解析模块查询解析器使用 LLM 将自然语言转换为结构化查询意图import openai from typing import Dict, Any import json class QueryParser: def __init__(self, api_key: str, model: str gpt-3.5-turbo): self.client openai.OpenAI(api_keyapi_key) self.model model def parse_natural_language(self, query: str, domain_schema: str) - Dict[str, Any]: 解析自然语言查询为结构化意图 prompt self._build_parsing_prompt(query, domain_schema) response self.client.chat.completions.create( modelself.model, messages[{role: user, content: prompt}], temperature0.1, max_tokens500 ) result_text response.choices[0].message.content return self._parse_json_response(result_text) def _build_parsing_prompt(self, query: str, domain_schema: str) - str: 构建解析提示词 return f 你是一个专业的数据查询解析器。请将用户的自然语言查询解析为结构化JSON格式。 领域数据模式 {domain_schema} 用户查询{query} 请输出以下JSON格式 {{ intent: 查询类型data_retrieval, aggregation, trend_analysis等, target_tables: [涉及的主要表名], selected_columns: [需要查询的字段], filter_conditions: [ {{ column: 字段名, operator: 操作符, , , LIKE等, value: 条件值, logical_operator: AND/OR }} ], aggregations: [ {{ type: 聚合类型COUNT, SUM, AVG等, column: 聚合字段, alias: 结果别名 }} ], sorting: [ {{ column: 排序字段, direction: ASC/DESC }} ], time_range: {{ column: 时间字段, start: 开始时间, end: 结束时间 }}, limit: 结果数量限制 }} 请确保 1. 只使用领域数据模式中存在的表和字段 2. 对模糊的时间描述进行具体化如最近一周转换为具体日期 3. 为聚合字段设置有意义的别名 4. 如果用户查询无法理解返回错误信息 def _parse_json_response(self, response_text: str) - Dict[str, Any]: 解析LLM的JSON响应 try: # 提取JSON部分LLM可能在响应中添加说明文字 start_idx response_text.find({) end_idx response_text.rfind(}) 1 json_str response_text[start_idx:end_idx] return json.loads(json_str) except json.JSONDecodeError as e: return {error: fJSON解析失败: {str(e)}, raw_response: response_text}4.3 SQL 生成器模块SQL 生成器将结构化意图转换为可执行的 SQL 查询from typing import Dict, List, Any class SQLGenerator: def __init__(self, metadata_manager: MetadataManager): self.metadata_manager metadata_manager def generate_sql(self, parsed_intent: Dict[str, Any]) - str: 根据解析意图生成SQL查询 if error in parsed_intent: raise ValueError(f解析失败: {parsed_intent[error]}) # 验证表和字段存在性 self._validate_tables_columns(parsed_intent) # 构建SQL各部分 select_clause self._build_select_clause(parsed_intent) from_clause self._build_from_clause(parsed_intent) where_clause self._build_where_clause(parsed_intent) group_by_clause self._build_group_by_clause(parsed_intent) order_by_clause self._build_order_by_clause(parsed_intent) limit_clause self._build_limit_clause(parsed_intent) # 组合完整SQL sql_parts [select_clause, from_clause] if where_clause: sql_parts.append(where_clause) if group_by_clause: sql_parts.append(group_by_clause) if order_by_clause: sql_parts.append(order_by_clause) if limit_clause: sql_parts.append(limit_clause) return .join(sql_parts) def _validate_tables_columns(self, parsed_intent: Dict): 验证涉及的表和字段是否存在 for table in parsed_intent.get(target_tables, []): if table not in self.metadata_manager.metadata_cache: self.metadata_manager.load_table_metadata(table) table_metadata self.metadata_manager.metadata_cache[table] available_columns [col[column_name] for col in table_metadata[columns]] # 验证查询字段 for column in parsed_intent.get(selected_columns, []): if column not in available_columns: raise ValueError(f字段 {column} 在表 {table} 中不存在) def _build_select_clause(self, parsed_intent: Dict) - str: 构建SELECT子句 columns parsed_intent.get(selected_columns, []) aggregations parsed_intent.get(aggregations, []) select_items [] # 添加普通字段 select_items.extend(columns) # 添加聚合字段 for agg in aggregations: agg_type agg[type] column agg[column] alias agg.get(alias, f{agg_type.lower()}_{column}) select_items.append(f{agg_type}({column}) AS {alias}) if not select_items: select_items [*] # 默认选择所有字段 return fSELECT {, .join(select_items)} def _build_where_clause(self, parsed_intent: Dict) - str: 构建WHERE子句 conditions parsed_intent.get(filter_conditions, []) time_range parsed_intent.get(time_range, {}) where_parts [] # 添加普通条件 for condition in conditions: column condition[column] operator condition[operator] value condition[value] logical_op condition.get(logical_operator, AND) # 处理字符串值 if isinstance(value, str) and operator ! IN: value f{value} where_parts.append(f{column} {operator} {value}) # 添加时间范围条件 if time_range: time_column time_range[column] start_time time_range.get(start) end_time time_range.get(end) if start_time: where_parts.append(f{time_column} {start_time}) if end_time: where_parts.append(f{time_column} {end_time}) if where_parts: return fWHERE { AND .join(where_parts)} return 4.4 查询验证与安全模块安全模块确保生成的查询符合权限要求和语法规范import re from typing import List class QueryValidator: def __init__(self, allowed_domains: List[str], blacklisted_keywords: List[str]): self.allowed_domains allowed_domains self.blacklisted_keywords blacklisted_keywords def validate_query_safety(self, sql: str, user_domain: str) - bool: 验证查询安全性 # 检查黑名单关键词 for keyword in self.blacklisted_keywords: if re.search(rf\b{keyword}\b, sql, re.IGNORECASE): raise SecurityError(f查询包含禁止的关键词: {keyword}) # 检查数据域权限 if not self._check_domain_permission(sql, user_domain): raise SecurityError(f用户无权访问查询中的数据域) # 检查查询复杂度 if self._is_query_too_complex(sql): raise SecurityError(查询过于复杂可能影响系统性能) return True def _check_domain_permission(self, sql: str, user_domain: str) - bool: 检查用户是否有权访问查询中的表 # 提取查询中涉及的表名 table_pattern r\b(FROM|JOIN)\s(\w) tables re.findall(table_pattern, sql, re.IGNORECASE) # 在实际实现中这里应该检查表所属的数据域 # 简化实现假设用户只能访问自己域的表格 for _, table in tables: if not table.startswith(user_domain _): return False return True def _is_query_too_complex(self, sql: str) - bool: 简单检查查询复杂度 complexity_score 0 complexity_score sql.count(JOIN) * 2 complexity_score sql.count(SUBSTRING) * 3 complexity_score sql.count(CASE WHEN) * 2 return complexity_score 10 # 可调整的阈值 class SecurityError(Exception): 自定义安全异常 pass5. 完整实战案例销售数据分析系统5.1 案例背景与数据准备假设我们有一个电商销售数据分析系统包含以下核心表-- 销售事实表 CREATE TABLE sales_fact ( sale_id BIGINT PRIMARY KEY, product_id VARCHAR(50), sale_date DATE, sale_amount DECIMAL(10,2), quantity INT, customer_id VARCHAR(50), region VARCHAR(50) ); -- 产品维度表 CREATE TABLE product_dim ( product_id VARCHAR(50) PRIMARY KEY, product_name VARCHAR(100), category VARCHAR(50), price DECIMAL(8,2) ); -- 客户维度表 CREATE TABLE customer_dim ( customer_id VARCHAR(50) PRIMARY KEY, customer_name VARCHAR(100), segment VARCHAR(50) );插入样本数据INSERT INTO product_dim VALUES (P001, 智能手机, 电子产品, 2999.00), (P002, 笔记本电脑, 电子产品, 5999.00), (P003, 办公椅, 家具, 899.00); INSERT INTO sales_fact VALUES (1, P001, 2024-01-15, 2999.00, 1, C001, 北京), (2, P002, 2024-01-16, 5999.00, 1, C002, 上海), (3, P001, 2024-01-17, 5998.00, 2, C003, 广州);5.2 自然语言查询处理流程现在让我们演示完整的自然语言查询处理流程# 主业务流程整合 from fastapi import FastAPI, HTTPException from pydantic import BaseModel app FastAPI(title自然语言查询框架) class QueryRequest(BaseModel): natural_language_query: str user_domain: str class QueryResponse(BaseModel): generated_sql: str execution_result: List[Dict] confidence_score: float app.post(/query, response_modelQueryResponse) async def process_natural_language_query(request: QueryRequest): 处理自然语言查询端点 try: # 1. 初始化各模块 metadata_manager MetadataManager(settings.database_url) query_parser QueryParser(settings.openai_api_key) sql_generator SQLGenerator(metadata_manager) validator QueryValidator(settings.allowed_data_domains, settings.blacklisted_keywords) # 2. 获取领域模式 domain_schema metadata_manager.get_domain_schema(request.user_domain) # 3. 解析自然语言 parsed_intent query_parser.parse_natural_language( request.natural_language_query, domain_schema ) # 4. 生成SQL generated_sql sql_generator.generate_sql(parsed_intent) # 5. 验证安全性 validator.validate_query_safety(generated_sql, request.user_domain) # 6. 执行查询简化演示 execution_result execute_safe_query(generated_sql, settings.max_query_results) return QueryResponse( generated_sqlgenerated_sql, execution_resultexecution_result, confidence_scorecalculate_confidence(parsed_intent) ) except Exception as e: raise HTTPException(status_code400, detailstr(e)) def execute_safe_query(sql: str, max_results: int) - List[Dict]: 安全执行SQL查询 # 在实际实现中这里应该使用参数化查询防止SQL注入 # 并添加查询超时和结果限制 limited_sql f{sql} LIMIT {max_results} # 使用SQLAlchemy执行查询 engine create_engine(settings.database_url) with engine.connect() as conn: result conn.execute(text(limited_sql)) return [dict(row) for row in result] def calculate_confidence(parsed_intent: Dict) - float: 计算查询解析的置信度 # 基于解析结果的完整性计算置信度 base_score 0.7 # 基础分数 if parsed_intent.get(target_tables): base_score 0.1 if parsed_intent.get(filter_conditions): base_score 0.1 if parsed_intent.get(time_range): base_score 0.1 return min(base_score, 1.0) # 不超过1.05.3 实际查询示例演示启动服务后我们可以测试各种自然语言查询示例1基本销售查询curl -X POST http://localhost:8000/query \ -H Content-Type: application/json \ -d { natural_language_query: 显示最近一个月销售额最高的10个产品, user_domain: sales }预期生成的SQLSELECT product_name, SUM(sale_amount) AS total_sales FROM sales_fact JOIN product_dim ON sales_fact.product_id product_dim.product_id WHERE sale_date 2024-01-01 GROUP BY product_name ORDER BY total_sales DESC LIMIT 10示例2复杂分析查询curl -X POST http://localhost:8000/query \ -H Content-Type: application/json \ -d { natural_language_query: 分析不同客户分段的平均订单金额和购买频率, user_domain: sales }预期生成的SQLSELECT customer_dim.segment, AVG(sales_fact.sale_amount) AS avg_order_value, COUNT(sales_fact.sale_id) AS order_count FROM sales_fact JOIN customer_dim ON sales_fact.customer_id customer_dim.customer_id GROUP BY customer_dim.segment6. 性能优化与生产部署6.1 缓存策略优化在生产环境中合理的缓存策略可以显著提升性能import redis from functools import lru_cache import hashlib class CachingQueryProcessor: def __init__(self, redis_url: str): self.redis_client redis.from_url(redis_url) def _get_query_hash(self, query: str, domain: str) - str: 生成查询哈希作为缓存键 content f{query}:{domain} return hashlib.md5(content.encode()).hexdigest() lru_cache(maxsize1000) def process_query_with_cache(self, query: str, domain: str) - Dict: 带缓存的查询处理 cache_key fnlq_cache:{self._get_query_hash(query, domain)} # 尝试从Redis获取缓存结果 cached_result self.redis_client.get(cache_key) if cached_result: return json.loads(cached_result) # 缓存未命中执行正常处理流程 result self._process_query(query, domain) # 缓存结果有效期1小时 self.redis_client.setex(cache_key, 3600, json.dumps(result)) return result6.2 LLM 调用优化减少 LLM API 调用成本和延迟class OptimizedLLMClient: def __init__(self, api_key: str): self.client openai.OpenAI(api_keyapi_key) self.request_cache {} def batch_process_queries(self, queries: List[str]) - List[Dict]: 批量处理相似查询 # 对查询进行聚类相似查询使用相同提示词 clustered_queries self._cluster_similar_queries(queries) results [] for cluster in clustered_queries: if len(cluster) 1: # 批量处理相似查询 batch_result self._process_batch(cluster) results.extend(batch_result) else: # 单个查询处理 results.append(self.process_single_query(cluster[0])) return results def _cluster_similar_queries(self, queries: List[str]) - List[List[str]]: 基于语义相似度对查询进行聚类 # 使用简单的文本相似度算法或嵌入模型 # 简化实现按查询长度和关键词聚类 clusters {} for query in queries: key self._extract_query_keywords(query) if key not in clusters: clusters[key] [] clusters[key].append(query) return list(clusters.values())6.3 生产环境配置创建生产环境配置文件docker-compose.prod.ymlversion: 3.8 services: nlq-api: build: . ports: - 8000:8000 environment: - ENVIRONMENTproduction - REDIS_URLredis://redis:6379 - DATABASE_URLpostgresql://user:passpostgres:5432/nlq_db depends_on: - redis - postgres redis: image: redis:7-alpine ports: - 6379:6379 postgres: image: postgres:13 environment: - POSTGRES_DBnlq_db - POSTGRES_USERuser - POSTGRES_PASSWORDpass volumes: - postgres_data:/var/lib/postgresql/data volumes: postgres_data:7. 常见问题与解决方案7.1 查询解析错误处理问题现象可能原因解决方案LLM返回非JSON格式提示词不够明确或模型理解偏差优化提示词添加JSON格式示例实现重试机制查询涉及不存在的表/字段元数据未及时更新或用户表述错误实现元数据验证提供字段建议功能时间范围解析错误自然语言时间描述模糊使用更精确的时间解析库要求用户确认错误处理代码示例class RobustQueryProcessor: def process_with_fallback(self, query: str, max_retries: int 3) - Dict: 带重试和降级的查询处理 for attempt in range(max_retries): try: return self._process_query(query) except openai.APIError as e: if attempt max_retries - 1: # 最后一次尝试失败使用降级方案 return self._fallback_processing(query) else: # 等待后重试 time.sleep(2 ** attempt) # 指数退避 except json.JSONDecodeError as e: # JSON解析错误尝试修复或使用备用解析器 fixed_result self._fix_json_response(e.response_text) if fixed_result: return fixed_result return self._fallback_processing(query)7.2 性能问题排查常见性能瓶颈及优化方案LLM API 延迟过高方案实现请求批处理、使用连接池、设置合理超时元数据查询缓慢方案建立元数据缓存、使用增量更新策略复杂查询执行超时方案添加查询超时控制、限制结果集大小、使用异步执行性能监控实现import time from prometheus_client import Counter, Histogram, generate_latest # 定义监控指标 QUERY_COUNTER Counter(nlq_queries_total, Total queries, [status]) QUERY_DURATION Histogram(nlq_query_duration_seconds, Query processing duration) class MonitoredQueryProcessor: QUERY_DURATION.time() def process_query(self, query: str) - Dict: start_time time.time() try: result self._internal_process(query) QUERY_COUNTER.labels(statussuccess).inc() return result except Exception as e: QUERY_COUNTER.labels(statuserror).inc() raise e finally: duration time.time() - start_time if duration 5.0: # 慢查询日志 self._log_slow_query(query, duration)8. 最佳实践与工程建议8.1 提示词工程优化有效的提示词设计是提升解析准确性的关键优质提示词特征明确角色定义清晰说明LLM扮演的角色结构化输出要求指定准确的JSON格式领域知识注入提供充分的业务上下文错误处理指导说明如何处理模糊或无法处理的查询提示词模板优化示例你是一个专业的{specific_domain}数据分析助手。请将用户的自然语言查询转换为结构化查询意图。 可用数据资源 {domain_schema} 查询处理规则 1. 只使用上述模式中存在的表和字段 2. 对时间描述进行标准化如上周转换为具体日期范围 3. 为聚合字段设置有意义的业务别名 4. 如果查询模糊请求用户澄清而非猜测 输出格式严格JSON {output_schema}8.2 安全防护体系构建多层次的安全防护class SecurityEnforcer: def __init__(self): self.validators [ SqlInjectionValidator(), DataDomainValidator(), QueryComplexityValidator(), SensitiveFieldValidator() ] def enforce_security(self, query: str, user_context: Dict) - bool: 执行多层次安全验证 violations [] for validator in self.validators: try: validator.validate(query, user_context) except SecurityViolation as e: violations.append(str(e)) if violations: raise SecurityError(f安全验证失败: {, .join(violations)}) return True class SqlInjectionValidator: def validate(self, query: str, user_context: Dict): SQL注入检测 # 使用正则表达式检测常见注入模式 injection_patterns [ r(\bUNION\b.*\bSELECT\b), r(\bDROP\b|\bDELETE\b|\bINSERT\b).*(\bTABLE\b|\bFROM\b), r(\bOR\b.*.*\bOR\b), r(--|\#|\/\*)\s ] for pattern in injection_patterns: if re.search(pattern, query, re.IGNORECASE): raise SecurityViolation(检测到可能的SQL注入模式)8.3 可扩展性设计确保框架能够适应不断变化的业务需求插件化架构设计from abc import ABC, abstractmethod from typing import Dict, Any class QueryParserPlugin(ABC): 查询解析插件基类 abstractmethod def supports_domain(self, domain: str) - bool: pass abstractmethod def parse(self, query: str, context: Dict) - Dict[str, Any]: pass class PluginManager: def __init__(self): self.plugins: List[QueryParserPlugin] [] def register_plugin(self, plugin: QueryParserPlugin): self.plugins.append(plugin) def get_parser_for_domain(self, domain: str) - QueryParserPlugin: for plugin in self.plugins: if plugin.supports_domain(domain): return plugin raise ValueError(f没有找到支持域 {domain} 的解析器) # 具体领域插件实现 class SalesQueryParser(QueryParserPlugin): def supports_domain(self, domain: str) - bool: return domain in [sales, ecommerce] def parse(self, query: str, context: Dict) - Dict[str, Any]: # 销售领域特定的解析逻辑