基于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 requests

3.3 配置文件设置

创建配置文件.env，包含必要的环境变量：

# .env 配置文件 OPENAI_API_KEY=your_openai_api_key_here DATABASE_URL=postgresql://user:password@localhost:5432/your_database VECTOR_DB_URL=your_vector_db_connection_string LOG_LEVEL=INFO MAX_QUERY_RESULTS=1000

对应的配置读取类：

# 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', encoding='utf-8') as f: json.dump(metadata, f, indent=2, ensure_ascii=False) # 使用示例 if __name__ == "__main__": extractor = MetadataExtractor("postgresql://user:pass@localhost/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_impl="duckdb+parquet", persist_directory=persist_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', encoding='utf-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(f"table_{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(f"col_{i}_{j}") # 批量添加文档 embeddings = self.encoder.encode(documents).tolist() self.collection.add( embeddings=embeddings, documents=documents, metadatas=metadatas, ids=ids ) def semantic_search(self, query: str, n_results: int = 5): """语义搜索相关元数据""" query_embedding = self.encoder.encode([query]).tolist() results = self.collection.query( query_embeddings=query_embedding, n_results=n_results ) return results

5. 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( model=self.model, messages=[ {"role": "system", "content": "你是一个专业的SQL专家。"}, {"role": "user", "content": prompt} ], temperature=0.1, # 低温度确保确定性 max_tokens=500 ) 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( model=self.model, messages=[{"role": "user", "content": prompt}], temperature=0.7, max_tokens=300 ) 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 optimized

6. 完整系统集成与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="自然语言查询生成框架", version="1.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_model=QueryResponse) 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( success=False, error=f"查询验证失败: {validation_result['error']}" ) # 4. 优化查询 optimized_query = QueryValidator.optimize_query(result["sql_query"]) return QueryResponse( success=True, sql_query=optimized_query, explanation=result["explanation"], execution_time=generation_time ) else: return QueryResponse( success=False, error=result["error"] ) except Exception as e: raise HTTPException(status_code=500, detail=f"服务器内部错误: {str(e)}") @app.get("/health") async def health_check(): """健康检查端点""" return {"status": "healthy", "version": "1.0.0"} if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000)

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="🔍", layout="wide" ) # 样式定制 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_html=True) # 应用标题 st.markdown('<div class="main-header">🔍 自然语言数据查询平台</div>', unsafe_allow_html=True) # 侧边栏配置 with st.sidebar: st.header("配置选项") api_endpoint = st.text_input( "API端点", value="http://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个产品", height=100 ) if st.button("生成查询", type="primary"): 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 class="success-query">✅ 查询生成成功</div>', unsafe_allow_html=True) # 显示生成的SQL st.subheader("生成的SQL查询") st.code(result["sql_query"], language="sql") # 显示解释 st.subheader("查询解释") st.info(result["explanation"]) # 这里可以添加查询执行和结果显示逻辑 else: st.markdown(f'<div class="error-query">❌ 生成失败: {result["error"]}</div>', unsafe_allow_html=True) else: st.error(f"API请求失败: {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, key=example): 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("生成查询", type="primary"): 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';.*--', r'union.*select', r'exec\(|sp_', r'xp_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 True

8.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_ENDPOINT=http://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_ok=True) # 日志格式 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 decorator