# EDA Service 项目架构与源码分析 ## 📋 项目概述 **EDA Service** 是一个基于 AI 的数据探索和分析服务，核心功能是将自然语言问题转换为 SQL 查询（Chat2SQL/Chat2Data），并提供数据库结构分析、问题推荐等智能数据分析能力。 ### 核心技术栈 - **Web 框架**: Flask (异步: gevent) - **任务队列**: Celery + Redis (异步任务处理) - **数据库**: MySQL + SQLAlchemy ORM - **AI 模型**: OpenAI GPT-4/GPT-4o, Claude 3 (Sonnet) - **向量存储**: ChromaDB (知识库检索) - **监控**: Prometheus + Sentry - **部署**: Docker + Pulumi (IaC) --- ## 🏗️ 项目整体架构 ``` ┌─────────────────────────────────────────────────────────────────┐ │ Client (用户/API调用) │ └────────────────────────────┬────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────┐ │ Flask Web Application (app.py) │ │ ┌──────────────────────────────────────────────────────────┐ │ │ │ Handlers (路由层) │ │ │ │ - edatask.py: 核心业务 API (Chat2Data/SQL) │ │ │ │ - context.py: 上下文管理 │ │ │ │ - chat2query.py: Chat2Query API │ │ │ │ - sqleditor.py: SQL 编辑器 │ │ │ └──────────────────────────────────────────────────────────┘ │ └────────────────────────────┬────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────┐ │ Celery Task Queue (异步任务处理) │ │ ┌──────────────────────────────────────────────────────────┐ │ │ │ Tasks (tasks/eda.py) │ │ │ │ - task_gen_db_schema: 生成数据库架构描述 │ │ │ │ - task_chat2data: 自然语言到数据查询 │ │ │ │ - task_suggest_questions: 智能问题推荐 │ │ │ │ - task_text2sql: 文本转SQL │ │ │ │ - task_sql_refine: SQL 修正优化 │ │ │ └──────────────────────────────────────────────────────────┘ │ └────────────────────────────┬────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────┐ │ Service Layer (业务逻辑层) │ │ ┌──────────────────────────────────────────────────────────┐ │ │ │ svc/eda_task.py: 核心 EDA 任务编排 │ │ │ │ svc/data_explore.py: 数据库探索与 schema 分析 │ │ │ │ svc/openai.py: OpenAI API 调用封装 │ │ │ │ svc/claude.py: Claude API 调用封装 │ │ │ │ svc/dataapi.py: SQL 执行器 (跨区域数据 API) │ │ │ │ svc/context.py: 上下文管理 (session/data context) │ │ │ │ svc/measuring.py: 计量统计 │ │ │ └──────────────────────────────────────────────────────────┘ │ └────────────────────────────┬────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────┐ │ Data Layer (数据存储层) │ │ ┌──────────────────────────────────────────────────────────┐ │ │ │ Models (SQLAlchemy ORM) │ │ │ │ - context.py: 数据/会话上下文 │ │ │ │ - edatask.py: 任务和任务作业 │ │ │ │ - ailog.py: AI 调用日志 │ │ │ │ - user_preference.py: 用户偏好设置 │ │ │ │ - domain_knowledge.py: 领域知识库 │ │ │ │ - feedback.py: 用户反馈 │ │ │ └──────────────────────────────────────────────────────────┘ │ │ ┌──────────────────────────────────────────────────────────┐ │ │ │ Storage │ │ │ │ - MySQL: 持久化存储 │ │ │ │ - Redis: 任务队列 + 缓存 │ │ │ │ - ChromaDB: 向量数据库 (知识库检索) │ │ │ └──────────────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────┐ │ External Services (外部服务) │ │ - OpenAI API (GPT-4/GPT-4o) │ │ - AWS Bedrock (Claude 3) │ │ - TiDB Cloud DataAPI (SQL 执行) │ │ - Sentry (错误追踪) │ │ - Prometheus (监控指标) │ └─────────────────────────────────────────────────────────────────┘ ``` --- ## 📁 目录结构详解 ``` eda_service/ │ ├── app.py # Flask 应用主入口 (Web Server) ├── worker.py # Celery Worker 入口 (异步任务处理) ├── config.py # 配置管理 (环境变量) ├── requirements.txt # Python 依赖 │ ├── handlers/ # 路由处理层 (Controller) │ ├── base.py # 基础工具函数 (权限校验、响应封装) │ ├── edatask.py # 核心业务 API (Chat2Data/SQL) │ ├── context.py # 上下文管理 API │ ├── chat2query.py # Chat2Query 接口 │ ├── sqleditor.py # SQL Editor 接口 │ └── inner.py # 内部接口 │ ├── tasks/ # Celery 异步任务 │ ├── celery_app.py # Celery 应用配置 │ ├── eda.py # 核心 EDA 任务定义 ⭐⭐⭐ │ └── utils.py # 任务工具函数 │ ├── svc/ # 业务逻辑层 (Service) │ ├── eda_task.py # EDA 任务编排核心 ⭐⭐⭐ │ ├── data_explore.py # 数据库 Schema 分析 ⭐⭐ │ ├── openai.py # OpenAI API 封装 │ ├── claude.py # Claude API 封装 │ ├── dataapi.py # SQL 执行器封装 │ ├── context.py # 上下文管理 │ ├── executor.py # SQL 执行器实现 │ ├── measuring.py # 计量统计 │ ├── feishu.py # 飞书通知 │ └── sample_data.py # 样例数据管理 │ ├── models/ # ORM 模型层 (Model) │ ├── base.py # SQLAlchemy 基础配置 │ ├── context.py # 上下文表 (数据/会话) │ ├── edatask.py # EDA 任务表 │ ├── ailog.py # AI 调用日志 │ ├── user_preference.py # 用户偏好 │ ├── domain_knowledge.py # 知识库 │ ├── feedback.py # 用户反馈 │ ├── api_key.py # API Key 管理 │ └── ratelimit.py # 限流管理 │ ├── utils/ # 工具函数 │ ├── database.py # 数据库连接工具 │ ├── sql.py # SQL 解析和验证 │ ├── credential.py # 凭证管理 │ ├── context.py # 上下文格式化 │ ├── logging.py # 日志工具 │ ├── local.py # 线程本地存储 │ └── template.py # Prompt 模板 │ ├── schema/ # API Schema 定义 │ └── schema.py # Marshmallow Schema │ ├── consts/ # 常量定义 │ └── consts.py # 全局常量 │ ├── errors/ # 错误处理 │ └── errors.py # 自定义异常 │ ├── domain_info/ # 领域知识 │ ├── index.py # 知识库管理 │ └── lengjing_v2.py # 特定领域知识 │ ├── spider/ # 测试数据集 (Spider/BIRD) │ └── database/ # 测试数据库 │ ├── alembic/ # 数据库迁移 │ └── versions/ # 迁移脚本 │ ├── deployments/ # 部署配置 │ └── pulumi/ # Pulumi IaC 配置 │ ├── tests/ # 测试用例 │ ├── test_chat2query_api.py │ └── test_chat2query_tasks.py │ └── docs/ # 文档 └── swagger.json # API 文档 ``` --- ## 🔑 核心模块详解 ### 1. **任务编排核心 (tasks/eda.py)** ⭐⭐⭐ 这是整个系统的核心任务调度层，定义了所有异步 Celery 任务。 #### 关键任务： ##### **task_gen_db_schema** ```python @app.task(bind=True, base=EDACerlryTaskBase, autoretry_for=(Exception,), retry_kwargs={'max_retries': 3}) def task_gen_db_schema(self, data_context_id, mode=MODE_EXPLORE, need_refine=False, set_default=False, metrics=None) ``` - **功能**: 自动分析数据库结构，生成表/列/关系描述 - **流程**: 1. 生成列描述 (task_gen_column_description) 2. 生成表描述 (task_gen_table_description) 3. 生成表关系 (task_gen_table_relations) 4. 提取业务实体 (task_gen_entity) 5. 生成数据库摘要 (task_gen_db_description) ##### **task_chat2data** ```python @app.task(bind=True, base=EDACerlryTaskBase, autoretry_for=(Exception,), retry_kwargs={'max_retries': 3}) def task_chat2data(self, data_context_id, session_context_id, raw_question, direct_resolve, auto_run, sql_to_data, data_to_chart, evidence, app_id, public_key, user_token, ...) ``` - **功能**: 将自然语言问题转换为 SQL 并执行 - **流程**: 1. 加载数据上下文 (data_context) 2. 设置会话上下文 (session_context) 3. 调用 EDATask.task_proxy 进行任务分解和执行 4. 生成结果 (SQL + 数据 + 图表) ##### **task_suggest_questions** ```python @app.task(bind=True, base=EDACerlryTaskBase, autoretry_for=(Exception,), retry_kwargs={'max_retries': 3}) def task_suggest_questions(self, session_context_id, data_context_id, app_id, public_key, user_token, question_num=5, question_only=True, metrics=None) ``` - **功能**: 基于数据库结构智能推荐分析问题 ##### **task_sql_refine** ```python @app.task(bind=True, base=EDACerlryTaskBase, autoretry_for=(Exception,), retry_kwargs={'max_retries': 3}) def task_sql_refine(self, data_context_id, session_context_id, sql, raw_question, feedback, sql_error, user_token, metrics) ``` - **功能**: 根据错误信息或用户反馈修正/重写 SQL #### 任务生命周期管理 (EDACerlryTaskBase) 继承自 Celery Task，实现了完整的任务生命周期钩子： - **before_start**: 设置上下文 (trace_id, user_token)，更新状态为 RUNNING - **on_success**: 持久化结果，更新状态为 DONE，记录指标 - **on_failure**: 记录错误，更新状态为 FAILED，上报指标 - **after_return**: 清理本地上下文 --- ### 2. **EDA 任务编排 (svc/eda_task.py)** ⭐⭐⭐ 这是业务逻辑的核心，实现了从自然语言到 SQL 的完整流程。 #### 核心类: EDATask ```python class EDATask: def __init__(self, session_context_id, binded_thread_id=""): self.session_context_id = session_context_id self.binded_thread_id = binded_thread_id self.tasks_to_be_resolved = deque() self.tasks_tree = {} # 任务树结构 ``` #### 关键方法： ##### **task_proxy** (任务代理入口) ```python def task_proxy(self, app_id, query, explaination="", query_id="", task_id="", direct_resolve=True, auto_run=True, sql_to_data=False, data_to_chart=False): # 1. 加载用户偏好 (语言、模型、模式) # 2. 任务分解 (如果 direct_resolve=False) # 3. 生成 SQL # 4. 执行 SQL (如果 auto_run=True) # 5. 生成图表 (如果 data_to_chart=True) # 6. yield 返回中间结果 (流式响应) ``` ##### **agent_text2sql** (核心 SQL 生成) ```python def agent_text2sql(self, task_description, db_context, relevant_tables, table_num=10, auto_revise=True, language=LANGUAGE_EN, ai_model=GPT_DEFAULT): # 1. 构建 Prompt (数据库 Schema + 问题) # 2. 知识库检索 (domain knowledge) # 3. 调用 LLM 生成 SQL # 4. SQL 后处理和修正 ``` ##### **agent_suggest_questions** (问题推荐) ```python def agent_suggest_questions(self, db_context, question_num, ai_model=GPT_DEFAULT, language=LANGUAGE_EN, question_level=SuggestQuestionLevel.SIMPLE): # 1. 提取数据库实体和关系 # 2. 构建推荐 Prompt # 3. 调用 LLM 生成推荐问题列表 ``` #### 任务类型支持 - **data_retrieval**: 数据查询任务 (SELECT) - **data_manipulation**: 数据修改任务 (INSERT/UPDATE/DELETE) - **task_breakdown**: 复杂任务分解 - **task_compare**: 对比分析 --- ### 3. **数据探索服务 (svc/data_explore.py)** ⭐⭐ 自动分析数据库结构，生成人类可读的描述。 #### 核心函数： ##### **setup_data_context** ```python def setup_data_context(data_context, app_id, public_key, user_token=""): # 1. 连接数据库 # 2. 获取表和列的元数据 # 3. 采样数据 (distinct values) # 4. 分析外键关系 ``` ##### **explore_column_description** ```python def explore_column_description(ctx, table_name, mode, language, ai_model): # 使用 LLM 分析列的业务含义 # 输入: 表名、列名、数据类型、采样值 # 输出: 列的业务描述 ``` ##### **explore_table_description** ```python def explore_table_description(table_name, mode, language, ai_model): # 基于列描述生成表的业务描述 ``` ##### **explore_tables_relationship** ```python def explore_tables_relationship(mode, relevant_table_num, ai_model): # 分析表之间的业务关系 (JOIN 路径) ``` ##### **entities_extract** ```python def entities_extract(language, ai_model): # 提取业务实体 (Entity) # 例如: User, Order, Product ``` --- ### 4. **上下文管理 (models/context.py & svc/context.py)** #### Context Model (数据模型) ```python class Context(Base, BaseMixin): _type = Column(String(16)) # DATA / SESSION / SESSION_CHAT / DOMAIN name = Column(String(32)) org_id = Column(String(64)) cluster_id = Column(String(64)) db_name = Column(String(64)) data = Column(LONGTEXT) # JSON 存储 data_context_id = Column(Integer) # 会话关联的数据上下文 thread_id = Column(String(64)) # OpenAI Assistant API Thread ID ``` #### 上下文类型： 1. **DATA Context** (数据上下文) - 存储数据库 Schema 分析结果 - 表/列描述、关系、实体 - 知识库关联 2. **SESSION Context** (会话上下文) - 多轮对话历史 - 任务树 (tasks_tree) - 引用的 data_context 3. **DOMAIN Context** (领域上下文) - 组织级别的领域知识 - 术语表、业务规则 #### context service (上下文服务) ```python class ContextService: def load_dataset_context(self, data_context_id): # 加载数据库 Schema 和描述 def load_session_context(self, session_context_id): # 加载会话历史和任务状态 def save_session_context(self, session_context_id, data): # 持久化会话状态 ``` --- ### 5. **AI 服务封装 (svc/openai.py & svc/claude.py)** #### OpenAI Service ```python def ai_completion(module, prompt, response_format="text", ai_model=GPT_DEFAULT, extra_info=None): """ 统一的 AI 调用接口 - 支持 GPT-4/GPT-4o/GPT-4o-mini - 自动 token 计算和限流 - 重试机制 (tenacity) - 日志记录 """ ``` #### Claude Service ```python def claude_completion(module, prompt, response_format="text", ai_model=Claude3.SONNET, extra_info=None): """ Claude API 调用 - 支持 Sonnet/Haiku/Opus - 通过 AWS Bedrock 调用 """ ``` #### OpenAI Assistant API ```python def ai_assistant_chat(thread_id, prompt, module, assistant_name=AssistantName.CHAT2SQL, ...): """ 使用 OpenAI Assistant API (支持多轮对话) - 自动管理 thread - 流式返回 """ ``` --- ### 6. **SQL 执行器 (svc/executor.py & svc/dataapi.py)** #### SQL Executor 支持多种数据库: - TiDB Cloud (DataAPI) - MySQL - PostgreSQL - BigQuery - SingleStore ```python class TiDBCloudDedicatedExecutor: def execute(self, sql): # 通过 DataAPI 执行 SQL # 返回结果: columns + rows def explain_sql(self, sql): # EXPLAIN SQL (验证 SQL 正确性) ``` --- ### 7. **用户偏好管理 (models/user_preference.py)** ```python class UserPreference: llm_model: str # GPT-4 / Claude-3 language_code: str # English / 中文 mode_type: str # spider / bird / normal rate_limit_per_day: int # 每日限流 suggest_question_level: str # simple / complex ``` --- ## 🔄 核心业务流程 ### 流程 1: Chat2Data (自然语言查询数据) ``` 1. 用户请求 └─> POST /api/v1beta/apps/{app_id}/chat2data Body: { cluster_id, database, question } 2. Handler 处理 (handlers/edatask.py) └─> handle_chat2data() ├─> 验证权限和限流 ├─> 查找或创建 data_context ├─> 创建 session_context └─> 提交异步任务: task_chat2data 3. Celery 任务执行 (tasks/eda.py) └─> task_chat2data() ├─> 加载 data_context ├─> 初始化 EDATask(session_context_id) └─> 调用 eda.task_proxy() 4. 任务编排 (svc/eda_task.py) └─> EDATask.task_proxy() ├─> 步骤 1: 加载用户偏好 ├─> 步骤 2: 检索相关表 (向量检索) ├─> 步骤 3: 知识库检索 (domain knowledge) ├─> 步骤 4: 生成 SQL (agent_text2sql) │ ├─> 构建 Prompt │ ├─> 调用 LLM (OpenAI/Claude) │ └─> SQL 后处理 ├─> 步骤 5: 执行 SQL (auto_run=True) │ └─> TiDBCloudDedicatedExecutor.execute() ├─> 步骤 6: 生成图表配置 (data_to_chart=True) └─> yield 返回结果 (流式更新) 5. 结果持久化 ├─> EDATaskJob.set_result() (保存到数据库) ├─> HistoryLog.create() (AI 调用日志) └─> 上报 Prometheus 指标 6. 客户端轮询 └─> GET /api/v1beta/apps/{app_id}/jobs/{job_id} └─> 返回任务状态和结果 ``` --- ### 流程 2: 数据库 Schema 分析 ``` 1. 用户创建 Data Summary └─> POST /api/v1beta/apps/{app_id}/data_summaries Body: { cluster_id, database, description } 2. 提交异步任务 └─> task_gen_db_schema(data_context_id) 3. Schema 分析流程 └─> task_gen_db_schema() │ ├─> 步骤 1: 设置数据上下文 │ └─> setup_data_context() │ ├─> 连接数据库 │ ├─> 获取所有表和列 │ ├─> 采样数据 (DISTINCT values) │ └─> 分析外键关系 │ ├─> 步骤 2: 生成列描述 (循环所有表) │ └─> task_gen_column_description() │ └─> explore_column_description() │ ├─> 构建 Prompt (表名、列名、类型、样例) │ ├─> 调用 LLM │ └─> 解析列描述 JSON │ ├─> 步骤 3: 生成表描述 (循环所有表) │ └─> task_gen_table_description() │ └─> explore_table_description() │ ├─> 基于列描述 │ ├─> 调用 LLM │ └─> 返回表业务含义 │ ├─> 步骤 4: 分析表关系 │ └─> task_gen_table_relations() │ └─> explore_tables_relationship() │ ├─> 分析外键和业务关系 │ └─> 生成 JOIN 路径 │ ├─> 步骤 5: 提取业务实体 │ └─> task_gen_entity() │ └─> entities_extract() │ ├─> 聚合表和列描述 │ ├─> 调用 LLM │ └─> 提取 Entity (User, Order, Product...) │ └─> 步骤 6: 生成数据库摘要 └─> task_gen_db_description() └─> explore_db_description() ├─> 基于实体和关系 └─> 生成数据库整体描述 4. 结果存储 └─> Context.update_data() (JSON 格式) ├─> db_schema: { table -> columns } ├─> table_descriptions: { table -> description } ├─> column_descriptions: { table -> { column -> desc } } ├─> table_relationships: [ { from, to, type } ] ├─> entity: { entity_name -> { summary, attributes } } └─> summary: "整体描述" ``` --- ### 流程 3: 智能问题推荐 ``` 1. 请求推荐问题 └─> POST /api/v1beta/apps/{app_id}/suggest_questions Body: { cluster_id, database, question_num: 5 } 2. 任务执行 └─> task_suggest_questions() └─> EDATask.task_suggest_questions() │ ├─> 检查本地缓存 (cluster_id + db_name) │ └─> 如有缓存直接返回 │ └─> 调用 agent_suggest_questions() ├─> 步骤 1: 加载数据上下文 │ ├─> 如果 status=done: 使用 entity schema │ └─> 否则: 使用原始 db_schema │ ├─> 步骤 2: 构建 Prompt │ ├─> 数据库摘要 │ ├─> 实体列表 │ └─> 问题类型 (Descriptive/Inferential/...) │ ├─> 步骤 3: 调用 LLM │ └─> 返回 JSON 数组 │ [{ type, question, keyword }, ...] │ └─> 步骤 4: 随机采样返回 ``` --- ## 🗂️ 数据库设计 ### 核心表结构 #### 1. **context** (上下文表) ```sql CREATE TABLE context ( id INT PRIMARY KEY AUTO_INCREMENT, _type VARCHAR(16), -- DATA / SESSION / SESSION_CHAT / DOMAIN name VARCHAR(32), org_id VARCHAR(64), -- 组织ID cluster_id VARCHAR(64), -- 集群ID db_name VARCHAR(64), -- 数据库名 data LONGTEXT, -- JSON 数据 (Schema/历史消息) data_context_id INT, -- 关联的数据上下文 thread_id VARCHAR(64), -- OpenAI Thread ID description VARCHAR(4096), -- 描述 is_default BOOLEAN, -- 是否为默认 created_at DATETIME, updated_at DATETIME ); ``` #### 2. **eda_task_job** (任务作业表) ```sql CREATE TABLE eda_task_job ( id INT PRIMARY KEY AUTO_INCREMENT, job_id VARCHAR(36) UNIQUE, -- UUID status VARCHAR(8), -- init / running / done / failed fn_name VARCHAR(128), -- 任务函数名 fn_args VARCHAR(4096), -- 任务参数 (JSON) result LONGTEXT, -- 结果 (JSON) ended_at DATETIME, -- 结束时间 org_id VARCHAR(64), -- 组织ID created_at DATETIME, updated_at DATETIME ); ``` #### 3. **history_log** (AI 调用日志) ```sql CREATE TABLE history_log ( id INT PRIMARY KEY AUTO_INCREMENT, status VARCHAR(16), -- success / failed trace_id VARCHAR(64), -- 追踪ID _type VARCHAR(32), -- chat2data / chat2sql org_id VARCHAR(64), cluster_id VARCHAR(64), database VARCHAR(64), question TEXT, -- 用户问题 data_summary_id INT, -- 数据上下文ID session_id INT, -- 会话ID input JSON, -- 输入 (问题、模式) output JSON, -- 输出 (SQL、数据) settings JSON, -- 设置 (模型、语言) token_count INT, -- Token 消耗 start_at DATETIME, created_at DATETIME ); ``` #### 4. **user_preference** (用户偏好) ```sql CREATE TABLE user_preference ( id INT PRIMARY KEY AUTO_INCREMENT, app_id INT, org_id VARCHAR(64), llm_model VARCHAR(32), -- GPT-4 / Claude-3 language_code VARCHAR(16), -- English / 中文 mode_type VARCHAR(32), -- normal / spider / bird rate_limit_per_day INT, -- 每日限额 suggest_question_level VARCHAR(16), -- simple / complex created_at DATETIME ); ``` #### 5. **domain_knowledge** (知识库) ```sql CREATE TABLE domain_knowledge ( id INT PRIMARY KEY AUTO_INCREMENT, org_id VARCHAR(64), app_id INT, knowledge_base_id INT, _type VARCHAR(32), -- term / example raw_data JSON, -- 原始数据 meta_data JSON, -- 元数据 is_deleted BOOLEAN, created_at DATETIME ); ``` --- ## 🔐 权限和认证 ### 认证方式 1. **Basic Auth** (API Key) ``` Authorization: Basic base64(public_key:private_key) ``` 2. **User Token** (内部服务) ``` X-User-Token: ``` ### 权限检查流程 ```python # handlers/base.py def check_permission(data_app_id): # 1. 获取 API Key api_key = get_access_key() # 2. 验证 API Key 是否属于 org_id org_id = get_org_id_from_header() if not is_apikey_belongs_to_org_id(api_key, org_id): raise PermissionDeniedException() # 3. 验证用户是否有集群权限 if not user_has_permission(user_token, cluster_id): raise PermissionDeniedException() ``` --- ## 📊 监控和日志 ### Prometheus 指标 ```python # 业务延迟 BIZ_LATENCY = Histogram('biz_latency_seconds', labelnames=("name", "result", "is_internal", "status_code", "llm_model")) # 任务延迟 TASK_LATENCY = Histogram('task_latency_seconds', labelnames=("name", "result", "einfo", "is_internal")) # 队列延迟 QUEUE_LATENCY = Histogram('queue_latency_seconds', labelnames=("name", "is_internal")) # HTTP 请求延迟 REQUEST_LATENCY = Histogram('request_latency_seconds', labelnames=("base_url", "method", "status_code")) # 队列长度 QUEUE_LENGTH = Gauge("queue_length", labelnames=("name",)) ``` ### 日志体系 ```python # utils/logging.py logging.info("session_id = %s, query = %s, result = %s", session_id, query, result) ``` - **Trace ID**: B3 格式 (分布式追踪) - **Session ID**: 会话追踪 - **Job ID**: 任务追踪 --- ## 🚀 部署架构 ### 容器化部署 ```dockerfile # Dockerfile FROM python:3.9 COPY requirements.txt . RUN pip install -r requirements.txt COPY . /app CMD ["gunicorn", "-c", "gunicorn_config.py", "app:app"] ``` ### 服务组件 1. **Web Server** (Gunicorn + Gevent) - 处理 HTTP 请求 - 异步 I/O 2. **Celery Worker** (Gevent Pool) - 执行异步任务 - 并发: 10 workers 3. **Redis** (Task Queue + Cache) - Celery Broker - 速率限制计数器 4. **MySQL** (持久化存储) - 上下文、任务、日志 ### 环境配置 ```bash # .env SQLALCHEMY_DATABASE_URI=mysql+pymysql://user:pass@host:3306/eda_service BROKER_URL=redis://localhost:6379/1 OPENAI_API_KEY=sk-xxx ACCOUNT_ADDR=http://account-service:11000 AIPOWER_ADDR=http://aipower-service:10010 SENTRY_DSN=https://xxx@sentry.io/xxx ENVIRONMENT=prod ``` --- ## 🧪 测试 ### 单元测试 ```bash pytest tests/test_chat2query_tasks.py ``` ### 集成测试 ```bash # 测试 Spider 数据集 python spider/runbird.py ``` ### E2E 测试 ```bash python scripts/e2etest.py ``` --- ## 🛠️ 开发指南 ### 本地开发环境搭建 ```bash # 1. 安装依赖 pip install -r requirements.txt # 2. 启动 Redis 和 MySQL docker-compose up -d redis mysql # 3. 数据库迁移 alembic upgrade head # 4. 启动服务 honcho start # 同时启动 web 和 worker ``` ### 添加新任务 1. 在 `tasks/eda.py` 定义任务: ```python @app.task(bind=True, base=EDACerlryTaskBase) def task_new_feature(self, param1, param2, metrics=None): # 任务逻辑 return result ``` 2. 在 `handlers/edatask.py` 添加路由: ```python @edatask_bp.post("/new-feature") def handle_new_feature(): job_id = enqueue(task_new_feature, org_id, param1, param2) return success(result={"job_id": job_id}) ``` ### 代码规范 - **Linter**: flake8 - **命名规范**: snake_case (Python PEP 8) - **日志**: 使用统一的 logging 模块 --- ## 🔍 关键技术点 ### 1. **向量检索 (ChromaDB)** 用于检索相关表和知识库： ```python # 构建向量索引 collection.add( documents=[table_description], metadatas=[{"table_name": table_name}], ids=[table_name] ) # 检索 results = collection.query( query_texts=[question], n_results=10 ) ``` ### 2. **Prompt Engineering** 关键 Prompt 模板： - **Text2SQL Prompt**: `svc/eda_task.py:agent_text2sql` - **Schema Description Prompt**: `svc/data_explore.py:explore_column_description` - **Question Suggestion Prompt**: `svc/eda_task.py:agent_suggest_questions` ### 3. **SQL 后处理** ```python # SQL 修正链 sql_revise_chains = [ "spider_eval_revise", # Spider 数据集格式修正 "ignore_case_revise", # 忽略大小写 "sql_condition_revise", # 条件优化 "tidb_syntax_revise", # TiDB 语法适配 ] ``` ### 4. **流式响应 (Yield)** ```python def task_proxy(self, ...): for task_tree, question_id in self._process_task(): # 中间结果流式返回 yield task_tree, question_id ``` ### 5. **重试机制 (Tenacity)** ```python @retry( retry=retry_if_not_exception_type(BadRequestException), stop=stop_after_attempt(3), wait=wait_random_exponential(min=1, max=60), before_sleep=before_sleep_log(logger, logging.WARNING) ) def ai_completion(...): # 调用 LLM ``` --- ## 📚 扩展阅读 ### 相关论文和资料 - **Spider Dataset**: Text-to-SQL 基准数据集 - **BIRD Dataset**: 跨领域 Text-to-SQL 数据集 - **LangChain**: LLM 应用开发框架 - **ChromaDB**: 向量数据库 ### API 文档 - Swagger UI: `/eda/apidocs/` - API Spec: `/eda/apidocs/swagger.json` --- ## 🤔 常见问题 ### Q1: 如何添加新的数据库支持？ A: 在 `svc/executor.py` 添加新的 Executor 类，实现 `execute()` 和 `explain_sql()` 方法。 ### Q2: 如何切换 LLM 模型？ A: 通过 `UserPreference` 配置 `llm_model` 字段 (GPT-4 / Claude-3)。 ### Q3: 如何优化 SQL 生成质量？ A: 1. 改进数据库 Schema 描述 (task_gen_db_schema) 2. 添加领域知识库 (domain_knowledge) 3. 优化 Prompt 模板 ### Q4: 如何处理大型数据库？ A: 1. 使用向量检索筛选相关表 2. 限制 relevant_table_num (默认 10) 3. 开启缓存 --- ## 📝 总结 EDA Service 是一个**复杂的 AI 驱动的数据分析系统**，核心能力包括： ✅ **自然语言到 SQL** (Chat2SQL/Chat2Data) ✅ **数据库自动分析** (Schema Understanding) ✅ **智能问题推荐** (Suggest Questions) ✅ **多轮对话** (Session Management) ✅ **知识库增强** (Domain Knowledge) ✅ **SQL 修正优化** (SQL Refine) **技术亮点**： - 异步任务处理 (Celery) - 向量检索优化 (ChromaDB) - 多 LLM 支持 (OpenAI/Claude) - 完善的监控体系 (Prometheus/Sentry) - 流式响应 (Yield) **适用场景**： - BI 工具的自然语言查询 - 数据分析平台 - SQL 辅助生成工具 - 数据探索和可视化 --- **作者**: EDA Service Team **最后更新**: 2025-11-05 **版本**: v1.0