# EDA Service 快速上手指南 🚀

## 一句话介绍

**EDA Service** 是一个将自然语言转换为 SQL 查询的 AI 服务，用户提问"最近一个月销售额是多少"，系统自动生成并执行 SQL，返回数据和图表。

---

## 5 分钟理解核心架构

### 整体架构图 (简化版)

```
用户提问 "Show me top 5 customers"
    ↓
Flask API (app.py)
    ↓
Celery Task (异步处理)
    ↓
EDATask (svc/eda_task.py)
    ├─> 1. 向量检索 (找相关表)
    ├─> 2. 知识库检索 (找业务规则)
    ├─> 3. LLM 生成 SQL (OpenAI/Claude)
    ├─> 4. 执行 SQL (TiDB Cloud)
    └─> 5. 返回结果 (JSON)
```

### 核心文件 (只需关注这 5 个)

| 文件 | 职责 | 重要度 |
|------|------|--------|
| `app.py` | Flask 入口，路由注册 | ⭐⭐ |
| `tasks/eda.py` | Celery 任务定义 (异步) | ⭐⭐⭐ |
| `svc/eda_task.py` | 核心业务逻辑 (Text2SQL) | ⭐⭐⭐⭐⭐ |
| `handlers/edatask.py` | API 路由处理 | ⭐⭐⭐ |
| `models/context.py` | 数据库模型 (ORM) | ⭐⭐ |

---

## 核心概念

### 1. **Context** (上下文)

系统有 3 种上下文：

- **Data Context** (数据上下文): 存储数据库 Schema、表/列描述
- **Session Context** (会话上下文): 存储多轮对话历史
- **Domain Context** (领域上下文): 存储领域知识库 (术语、规则)

**比喻**: Context 就像是 AI 的"记忆"，Data Context 是对数据库的理解，Session Context 是对话历史，Domain Context 是业务知识。

### 2. **Task** (任务)

用户请求会被转换为异步任务，通过 Celery 执行：

```python
# 常见任务类型
task_chat2data          # 自然语言 → SQL + 数据
task_gen_db_schema      # 自动分析数据库结构
task_suggest_questions  # 智能推荐问题
task_sql_refine         # SQL 修正优化
```

### 3. **任务生命周期**

```
1. INIT (初始化)
   ↓
2. RUNNING (执行中)
   ↓
3. DONE (完成) / FAILED (失败)
```

客户端通过轮询 `/jobs/{job_id}` 获取任务状态和结果。

---

## 核心流程详解

### 流程 1: Chat2Data (最重要)

**用户提问 → 生成 SQL → 执行 → 返回数据**

#### Step 1: 用户发起请求

```bash
POST /api/v1beta/apps/123/endpoint/v3/chat2data
{
  "cluster_id": "xxx",
  "database": "mydb",
  "question": "Show me top 5 customers by revenue"
}
```

#### Step 2: 系统处理流程

```python
# 1. 验证权限 (handlers/edatask.py)
handle_chat2data_v3()
  └─> check_permission()

# 2. 创建/查找上下文
data_context = find_or_create_data_context(cluster_id, database)
session_context = create_session_context(data_context)

# 3. 提交异步任务
job_id = enqueue(task_chat2data,
    data_context_id=data_context.id,
    session_context_id=session_context.id,
    question="Show me top 5 customers by revenue"
)

# 4. 返回 job_id
return {"job_id": job_id}
```

#### Step 3: Celery 任务执行

```python
# tasks/eda.py
@app.task
def task_chat2data(data_context_id, session_context_id, question, ...):
    # 1. 加载数据上下文
    data_context = Context.get_by_id(data_context_id)
    db_schema = data_context.load_data()["db_schema"]

    # 2. 初始化 EDA 任务
    eda = EDATask(session_context_id)

    # 3. 执行任务代理 (核心)
    for task_tree, question_id in eda.task_proxy(question):
        # 流式更新结果
        yield task_tree
```

#### Step 4: 核心 Text2SQL 逻辑

