在数字化转型浪潮中企业积累了大量领域特定的元数据但非技术背景的业务人员往往难以直接利用这些数据。传统的数据查询方式需要掌握SQL等专业语言形成了较高的使用门槛。最近在项目中探索了一种基于大语言模型的自然语言查询生成框架能够将普通用户的自然语言问题自动转换为结构化的数据查询语句显著降低了数据访问的技术门槛。本文将详细介绍如何构建一个可复用的自然语言到领域元数据查询的LLM框架涵盖核心概念、架构设计、完整实现代码以及实际部署方案。无论你是数据工程师、全栈开发者还是技术负责人都能从中获得可直接复用的技术方案。1. 自然语言查询框架的核心概念1.1 什么是自然语言到结构化查询的转换自然语言到结构化查询的转换NL-to-Query是指将用户用日常语言提出的数据需求自动转换为计算机可执行的结构化查询语言如SQL、SPARQL等的过程。这种技术的关键价值在于打破了业务需求与技术实现之间的语言障碍。传统的数据访问流程中业务人员需要向技术人员描述需求技术人员再编写相应的查询语句。而NL-to-Query框架能够实现端到端的自动化转换大大提升了数据访问的效率和易用性。1.2 领域特定元数据的特点与挑战领域特定元数据是指某个专业领域内描述数据特征的信息比如电商领域的商品分类、库存状态医疗领域的病历结构、诊断代码等。这类元数据通常具有以下特点结构化程度高有明确的schema定义和关系模型专业术语密集包含大量领域特有的概念和缩写访问权限复杂不同角色对数据的访问权限差异很大查询模式固定业务场景下的查询需求相对集中但变化频繁这些特点给自然语言查询带来了独特挑战需要准确理解领域术语、正确处理业务逻辑约束、确保查询结果的安全性和准确性。1.3 LLM在查询生成中的角色定位大语言模型LLM在这个框架中扮演着语义理解逻辑转换的双重角色。一方面LLM需要理解用户自然语言中的真实意图另一方面它要将这种意图映射到具体的数据库结构和查询逻辑上。重要的是要认识到LLM的局限性它不直接访问数据库也不执行查询而是生成正确的查询语句。这种设计既保证了数据安全又充分发挥了LLM的语言理解能力。2. 框架架构设计与技术选型2.1 整体系统架构我们的可复用框架采用分层设计确保各组件职责清晰、易于扩展用户界面层 → 自然语言处理层 → 查询生成层 → 查询执行层 → 结果展示层每一层都通过标准接口与相邻层通信支持独立升级和替换。这种设计使得框架能够适应不同的业务场景和技术栈。2.2 核心组件技术选型基于当前技术生态的成熟度和社区支持我们推荐以下技术组合LLM引擎OpenAI GPT系列或开源替代品如Llama、ChatGLM向量数据库Pinecone、Chroma或Milvus用于元数据语义检索传统数据库根据业务需求选择PostgreSQL、MySQL或专用图数据库后端框架FastAPI或Spring Boot提供RESTful API接口前端界面Streamlit或Vue.js构建交互式查询界面这种技术组合在性能、成本和可维护性之间取得了良好平衡且都有活跃的社区支持。2.3 元数据管理策略有效的元数据管理是框架成功的关键。我们采用三层元数据管理策略结构元数据数据库表结构、字段类型、关系约束语义元数据业务术语与数据库字段的映射关系权限元数据数据访问权限规则和业务逻辑约束这些元数据以JSON Schema格式存储既便于LLM理解也方便人工维护和更新。3. 环境准备与依赖配置3.1 开发环境要求确保你的开发环境满足以下要求Python 3.8 运行环境至少8GB可用内存稳定的网络连接用于访问LLM API数据库环境本地或远程3.2 核心依赖安装创建新的Python虚拟环境并安装所需依赖# 创建并激活虚拟环境 python -m venv nl2query-env source nl2query-env/bin/activate # Linux/Mac # nl2query-env\Scripts\activate # Windows # 安装核心依赖 pip install openai langchain sqlalchemy psycopg2-binary pip install fastapi uvicorn streamlit pip install python-dotenv requests3.3 配置文件设置创建配置文件.env包含必要的环境变量# .env 配置文件 OPENAI_API_KEYyour_openai_api_key_here DATABASE_URLpostgresql://user:passwordlocalhost:5432/your_database VECTOR_DB_URLyour_vector_db_connection_string LOG_LEVELINFO MAX_QUERY_RESULTS1000对应的配置读取类# config.py import os from dotenv import load_dotenv load_dotenv() class Config: OPENAI_API_KEY os.getenv(OPENAI_API_KEY) DATABASE_URL os.getenv(DATABASE_URL) VECTOR_DB_URL os.getenv(VECTOR_DB_URL) MAX_QUERY_RESULTS int(os.getenv(MAX_QUERY_RESULTS, 1000)) classmethod def validate(cls): 验证必要配置是否完整 if not cls.OPENAI_API_KEY: raise ValueError(OPENAI_API_KEY必须配置) if not cls.DATABASE_URL: raise ValueError(DATABASE_URL必须配置)4. 元数据采集与向量化处理4.1 数据库元数据自动提取实现自动化的数据库元数据采集支持多种数据库类型# metadata_extractor.py from sqlalchemy import create_engine, inspect import json from typing import Dict, List class MetadataExtractor: def __init__(self, database_url: str): self.engine create_engine(database_url) self.inspector inspect(self.engine) def extract_schema_metadata(self) - Dict: 提取完整的数据库schema元数据 metadata { tables: [], relationships: [], constraints: [] } # 提取表信息 for table_name in self.inspector.get_table_names(): table_metadata { name: table_name, columns: [], primary_key: self.inspector.get_pk_constraint(table_name) } # 提取列信息 for column in self.inspector.get_columns(table_name): column_info { name: column[name], type: str(column[type]), nullable: column[nullable], default: column.get(default) } table_metadata[columns].append(column_info) # 提取外键关系 foreign_keys self.inspector.get_foreign_keys(table_name) for fk in foreign_keys: relationship { from_table: table_name, from_column: fk[constrained_columns][0], to_table: fk[referred_table], to_column: fk[referred_columns][0] } metadata[relationships].append(relationship) metadata[tables].append(table_metadata) return metadata def save_metadata(self, file_path: str): 保存元数据到文件 metadata self.extract_schema_metadata() with open(file_path, w, encodingutf-8) as f: json.dump(metadata, f, indent2, ensure_asciiFalse) # 使用示例 if __name__ __main__: extractor MetadataExtractor(postgresql://user:passlocalhost/db) extractor.save_metadata(database_metadata.json)4.2 元数据向量化存储将文本元数据转换为向量表示便于语义检索# vector_store.py import chromadb from chromadb.config import Settings from sentence_transformers import SentenceTransformer import json class MetadataVectorStore: def __init__(self, persist_directory: str ./chroma_db): self.client chromadb.Client(Settings( chroma_db_implduckdbparquet, persist_directorypersist_directory )) self.encoder SentenceTransformer(all-MiniLM-L6-v2) self.collection self.client.get_or_create_collection(metadata) def add_metadata(self, metadata_file: str): 将元数据添加到向量数据库 with open(metadata_file, r, encodingutf-8) as f: metadata json.load(f) documents [] metadatas [] ids [] # 处理表级元数据 for i, table in enumerate(metadata[tables]): table_text f表{table[name]}包含列{, .join(col[name] for col in table[columns])} documents.append(table_text) metadatas.append({type: table, table_name: table[name]}) ids.append(ftable_{i}) # 处理列级元数据 for j, column in enumerate(table[columns]): col_text f表{table[name]}的列{column[name]}类型为{column[type]} documents.append(col_text) metadatas.append({ type: column, table_name: table[name], column_name: column[name] }) ids.append(fcol_{i}_{j}) # 批量添加文档 embeddings self.encoder.encode(documents).tolist() self.collection.add( embeddingsembeddings, documentsdocuments, metadatasmetadatas, idsids ) def semantic_search(self, query: str, n_results: int 5): 语义搜索相关元数据 query_embedding self.encoder.encode([query]).tolist() results self.collection.query( query_embeddingsquery_embedding, n_resultsn_results ) return results5. LLM查询生成器实现5.1 提示词工程设计设计有效的提示词模板是查询生成准确性的关键# prompt_templates.py class PromptTemplates: staticmethod def get_sql_generation_prompt(user_query: str, schema_context: str) - str: 生成SQL查询的提示词模板 return f 你是一个专业的SQL查询生成器。请根据用户的问题生成准确的SQL查询语句。 数据库Schema信息 {schema_context} 用户问题{user_query} 请遵循以下规则 1. 只生成SQL语句不要额外解释 2. 使用准确的表名和列名 3. 包含必要的WHERE条件 4. 考虑性能避免SELECT * 5. 如果需要关联查询使用JOIN语句 生成的SQL语句 staticmethod def get_query_explanation_prompt(sql_query: str) - str: 生成查询解释的提示词模板 return f 请用通俗易懂的语言解释以下SQL查询的含义 {sql_query} 解释时请说明 1. 这个查询要获取什么数据 2. 涉及哪些表和字段 3. 查询条件是什么 4. 可能返回什么样的结果 请用中文回答 5.2 LLM查询生成核心逻辑实现基于LLM的查询生成核心组件# query_generator.py import openai from typing import Dict, Optional import logging class LLMQueryGenerator: def __init__(self, api_key: str, model: str gpt-3.5-turbo): openai.api_key api_key self.model model self.logger logging.getLogger(__name__) def generate_sql_query(self, user_query: str, schema_context: str) - Dict: 生成SQL查询语句 try: prompt PromptTemplates.get_sql_generation_prompt(user_query, schema_context) response openai.ChatCompletion.create( modelself.model, messages[ {role: system, content: 你是一个专业的SQL专家。}, {role: user, content: prompt} ], temperature0.1, # 低温度确保确定性 max_tokens500 ) sql_query response.choices[0].message.content.strip() # 清理SQL语句移除可能的代码块标记 if sql_query.startswith(sql): sql_query sql_query[6:] if sql_query.endswith(): sql_query sql_query[:-3] return { success: True, sql_query: sql_query, explanation: self._generate_explanation(sql_query) } except Exception as e: self.logger.error(f生成SQL查询时出错: {e}) return { success: False, error: str(e), sql_query: None, explanation: None } def _generate_explanation(self, sql_query: str) - str: 生成查询解释 try: prompt PromptTemplates.get_query_explanation_prompt(sql_query) response openai.ChatCompletion.create( modelself.model, messages[{role: user, content: prompt}], temperature0.7, max_tokens300 ) return response.choices[0].message.content.strip() except Exception as e: self.logger.error(f生成解释时出错: {e}) return 无法生成查询解释5.3 查询验证与优化实现查询语句的验证和优化机制# query_validator.py import sqlparse from sql_metadata import Parser from typing import Dict, List class QueryValidator: staticmethod def validate_sql_query(sql_query: str, allowed_tables: List[str]) - Dict: 验证SQL查询的合法性 try: # 解析SQL语句 parsed sqlparse.parse(sql_query) if not parsed: return {valid: False, error: 无法解析SQL语句} # 使用sql_metadata进行更深入的分析 parser Parser(sql_query) # 检查涉及的表是否在允许范围内 query_tables parser.tables unauthorized_tables set(query_tables) - set(allowed_tables) if unauthorized_tables: return { valid: False, error: f查询涉及未授权的表: {, .join(unauthorized_tables)} } # 检查是否有危险操作如DROP、DELETE等 dangerous_keywords [drop, delete, truncate, alter, create] statement parsed[0] if any(keyword in statement.value.lower() for keyword in dangerous_keywords): return {valid: False, error: 查询包含危险操作} return {valid: True, tables: query_tables, columns: parser.columns} except Exception as e: return {valid: False, error: f验证过程中出错: {str(e)}} staticmethod def optimize_query(sql_query: str) - str: 优化SQL查询语句 # 基本的查询优化逻辑 optimized sql_query # 移除不必要的空格和换行 optimized .join(optimized.split()) # 这里可以添加更多优化规则 # 比如确保使用索引友好的查询条件等 return optimized6. 完整系统集成与API设计6.1 FastAPI后端服务实现创建完整的RESTful API服务# main.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Optional import uvicorn from config import Config from query_generator import LLMQueryGenerator from query_validator import QueryValidator from vector_store import MetadataVectorStore app FastAPI(title自然语言查询生成框架, version1.0.0) # 数据模型定义 class QueryRequest(BaseModel): question: str user_id: Optional[str] None context: Optional[str] None class QueryResponse(BaseModel): success: bool sql_query: Optional[str] None explanation: Optional[str] None error: Optional[str] None execution_time: Optional[float] None # 全局组件初始化 config Config() query_generator LLMQueryGenerator(config.OPENAI_API_KEY) vector_store MetadataVectorStore() allowed_tables [sales, products, customers] # 示例允许的表 app.post(/generate-query, response_modelQueryResponse) async def generate_query(request: QueryRequest): 生成自然语言查询的SQL语句 try: # 1. 语义搜索相关元数据 schema_context vector_store.semantic_search(request.question) # 2. 生成SQL查询 start_time time.time() result query_generator.generate_sql_query(request.question, str(schema_context)) generation_time time.time() - start_time if result[success]: # 3. 验证查询安全性 validation_result QueryValidator.validate_sql_query( result[sql_query], allowed_tables ) if not validation_result[valid]: return QueryResponse( successFalse, errorf查询验证失败: {validation_result[error]} ) # 4. 优化查询 optimized_query QueryValidator.optimize_query(result[sql_query]) return QueryResponse( successTrue, sql_queryoptimized_query, explanationresult[explanation], execution_timegeneration_time ) else: return QueryResponse( successFalse, errorresult[error] ) except Exception as e: raise HTTPException(status_code500, detailf服务器内部错误: {str(e)}) app.get(/health) async def health_check(): 健康检查端点 return {status: healthy, version: 1.0.0} if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8000)6.2 Streamlit前端界面创建用户友好的查询界面# app.py import streamlit as st import requests import pandas as pd from datetime import datetime # 页面配置 st.set_page_config( page_title自然语言数据查询, page_icon, layoutwide ) # 样式定制 st.markdown( style .main-header { font-size: 2.5rem; color: #1f77b4; text-align: center; margin-bottom: 2rem; } .success-query { background-color: #d4edda; padding: 1rem; border-radius: 0.5rem; margin: 1rem 0; } .error-query { background-color: #f8d7da; padding: 1rem; border-radius: 0.5rem; margin: 1rem 0; } /style , unsafe_allow_htmlTrue) # 应用标题 st.markdown(div classmain-header 自然语言数据查询平台/div, unsafe_allow_htmlTrue) # 侧边栏配置 with st.sidebar: st.header(配置选项) api_endpoint st.text_input( API端点, valuehttp://localhost:8000, help后端服务的访问地址 ) max_results st.slider(最大返回结果数, 10, 1000, 100) # 主界面 col1, col2 st.columns([2, 1]) with col1: st.subheader(输入你的数据问题) user_question st.text_area( 用自然语言描述你的数据需求, placeholder例如显示最近一个月销售额最高的10个产品, height100 ) if st.button(生成查询, typeprimary): if user_question.strip(): with st.spinner(正在生成查询语句...): try: response requests.post( f{api_endpoint}/generate-query, json{question: user_question} ) if response.status_code 200: result response.json() if result[success]: st.markdown(div classsuccess-query✅ 查询生成成功/div, unsafe_allow_htmlTrue) # 显示生成的SQL st.subheader(生成的SQL查询) st.code(result[sql_query], languagesql) # 显示解释 st.subheader(查询解释) st.info(result[explanation]) # 这里可以添加查询执行和结果显示逻辑 else: st.markdown(fdiv classerror-query❌ 生成失败: {result[error]}/div, unsafe_allow_htmlTrue) else: st.error(fAPI请求失败: {response.status_code}) except requests.exceptions.RequestException as e: st.error(f连接后端服务失败: {e}) with col2: st.subheader(使用示例) examples [ 显示上季度每个地区的销售总额, 找出库存量低于安全库存的产品, 计算每个客户的平均订单金额, 最近一周销量最好的产品类别 ] for example in examples: if st.button(example, keyexample): st.session_state.user_question example # 历史记录功能 if query_history not in st.session_state: st.session_state.query_history [] if user_question and st.button(生成查询, typeprimary): st.session_state.query_history.append({ timestamp: datetime.now(), question: user_question, success: True # 简化示例 })7. 实际应用案例与测试7.1 电商数据分析场景以电商数据分析为例演示框架的实际应用-- 示例数据库结构 CREATE TABLE products ( product_id INT PRIMARY KEY, product_name VARCHAR(255), category VARCHAR(100), price DECIMAL(10,2), stock_quantity INT ); CREATE TABLE sales ( sale_id INT PRIMARY KEY, product_id INT, sale_date DATE, quantity INT, amount DECIMAL(10,2), FOREIGN KEY (product_id) REFERENCES products(product_id) ); CREATE TABLE customers ( customer_id INT PRIMARY KEY, customer_name VARCHAR(255), region VARCHAR(100), registration_date DATE );测试用例及生成结果# test_cases.py test_cases [ { question: 显示最近一个月销售额最高的10个产品, expected_tables: [sales, products], expected_columns: [product_name, total_sales] }, { question: 找出库存量低于100的产品并按类别分组, expected_tables: [products], expected_columns: [category, product_name, stock_quantity] }, { question: 计算每个地区客户的平均消费金额, expected_tables: [customers, sales], expected_columns: [region, avg_amount] } ] def run_test_cases(): 运行测试用例验证框架效果 for i, test_case in enumerate(test_cases, 1): print(f测试用例 {i}: {test_case[question]}) response query_generator.generate_sql_query( test_case[question], get_schema_context() ) if response[success]: validation QueryValidator.validate_sql_query( response[sql_query], test_case[expected_tables] ) if validation[valid]: print(✅ 测试通过) print(f生成的SQL: {response[sql_query]}) else: print(❌ 测试失败) print(f错误: {validation[error]}) else: print(❌ 生成失败) print(f错误: {response[error]}) print(- * 50)7.2 性能测试与优化实现性能监控和优化机制# performance_monitor.py import time from functools import wraps from typing import Dict, List import statistics class PerformanceMonitor: def __init__(self): self.metrics { query_generation_time: [], validation_time: [], total_processing_time: [] } def track_performance(self, metric_name: str): 性能追踪装饰器 def decorator(func): wraps(func) def wrapper(*args, **kwargs): start_time time.time() result func(*args, **kwargs) end_time time.time() execution_time end_time - start_time self.metrics[metric_name].append(execution_time) return result return wrapper return decorator def get_performance_report(self) - Dict: 生成性能报告 report {} for metric, times in self.metrics.items(): if times: report[metric] { count: len(times), avg_time: statistics.mean(times), max_time: max(times), min_time: min(times) } return report def check_sla_violations(self, sla_limits: Dict) - List[str]: 检查SLA违规情况 violations [] for metric, limit in sla_limits.items(): if metric in self.metrics and self.metrics[metric]: avg_time statistics.mean(self.metrics[metric]) if avg_time limit: violations.append(f{metric}平均时间{avg_time:.2f}s超过限制{limit}s) return violations # 使用示例 monitor PerformanceMonitor() monitor.track_performance(query_generation_time) def generate_query_with_monitoring(question: str, context: str): return query_generator.generate_sql_query(question, context)8. 安全考虑与权限控制8.1 查询安全防护机制实现多层次的安全防护# security_manager.py import re from typing import Set class SecurityManager: def __init__(self): self.dangerous_patterns [ r\b(drop|delete|truncate|alter|create|insert|update)\b, r;.*--, runion.*select, rexec\(|sp_, rxp_cmdshell ] self.allowed_functions { aggregate: [count, sum, avg, min, max], string: [concat, substring, lower, upper], date: [date_format, year, month, day] } def validate_query_safety(self, sql_query: str, user_role: str) - Dict: 全面验证查询安全性 safety_checks { syntax_check: self._check_sql_syntax(sql_query), dangerous_operations: self._check_dangerous_operations(sql_query), function_whitelist: self._check_function_whitelist(sql_query, user_role), data_access: self._check_data_access_patterns(sql_query, user_role) } # 汇总检查结果 all_passed all(safety_checks.values()) failed_checks [check for check, passed in safety_checks.items() if not passed] return { safe: all_passed, failed_checks: failed_checks, details: safety_checks } def _check_dangerous_operations(self, sql_query: str) - bool: 检查危险操作模式 query_lower sql_query.lower() for pattern in self.dangerous_patterns: if re.search(pattern, query_lower): return False return True def _check_function_whitelist(self, sql_query: str, user_role: str) - bool: 检查函数使用是否在允许范围内 # 根据用户角色确定允许的函数集合 if user_role analyst: allowed_functions self.allowed_functions[aggregate] self.allowed_functions[date] elif user_role admin: return True # 管理员无限制 else: allowed_functions self.allowed_functions[aggregate] # 提取SQL中的函数调用并进行验证 function_pattern r\b(\w)\s*\( used_functions re.findall(function_pattern, sql_query.lower()) for func in used_functions: if func not in allowed_functions and func not in [select, from, where]: return False return True8.2 基于角色的访问控制实现细粒度的权限管理# rbac_manager.py from typing import Dict, List, Set class RBACManager: def __init__(self): self.roles { viewer: { allowed_tables: {products, sales}, allowed_operations: {select}, max_rows: 1000, sensitive_columns: {password, salary} }, analyst: { allowed_tables: {products, sales, customers}, allowed_operations: {select, create_temp_table}, max_rows: 10000, sensitive_columns: {password} }, admin: { allowed_tables: {*}, allowed_operations: {*}, max_rows: None, sensitive_columns: set() } } def get_user_permissions(self, user_role: str) - Dict: 获取用户权限配置 return self.roles.get(user_role, self.roles[viewer]) def apply_query_restrictions(self, sql_query: str, user_role: str) - str: 根据用户角色应用查询限制 permissions self.get_user_permissions(user_role) # 这里可以添加具体的查询重写逻辑 # 比如添加行数限制、隐藏敏感列等 if permissions[max_rows]: # 为查询添加LIMIT子句如果还没有的话 if limit not in sql_query.lower(): sql_query f LIMIT {permissions[max_rows]} return sql_query def validate_table_access(self, tables: List[str], user_role: str) - bool: 验证用户对表的访问权限 permissions self.get_user_permissions(user_role) allowed_tables permissions[allowed_tables] if * in allowed_tables: return True return all(table in allowed_tables for table in tables)9. 部署与运维最佳实践9.1 Docker容器化部署创建完整的Docker部署配置# Dockerfile FROM python:3.9-slim WORKDIR /app # 安装系统依赖 RUN apt-get update apt-get install -y \ gcc \ rm -rf /var/lib/apt/lists/* # 复制依赖文件并安装 COPY requirements.txt . 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配置# docker-compose.yml version: 3.8 services: nl2query-api: build: . ports: - 8000:8000 environment: - OPENAI_API_KEY${OPENAI_API_KEY} - DATABASE_URL${DATABASE_URL} volumes: - ./logs:/app/logs healthcheck: test: [CMD, curl, -f, http://localhost:8000/health] interval: 30s timeout: 10s retries: 3 streamlit-ui: build: context: . dockerfile: Dockerfile.streamlit ports: - 8501:8501 depends_on: - nl2query-api environment: - API_ENDPOINThttp://nl2query-api:8000 # 可以添加数据库、缓存等依赖服务9.2 监控与日志配置实现完整的监控和日志体系# logging_config.py import logging import sys from pathlib import Path def setup_logging(log_level: str INFO, log_file: str app.log): 配置日志系统 # 创建日志目录 log_path Path(logs) log_path.mkdir(exist_okTrue) # 日志格式 formatter logging.Formatter( %(asctime)s - %(name)s - %(levelname)s - %(message)s ) # 根日志记录器 root_logger logging.getLogger() root_logger.setLevel(getattr(logging, log_level.upper())) # 控制台处理器 console_handler logging.StreamHandler(sys.stdout) console_handler.setFormatter(formatter) root_logger.addHandler(console_handler) # 文件处理器 file_handler logging.FileHandler(log_path / log_file) file_handler.setFormatter(formatter) root_logger.addHandler(file_handler) # 避免重复日志 root_logger.propagate False # 性能监控装饰器 def log_performance(operation_name: str): 性能日志装饰器 def decorator(func): wraps(func) def wrapper(*args, **kwargs): logger logging.getLogger(func.__module__) start_time time.time() try: result func(*args, **kwargs) execution_time time.time() - start_time logger.info( f{operation_name} completed in {execution_time:.2f}s, extra{operation: operation_name, duration: execution_time} ) return result except Exception as e: execution_time time.time() - start_time logger.error( f{operation_name} failed after {execution_time:.2f}s: {str(e)}, extra{operation: operation_name, error: str(e)} ) raise return wrapper return decorator10. 常见问题与解决方案10.1 查询生成准确性问题问题现象LLM生成的SQL查询语法正确但逻辑不符合业务需求。解决方案增强上下文提示提供更详细的业务规则和示例实现多轮对话允许用户澄清模糊需求
在数字化转型浪潮中企业积累了大量领域特定的元数据但非技术背景的业务人员往往难以直接利用这些数据。传统的数据查询方式需要掌握SQL等专业语言形成了较高的使用门槛。最近在项目中探索了一种基于大语言模型的自然语言查询生成框架能够将普通用户的自然语言问题自动转换为结构化的数据查询语句显著降低了数据访问的技术门槛。本文将详细介绍如何构建一个可复用的自然语言到领域元数据查询的LLM框架涵盖核心概念、架构设计、完整实现代码以及实际部署方案。无论你是数据工程师、全栈开发者还是技术负责人都能从中获得可直接复用的技术方案。1. 自然语言查询框架的核心概念1.1 什么是自然语言到结构化查询的转换自然语言到结构化查询的转换NL-to-Query是指将用户用日常语言提出的数据需求自动转换为计算机可执行的结构化查询语言如SQL、SPARQL等的过程。这种技术的关键价值在于打破了业务需求与技术实现之间的语言障碍。传统的数据访问流程中业务人员需要向技术人员描述需求技术人员再编写相应的查询语句。而NL-to-Query框架能够实现端到端的自动化转换大大提升了数据访问的效率和易用性。1.2 领域特定元数据的特点与挑战领域特定元数据是指某个专业领域内描述数据特征的信息比如电商领域的商品分类、库存状态医疗领域的病历结构、诊断代码等。这类元数据通常具有以下特点结构化程度高有明确的schema定义和关系模型专业术语密集包含大量领域特有的概念和缩写访问权限复杂不同角色对数据的访问权限差异很大查询模式固定业务场景下的查询需求相对集中但变化频繁这些特点给自然语言查询带来了独特挑战需要准确理解领域术语、正确处理业务逻辑约束、确保查询结果的安全性和准确性。1.3 LLM在查询生成中的角色定位大语言模型LLM在这个框架中扮演着语义理解逻辑转换的双重角色。一方面LLM需要理解用户自然语言中的真实意图另一方面它要将这种意图映射到具体的数据库结构和查询逻辑上。重要的是要认识到LLM的局限性它不直接访问数据库也不执行查询而是生成正确的查询语句。这种设计既保证了数据安全又充分发挥了LLM的语言理解能力。2. 框架架构设计与技术选型2.1 整体系统架构我们的可复用框架采用分层设计确保各组件职责清晰、易于扩展用户界面层 → 自然语言处理层 → 查询生成层 → 查询执行层 → 结果展示层每一层都通过标准接口与相邻层通信支持独立升级和替换。这种设计使得框架能够适应不同的业务场景和技术栈。2.2 核心组件技术选型基于当前技术生态的成熟度和社区支持我们推荐以下技术组合LLM引擎OpenAI GPT系列或开源替代品如Llama、ChatGLM向量数据库Pinecone、Chroma或Milvus用于元数据语义检索传统数据库根据业务需求选择PostgreSQL、MySQL或专用图数据库后端框架FastAPI或Spring Boot提供RESTful API接口前端界面Streamlit或Vue.js构建交互式查询界面这种技术组合在性能、成本和可维护性之间取得了良好平衡且都有活跃的社区支持。2.3 元数据管理策略有效的元数据管理是框架成功的关键。我们采用三层元数据管理策略结构元数据数据库表结构、字段类型、关系约束语义元数据业务术语与数据库字段的映射关系权限元数据数据访问权限规则和业务逻辑约束这些元数据以JSON Schema格式存储既便于LLM理解也方便人工维护和更新。3. 环境准备与依赖配置3.1 开发环境要求确保你的开发环境满足以下要求Python 3.8 运行环境至少8GB可用内存稳定的网络连接用于访问LLM API数据库环境本地或远程3.2 核心依赖安装创建新的Python虚拟环境并安装所需依赖# 创建并激活虚拟环境 python -m venv nl2query-env source nl2query-env/bin/activate # Linux/Mac # nl2query-env\Scripts\activate # Windows # 安装核心依赖 pip install openai langchain sqlalchemy psycopg2-binary pip install fastapi uvicorn streamlit pip install python-dotenv requests3.3 配置文件设置创建配置文件.env包含必要的环境变量# .env 配置文件 OPENAI_API_KEYyour_openai_api_key_here DATABASE_URLpostgresql://user:passwordlocalhost:5432/your_database VECTOR_DB_URLyour_vector_db_connection_string LOG_LEVELINFO MAX_QUERY_RESULTS1000对应的配置读取类# config.py import os from dotenv import load_dotenv load_dotenv() class Config: OPENAI_API_KEY os.getenv(OPENAI_API_KEY) DATABASE_URL os.getenv(DATABASE_URL) VECTOR_DB_URL os.getenv(VECTOR_DB_URL) MAX_QUERY_RESULTS int(os.getenv(MAX_QUERY_RESULTS, 1000)) classmethod def validate(cls): 验证必要配置是否完整 if not cls.OPENAI_API_KEY: raise ValueError(OPENAI_API_KEY必须配置) if not cls.DATABASE_URL: raise ValueError(DATABASE_URL必须配置)4. 元数据采集与向量化处理4.1 数据库元数据自动提取实现自动化的数据库元数据采集支持多种数据库类型# metadata_extractor.py from sqlalchemy import create_engine, inspect import json from typing import Dict, List class MetadataExtractor: def __init__(self, database_url: str): self.engine create_engine(database_url) self.inspector inspect(self.engine) def extract_schema_metadata(self) - Dict: 提取完整的数据库schema元数据 metadata { tables: [], relationships: [], constraints: [] } # 提取表信息 for table_name in self.inspector.get_table_names(): table_metadata { name: table_name, columns: [], primary_key: self.inspector.get_pk_constraint(table_name) } # 提取列信息 for column in self.inspector.get_columns(table_name): column_info { name: column[name], type: str(column[type]), nullable: column[nullable], default: column.get(default) } table_metadata[columns].append(column_info) # 提取外键关系 foreign_keys self.inspector.get_foreign_keys(table_name) for fk in foreign_keys: relationship { from_table: table_name, from_column: fk[constrained_columns][0], to_table: fk[referred_table], to_column: fk[referred_columns][0] } metadata[relationships].append(relationship) metadata[tables].append(table_metadata) return metadata def save_metadata(self, file_path: str): 保存元数据到文件 metadata self.extract_schema_metadata() with open(file_path, w, encodingutf-8) as f: json.dump(metadata, f, indent2, ensure_asciiFalse) # 使用示例 if __name__ __main__: extractor MetadataExtractor(postgresql://user:passlocalhost/db) extractor.save_metadata(database_metadata.json)4.2 元数据向量化存储将文本元数据转换为向量表示便于语义检索# vector_store.py import chromadb from chromadb.config import Settings from sentence_transformers import SentenceTransformer import json class MetadataVectorStore: def __init__(self, persist_directory: str ./chroma_db): self.client chromadb.Client(Settings( chroma_db_implduckdbparquet, persist_directorypersist_directory )) self.encoder SentenceTransformer(all-MiniLM-L6-v2) self.collection self.client.get_or_create_collection(metadata) def add_metadata(self, metadata_file: str): 将元数据添加到向量数据库 with open(metadata_file, r, encodingutf-8) as f: metadata json.load(f) documents [] metadatas [] ids [] # 处理表级元数据 for i, table in enumerate(metadata[tables]): table_text f表{table[name]}包含列{, .join(col[name] for col in table[columns])} documents.append(table_text) metadatas.append({type: table, table_name: table[name]}) ids.append(ftable_{i}) # 处理列级元数据 for j, column in enumerate(table[columns]): col_text f表{table[name]}的列{column[name]}类型为{column[type]} documents.append(col_text) metadatas.append({ type: column, table_name: table[name], column_name: column[name] }) ids.append(fcol_{i}_{j}) # 批量添加文档 embeddings self.encoder.encode(documents).tolist() self.collection.add( embeddingsembeddings, documentsdocuments, metadatasmetadatas, idsids ) def semantic_search(self, query: str, n_results: int 5): 语义搜索相关元数据 query_embedding self.encoder.encode([query]).tolist() results self.collection.query( query_embeddingsquery_embedding, n_resultsn_results ) return results5. LLM查询生成器实现5.1 提示词工程设计设计有效的提示词模板是查询生成准确性的关键# prompt_templates.py class PromptTemplates: staticmethod def get_sql_generation_prompt(user_query: str, schema_context: str) - str: 生成SQL查询的提示词模板 return f 你是一个专业的SQL查询生成器。请根据用户的问题生成准确的SQL查询语句。 数据库Schema信息 {schema_context} 用户问题{user_query} 请遵循以下规则 1. 只生成SQL语句不要额外解释 2. 使用准确的表名和列名 3. 包含必要的WHERE条件 4. 考虑性能避免SELECT * 5. 如果需要关联查询使用JOIN语句 生成的SQL语句 staticmethod def get_query_explanation_prompt(sql_query: str) - str: 生成查询解释的提示词模板 return f 请用通俗易懂的语言解释以下SQL查询的含义 {sql_query} 解释时请说明 1. 这个查询要获取什么数据 2. 涉及哪些表和字段 3. 查询条件是什么 4. 可能返回什么样的结果 请用中文回答 5.2 LLM查询生成核心逻辑实现基于LLM的查询生成核心组件# query_generator.py import openai from typing import Dict, Optional import logging class LLMQueryGenerator: def __init__(self, api_key: str, model: str gpt-3.5-turbo): openai.api_key api_key self.model model self.logger logging.getLogger(__name__) def generate_sql_query(self, user_query: str, schema_context: str) - Dict: 生成SQL查询语句 try: prompt PromptTemplates.get_sql_generation_prompt(user_query, schema_context) response openai.ChatCompletion.create( modelself.model, messages[ {role: system, content: 你是一个专业的SQL专家。}, {role: user, content: prompt} ], temperature0.1, # 低温度确保确定性 max_tokens500 ) sql_query response.choices[0].message.content.strip() # 清理SQL语句移除可能的代码块标记 if sql_query.startswith(sql): sql_query sql_query[6:] if sql_query.endswith(): sql_query sql_query[:-3] return { success: True, sql_query: sql_query, explanation: self._generate_explanation(sql_query) } except Exception as e: self.logger.error(f生成SQL查询时出错: {e}) return { success: False, error: str(e), sql_query: None, explanation: None } def _generate_explanation(self, sql_query: str) - str: 生成查询解释 try: prompt PromptTemplates.get_query_explanation_prompt(sql_query) response openai.ChatCompletion.create( modelself.model, messages[{role: user, content: prompt}], temperature0.7, max_tokens300 ) return response.choices[0].message.content.strip() except Exception as e: self.logger.error(f生成解释时出错: {e}) return 无法生成查询解释5.3 查询验证与优化实现查询语句的验证和优化机制# query_validator.py import sqlparse from sql_metadata import Parser from typing import Dict, List class QueryValidator: staticmethod def validate_sql_query(sql_query: str, allowed_tables: List[str]) - Dict: 验证SQL查询的合法性 try: # 解析SQL语句 parsed sqlparse.parse(sql_query) if not parsed: return {valid: False, error: 无法解析SQL语句} # 使用sql_metadata进行更深入的分析 parser Parser(sql_query) # 检查涉及的表是否在允许范围内 query_tables parser.tables unauthorized_tables set(query_tables) - set(allowed_tables) if unauthorized_tables: return { valid: False, error: f查询涉及未授权的表: {, .join(unauthorized_tables)} } # 检查是否有危险操作如DROP、DELETE等 dangerous_keywords [drop, delete, truncate, alter, create] statement parsed[0] if any(keyword in statement.value.lower() for keyword in dangerous_keywords): return {valid: False, error: 查询包含危险操作} return {valid: True, tables: query_tables, columns: parser.columns} except Exception as e: return {valid: False, error: f验证过程中出错: {str(e)}} staticmethod def optimize_query(sql_query: str) - str: 优化SQL查询语句 # 基本的查询优化逻辑 optimized sql_query # 移除不必要的空格和换行 optimized .join(optimized.split()) # 这里可以添加更多优化规则 # 比如确保使用索引友好的查询条件等 return optimized6. 完整系统集成与API设计6.1 FastAPI后端服务实现创建完整的RESTful API服务# main.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Optional import uvicorn from config import Config from query_generator import LLMQueryGenerator from query_validator import QueryValidator from vector_store import MetadataVectorStore app FastAPI(title自然语言查询生成框架, version1.0.0) # 数据模型定义 class QueryRequest(BaseModel): question: str user_id: Optional[str] None context: Optional[str] None class QueryResponse(BaseModel): success: bool sql_query: Optional[str] None explanation: Optional[str] None error: Optional[str] None execution_time: Optional[float] None # 全局组件初始化 config Config() query_generator LLMQueryGenerator(config.OPENAI_API_KEY) vector_store MetadataVectorStore() allowed_tables [sales, products, customers] # 示例允许的表 app.post(/generate-query, response_modelQueryResponse) async def generate_query(request: QueryRequest): 生成自然语言查询的SQL语句 try: # 1. 语义搜索相关元数据 schema_context vector_store.semantic_search(request.question) # 2. 生成SQL查询 start_time time.time() result query_generator.generate_sql_query(request.question, str(schema_context)) generation_time time.time() - start_time if result[success]: # 3. 验证查询安全性 validation_result QueryValidator.validate_sql_query( result[sql_query], allowed_tables ) if not validation_result[valid]: return QueryResponse( successFalse, errorf查询验证失败: {validation_result[error]} ) # 4. 优化查询 optimized_query QueryValidator.optimize_query(result[sql_query]) return QueryResponse( successTrue, sql_queryoptimized_query, explanationresult[explanation], execution_timegeneration_time ) else: return QueryResponse( successFalse, errorresult[error] ) except Exception as e: raise HTTPException(status_code500, detailf服务器内部错误: {str(e)}) app.get(/health) async def health_check(): 健康检查端点 return {status: healthy, version: 1.0.0} if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8000)6.2 Streamlit前端界面创建用户友好的查询界面# app.py import streamlit as st import requests import pandas as pd from datetime import datetime # 页面配置 st.set_page_config( page_title自然语言数据查询, page_icon, layoutwide ) # 样式定制 st.markdown( style .main-header { font-size: 2.5rem; color: #1f77b4; text-align: center; margin-bottom: 2rem; } .success-query { background-color: #d4edda; padding: 1rem; border-radius: 0.5rem; margin: 1rem 0; } .error-query { background-color: #f8d7da; padding: 1rem; border-radius: 0.5rem; margin: 1rem 0; } /style , unsafe_allow_htmlTrue) # 应用标题 st.markdown(div classmain-header 自然语言数据查询平台/div, unsafe_allow_htmlTrue) # 侧边栏配置 with st.sidebar: st.header(配置选项) api_endpoint st.text_input( API端点, valuehttp://localhost:8000, help后端服务的访问地址 ) max_results st.slider(最大返回结果数, 10, 1000, 100) # 主界面 col1, col2 st.columns([2, 1]) with col1: st.subheader(输入你的数据问题) user_question st.text_area( 用自然语言描述你的数据需求, placeholder例如显示最近一个月销售额最高的10个产品, height100 ) if st.button(生成查询, typeprimary): if user_question.strip(): with st.spinner(正在生成查询语句...): try: response requests.post( f{api_endpoint}/generate-query, json{question: user_question} ) if response.status_code 200: result response.json() if result[success]: st.markdown(div classsuccess-query✅ 查询生成成功/div, unsafe_allow_htmlTrue) # 显示生成的SQL st.subheader(生成的SQL查询) st.code(result[sql_query], languagesql) # 显示解释 st.subheader(查询解释) st.info(result[explanation]) # 这里可以添加查询执行和结果显示逻辑 else: st.markdown(fdiv classerror-query❌ 生成失败: {result[error]}/div, unsafe_allow_htmlTrue) else: st.error(fAPI请求失败: {response.status_code}) except requests.exceptions.RequestException as e: st.error(f连接后端服务失败: {e}) with col2: st.subheader(使用示例) examples [ 显示上季度每个地区的销售总额, 找出库存量低于安全库存的产品, 计算每个客户的平均订单金额, 最近一周销量最好的产品类别 ] for example in examples: if st.button(example, keyexample): st.session_state.user_question example # 历史记录功能 if query_history not in st.session_state: st.session_state.query_history [] if user_question and st.button(生成查询, typeprimary): st.session_state.query_history.append({ timestamp: datetime.now(), question: user_question, success: True # 简化示例 })7. 实际应用案例与测试7.1 电商数据分析场景以电商数据分析为例演示框架的实际应用-- 示例数据库结构 CREATE TABLE products ( product_id INT PRIMARY KEY, product_name VARCHAR(255), category VARCHAR(100), price DECIMAL(10,2), stock_quantity INT ); CREATE TABLE sales ( sale_id INT PRIMARY KEY, product_id INT, sale_date DATE, quantity INT, amount DECIMAL(10,2), FOREIGN KEY (product_id) REFERENCES products(product_id) ); CREATE TABLE customers ( customer_id INT PRIMARY KEY, customer_name VARCHAR(255), region VARCHAR(100), registration_date DATE );测试用例及生成结果# test_cases.py test_cases [ { question: 显示最近一个月销售额最高的10个产品, expected_tables: [sales, products], expected_columns: [product_name, total_sales] }, { question: 找出库存量低于100的产品并按类别分组, expected_tables: [products], expected_columns: [category, product_name, stock_quantity] }, { question: 计算每个地区客户的平均消费金额, expected_tables: [customers, sales], expected_columns: [region, avg_amount] } ] def run_test_cases(): 运行测试用例验证框架效果 for i, test_case in enumerate(test_cases, 1): print(f测试用例 {i}: {test_case[question]}) response query_generator.generate_sql_query( test_case[question], get_schema_context() ) if response[success]: validation QueryValidator.validate_sql_query( response[sql_query], test_case[expected_tables] ) if validation[valid]: print(✅ 测试通过) print(f生成的SQL: {response[sql_query]}) else: print(❌ 测试失败) print(f错误: {validation[error]}) else: print(❌ 生成失败) print(f错误: {response[error]}) print(- * 50)7.2 性能测试与优化实现性能监控和优化机制# performance_monitor.py import time from functools import wraps from typing import Dict, List import statistics class PerformanceMonitor: def __init__(self): self.metrics { query_generation_time: [], validation_time: [], total_processing_time: [] } def track_performance(self, metric_name: str): 性能追踪装饰器 def decorator(func): wraps(func) def wrapper(*args, **kwargs): start_time time.time() result func(*args, **kwargs) end_time time.time() execution_time end_time - start_time self.metrics[metric_name].append(execution_time) return result return wrapper return decorator def get_performance_report(self) - Dict: 生成性能报告 report {} for metric, times in self.metrics.items(): if times: report[metric] { count: len(times), avg_time: statistics.mean(times), max_time: max(times), min_time: min(times) } return report def check_sla_violations(self, sla_limits: Dict) - List[str]: 检查SLA违规情况 violations [] for metric, limit in sla_limits.items(): if metric in self.metrics and self.metrics[metric]: avg_time statistics.mean(self.metrics[metric]) if avg_time limit: violations.append(f{metric}平均时间{avg_time:.2f}s超过限制{limit}s) return violations # 使用示例 monitor PerformanceMonitor() monitor.track_performance(query_generation_time) def generate_query_with_monitoring(question: str, context: str): return query_generator.generate_sql_query(question, context)8. 安全考虑与权限控制8.1 查询安全防护机制实现多层次的安全防护# security_manager.py import re from typing import Set class SecurityManager: def __init__(self): self.dangerous_patterns [ r\b(drop|delete|truncate|alter|create|insert|update)\b, r;.*--, runion.*select, rexec\(|sp_, rxp_cmdshell ] self.allowed_functions { aggregate: [count, sum, avg, min, max], string: [concat, substring, lower, upper], date: [date_format, year, month, day] } def validate_query_safety(self, sql_query: str, user_role: str) - Dict: 全面验证查询安全性 safety_checks { syntax_check: self._check_sql_syntax(sql_query), dangerous_operations: self._check_dangerous_operations(sql_query), function_whitelist: self._check_function_whitelist(sql_query, user_role), data_access: self._check_data_access_patterns(sql_query, user_role) } # 汇总检查结果 all_passed all(safety_checks.values()) failed_checks [check for check, passed in safety_checks.items() if not passed] return { safe: all_passed, failed_checks: failed_checks, details: safety_checks } def _check_dangerous_operations(self, sql_query: str) - bool: 检查危险操作模式 query_lower sql_query.lower() for pattern in self.dangerous_patterns: if re.search(pattern, query_lower): return False return True def _check_function_whitelist(self, sql_query: str, user_role: str) - bool: 检查函数使用是否在允许范围内 # 根据用户角色确定允许的函数集合 if user_role analyst: allowed_functions self.allowed_functions[aggregate] self.allowed_functions[date] elif user_role admin: return True # 管理员无限制 else: allowed_functions self.allowed_functions[aggregate] # 提取SQL中的函数调用并进行验证 function_pattern r\b(\w)\s*\( used_functions re.findall(function_pattern, sql_query.lower()) for func in used_functions: if func not in allowed_functions and func not in [select, from, where]: return False return True8.2 基于角色的访问控制实现细粒度的权限管理# rbac_manager.py from typing import Dict, List, Set class RBACManager: def __init__(self): self.roles { viewer: { allowed_tables: {products, sales}, allowed_operations: {select}, max_rows: 1000, sensitive_columns: {password, salary} }, analyst: { allowed_tables: {products, sales, customers}, allowed_operations: {select, create_temp_table}, max_rows: 10000, sensitive_columns: {password} }, admin: { allowed_tables: {*}, allowed_operations: {*}, max_rows: None, sensitive_columns: set() } } def get_user_permissions(self, user_role: str) - Dict: 获取用户权限配置 return self.roles.get(user_role, self.roles[viewer]) def apply_query_restrictions(self, sql_query: str, user_role: str) - str: 根据用户角色应用查询限制 permissions self.get_user_permissions(user_role) # 这里可以添加具体的查询重写逻辑 # 比如添加行数限制、隐藏敏感列等 if permissions[max_rows]: # 为查询添加LIMIT子句如果还没有的话 if limit not in sql_query.lower(): sql_query f LIMIT {permissions[max_rows]} return sql_query def validate_table_access(self, tables: List[str], user_role: str) - bool: 验证用户对表的访问权限 permissions self.get_user_permissions(user_role) allowed_tables permissions[allowed_tables] if * in allowed_tables: return True return all(table in allowed_tables for table in tables)9. 部署与运维最佳实践9.1 Docker容器化部署创建完整的Docker部署配置# Dockerfile FROM python:3.9-slim WORKDIR /app # 安装系统依赖 RUN apt-get update apt-get install -y \ gcc \ rm -rf /var/lib/apt/lists/* # 复制依赖文件并安装 COPY requirements.txt . 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配置# docker-compose.yml version: 3.8 services: nl2query-api: build: . ports: - 8000:8000 environment: - OPENAI_API_KEY${OPENAI_API_KEY} - DATABASE_URL${DATABASE_URL} volumes: - ./logs:/app/logs healthcheck: test: [CMD, curl, -f, http://localhost:8000/health] interval: 30s timeout: 10s retries: 3 streamlit-ui: build: context: . dockerfile: Dockerfile.streamlit ports: - 8501:8501 depends_on: - nl2query-api environment: - API_ENDPOINThttp://nl2query-api:8000 # 可以添加数据库、缓存等依赖服务9.2 监控与日志配置实现完整的监控和日志体系# logging_config.py import logging import sys from pathlib import Path def setup_logging(log_level: str INFO, log_file: str app.log): 配置日志系统 # 创建日志目录 log_path Path(logs) log_path.mkdir(exist_okTrue) # 日志格式 formatter logging.Formatter( %(asctime)s - %(name)s - %(levelname)s - %(message)s ) # 根日志记录器 root_logger logging.getLogger() root_logger.setLevel(getattr(logging, log_level.upper())) # 控制台处理器 console_handler logging.StreamHandler(sys.stdout) console_handler.setFormatter(formatter) root_logger.addHandler(console_handler) # 文件处理器 file_handler logging.FileHandler(log_path / log_file) file_handler.setFormatter(formatter) root_logger.addHandler(file_handler) # 避免重复日志 root_logger.propagate False # 性能监控装饰器 def log_performance(operation_name: str): 性能日志装饰器 def decorator(func): wraps(func) def wrapper(*args, **kwargs): logger logging.getLogger(func.__module__) start_time time.time() try: result func(*args, **kwargs) execution_time time.time() - start_time logger.info( f{operation_name} completed in {execution_time:.2f}s, extra{operation: operation_name, duration: execution_time} ) return result except Exception as e: execution_time time.time() - start_time logger.error( f{operation_name} failed after {execution_time:.2f}s: {str(e)}, extra{operation: operation_name, error: str(e)} ) raise return wrapper return decorator10. 常见问题与解决方案10.1 查询生成准确性问题问题现象LLM生成的SQL查询语法正确但逻辑不符合业务需求。解决方案增强上下文提示提供更详细的业务规则和示例实现多轮对话允许用户澄清模糊需求