```python
# svc/eda_task.py
class EDATask:
    def task_proxy(self, question):
        # 1. 向量检索相关表 (ChromaDB)
        relevant_tables = self.vector_search(question, top_k=10)

        # 2. 知识库检索 (Domain Knowledge)
        domain_knowledge = self.knowledge_search(question)

        # 3. 构建 Prompt
        prompt = f"""
        Database Schema:
        {relevant_tables}

        Domain Knowledge:
        {domain_knowledge}

        Question: {question}

        Generate SQL:
        """

        # 4. 调用 LLM (OpenAI/Claude)
        sql = ai_completion("text2sql", prompt, response_format="json")

        # 5. 执行 SQL
        executor = get_sql_executor(data_context)
        result = executor.execute(sql)

        # 6. 返回结果
        return {
            "sql": sql,
            "data": result,
            "columns": result["columns"],
            "rows": result["rows"]
        }
```

#### Step 5: 客户端轮询结果

```bash
GET /api/v1beta/apps/123/jobs/{job_id}
Response:
{
  "job_id": "xxx",
  "status": "done",
  "result": {
    "type": "data_retrieval",
    "sql": "SELECT customer_name, SUM(revenue) as total FROM ...",
    "data": {
      "columns": ["customer_name", "total"],
      "rows": [["Alice", 10000], ["Bob", 8000], ...]
    }
  }
}
```

---

### 流程 2: 数据库自动分析

**连接数据库 → 分析表结构 → 生成业务描述**

```python
# 1. 用户创建 Data Summary
POST /api/v1beta/apps/123/data_summaries
{
  "cluster_id": "xxx",
  "database": "mydb",
  "description": "Sales database"
}

# 2. 后台执行分析任务
task_gen_db_schema(data_context_id)
  │
  ├─> 步骤 1: 连接数据库，获取所有表和列
  ├─> 步骤 2: 采样数据 (DISTINCT values, 外键关系)
  ├─> 步骤 3: 使用 LLM 分析每个列的业务含义
  │   例如: "customer_id" → "客户唯一标识符"
  ├─> 步骤 4: 生成表的业务描述
  │   例如: "orders 表存储所有订单信息"
  ├─> 步骤 5: 分析表之间的关系
  │   例如: "orders.customer_id → customers.id"
  ├─> 步骤 6: 提取业务实体
  │   例如: Customer, Order, Product
  └─> 步骤 7: 生成数据库整体摘要

# 3. 结果存储在 Context.data (JSON)
{
  "db_schema": { ... },
  "table_descriptions": { ... },
  "column_descriptions": { ... },
  "entity": { ... },
  "summary": "This is a sales database with customers, orders, and products"
}
```

---

## 数据库表关系

```
context (上下文表)
  ├─> type = "DATA"     → 数据库 Schema 分析结果
  ├─> type = "SESSION"  → 多轮对话历史
  └─> type = "DOMAIN"   → 领域知识库

eda_task_job (任务作业表)
  └─> 存储所有异步任务的状态和结果

history_log (AI 调用日志)
  └─> 记录每次 AI 调用 (问题、SQL、结果、Token)

user_preference (用户偏好)
  └─> 模型选择、语言、限流配置

domain_knowledge (知识库)
  └─> 术语表、示例 SQL、业务规则
```

---

## 环境变量配置

```bash
# 必需配置
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  # AI 算力服务
SENTRY_DSN=https://xxx@sentry.io/xxx       # 错误追踪
ENVIRONMENT=dev/staging/prod               # 环境标识
DISABLE_RATE_LIMIT=true                    # 禁用限流 (开发用)
```

---

## 本地开发

### 启动服务

```bash
# 1. 安装依赖
pip install -r requirements.txt

# 2. 启动 Redis 和 MySQL
docker run -d -p 6379:6379 redis
docker run -d -p 3306:3306 -e MYSQL_ROOT_PASSWORD=root mysql:8.0

# 3. 创建数据库
mysql -u root -p -e "CREATE DATABASE eda_service"

# 4. 数据库迁移
alembic upgrade head

# 5. 启动服务 (Web + Worker)
honcho start
# 或分别启动
# Terminal 1: ./runapp.sh      (Web Server)
# Terminal 2: ./runworker.sh   (Celery Worker)
```

### 测试 API

```bash
# 创建 Data Summary
curl -X POST http://localhost:5000/api/v1beta/apps/1/data_summaries \
  -H "Content-Type: application/json" \
  -u "public_key:private_key" \
  -d '{
    "cluster_id": "xxx",
    "database": "test",
    "description": "Test DB"
  }'

# Chat2Data
curl -X POST http://localhost:5000/api/v1beta/apps/1/endpoint/v3/chat2data \
  -H "Content-Type: application/json" \
  -u "public_key:private_key" \
  -d '{
    "cluster_id": "xxx",
    "database": "test",
    "question": "Show me all tables"
  }'

# 查询任务结果
curl http://localhost:5000/api/v1beta/apps/1/jobs/{job_id} \
  -u "public_key:private_key"
```

---

## 常见调试技巧

### 1. 查看日志

```bash
# Web Server 日志
tail -f logs/app.log

# Celery Worker 日志
tail -f logs/worker.log
```

### 2. 查看任务队列

```bash
# Redis 队列长度
redis-cli LLEN celery

# 查看任务详情
redis-cli LRANGE celery 0 -1
```

### 3. 查看数据库

```sql
-- 查看最近的任务
SELECT * FROM eda_task_job ORDER BY created_at DESC LIMIT 10;

-- 查看 AI 调用日志
SELECT * FROM history_log ORDER BY created_at DESC LIMIT 10;

-- 查看上下文
SELECT id, _type, name, org_id, cluster_id, db_name
FROM context ORDER BY created_at DESC LIMIT 10;
```

### 4. 手动触发任务 (调试用)

```python
from tasks.eda import task_chat2data
from models.base import Session

session = Session()

# 同步执行 (不通过 Celery)
result = task_chat2data.apply(args=[
    data_context_id,
    session_context_id,
    "Show me all customers",
    True,  # direct_resolve
    True,  # auto_run
    False, # sql_to_data
    False, # data_to_chart
    "",    # evidence
    1,     # app_id
    "xxx", # public_key
]).get()

print(result)
```

---

## 添加新功能示例

### 场景: 添加 SQL 格式化功能

#### 1. 定义任务 (`tasks/eda.py`)

```python
@app.task(bind=True, base=EDACerlryTaskBase)
def task_format_sql(self, sql, metrics=None):
    """格式化 SQL"""
    import sqlparse

    formatted_sql = sqlparse.format(
        sql,
        reindent=True,
        keyword_case='upper'
    )

    return {"formatted_sql": formatted_sql}
```

#### 2. 添加路由 (`handlers/edatask.py`)

```python
@edatask_v3_bp.post("/format_sql")
def handle_format_sql(data_app_id: int):
    payload = request.get_json()
    sql = payload["sql"]

    org_id = get_org_id_from_header()

    job_id = enqueue(
        task_format_sql, org_id, sql
    )

    return success(result={"job_id": job_id})
```

#### 3. 测试

```bash
curl -X POST http://localhost:5000/api/v1beta/apps/1/endpoint/v3/format_sql \
  -H "Content-Type: application/json" \
  -u "public_key:private_key" \
  -d '{"sql": "select * from users where id=1"}'
```

---

## 性能优化建议

### 1. 向量检索优化

```python
# 减少检索的表数量
relevant_tables = vector_search(question, top_k=5)  # 从 10 减到 5

# 使用缓存
@lru_cache(maxsize=1000)
def get_table_description(table_name):
    ...
```

### 2. LLM 调用优化

```python
# 使用更快的模型
ai_model = GPT_4O_MINI  # 而不是 GPT_4O

# 并行调用
import asyncio
results = await asyncio.gather(
    ai_completion_async("text2sql", prompt1),
    ai_completion_async("text2sql", prompt2)
)
```

### 3. 数据库优化

```sql
-- 添加索引
CREATE INDEX idx_context_cluster_db ON context(cluster_id, db_name);
CREATE INDEX idx_job_status ON eda_task_job(status, created_at);
```

---

## 常见问题 FAQ

### Q1: 为什么任务一直是 RUNNING 状态？

**A**: 可能原因：
1. Celery Worker 没有启动 → 运行 `./runworker.sh`
2. 任务执行超时 (> 10 分钟) → 检查 Worker 日志
3. Redis 连接失败 → 检查 `BROKER_URL` 配置

### Q2: SQL 生成质量不好怎么办？

**A**: 优化方向：
1. 完善数据库 Schema 描述 (运行 `task_gen_db_schema`)
2. 添加领域知识库 (domain_knowledge 表)
3. 调整 Prompt 模板 (`svc/eda_task.py`)
4. 切换更强的模型 (GPT-4o)

### Q3: 如何支持新的数据库？

**A**: 实现新的 Executor：

```python
# svc/executor.py
class NewDBExecutor:
    def execute(self, sql):
        # 连接数据库
        conn = connect_to_newdb()

        # 执行 SQL
        cursor = conn.execute(sql)

        # 返回结果
        return {
            "columns": [desc[0] for desc in cursor.description],
            "rows": cursor.fetchall()
        }
```

### Q4: 内存占用过高？

**A**: 优化方向：
1. 减少 Celery Worker 并发数 (默认 10)
2. 限制结果集大小 (`LIMIT 1000`)
3. 使用分页查询
4. 定期清理过期任务 (`mark_expired_jobs`)

---

## 学习路径建议

### 新人学习路径 (按顺序)

1. **第 1 天**: 理解整体架构
   - 阅读本文档 + `PROJECT_ARCHITECTURE.md`
   - 搭建本地开发环境
   - 运行一个简单的 Chat2Data 请求

2. **第 2-3 天**: 理解核心流程
   - 阅读 `tasks/eda.py` (任务定义)
   - 阅读 `svc/eda_task.py` (核心逻辑)
   - 单步调试 `task_chat2data` 流程

3. **第 4-5 天**: 理解数据模型
   - 阅读 `models/context.py`
   - 查看数据库表结构
   - 理解 Context 的作用

4. **第 6-7 天**: 实践开发
   - 修改一个 Prompt 模板
   - 添加一个简单的新功能
   - 写一个单元测试

5. **第 2 周**: 深入研究
   - 阅读 `svc/data_explore.py` (Schema 分析)
   - 理解向量检索 (ChromaDB)
   - 优化 SQL 生成质量

### 推荐阅读顺序

```
1. QUICK_START.md (本文) ⭐⭐⭐⭐⭐
2. PROJECT_ARCHITECTURE.md ⭐⭐⭐⭐
3. tasks/eda.py (任务定义) ⭐⭐⭐⭐⭐
4. svc/eda_task.py (核心逻辑) ⭐⭐⭐⭐⭐
5. handlers/edatask.py (API 路由) ⭐⭐⭐
6. models/context.py (数据模型) ⭐⭐⭐
7. svc/data_explore.py (Schema 分析) ⭐⭐
8. svc/openai.py (LLM 封装) ⭐⭐
```

---

## 重要提示

1. **不要直接修改线上数据库** - 使用迁移脚本 (Alembic)
2. **注意 API Key 安全** - 不要提交到代码仓库
3. **LLM Token 消耗** - 注意成本控制
4. **测试环境** - 使用 Spider/BIRD 数据集测试
5. **监控指标** - 关注 Prometheus 和 Sentry

---

## 获取帮助

- **文档**: `/docs/swagger.json`
- **日志**: 查看 Sentry 错误追踪
- **测试**: 运行 `pytest tests/`
- **问题**: 查看 GitHub Issues

---

**祝你快速上手！** 🎉

有问题随时查阅 `PROJECT_ARCHITECTURE.md` 获取更详细的信息。