00 快速复现清单：掌柜问数 — Text2SQL 智能体全栈系统

🏃 本文档目标：复制粘贴 + 回车 = 跑通完整项目 ❌ 不讲原理——原理在 01~07 篇 💬 卡住了？ 把错误信息复制给 AI："按清单第 X 步卡住了，错误信息：..."

0. 前置检查

bash

# 确认以下工具已安装
docker --version            # Docker ≥ 24
python --version            # Python ≥ 3.12
node --version              # Node ≥ 18（前端需要）

# 确认以下端口未被占用
# 3306(MySQL) 9200(ES) 5601(Kibana) 6333+6334(Qdrant) 8081(Embedding)

1. 阶段一：启动基础设施（≈10 min）

Step 1.1 创建项目目录

bash

mkdir data-agent && cd data-agent
mkdir -p conf logs prompts app/{agent/nodes,api/{routers,schemas},clients,conf,core,entities,models,prompt,repositories/{mysql/{meta/mappers,dw},qdrant,es},scripts,services}
touch app/__init__.py app/agent/__init__.py app/agent/nodes/__init__.py app/api/__init__.py app/api/routers/__init__.py app/api/schemas/__init__.py app/clients/__init__.py app/conf/__init__.py app/core/__init__.py app/entities/__init__.py app/models/__init__.py app/prompt/__init__.py app/repositories/__init__.py app/repositories/es/__init__.py app/repositories/mysql/__init__.py app/repositories/mysql/dw/__init__.py app/repositories/mysql/meta/__init__.py app/repositories/mysql/meta/mappers/__init__.py app/repositories/qdrant/__init__.py app/scripts/__init__.py app/services/__init__.py

Step 1.2 启动 Docker 服务

bash

# 将资料中的 docker-compose 目录拷贝到项目根目录
# 进入 docker-compose.yaml 所在目录
docker compose up -d

# 验证各服务
curl -s http://localhost:6333 > /dev/null && echo "✅ Qdrant OK" || echo "❌ Qdrant"
curl -s http://localhost:9200 > /dev/null && echo "✅ ES OK" || echo "❌ ES"
curl -s http://localhost:3306 > /dev/null && echo "✅ MySQL OK" || echo "❌ MySQL"
curl -s http://localhost:8081/health > /dev/null 2>&1 && echo "✅ Embedding OK" || echo "❌ Embedding"

Step 1.3 创建 Python 虚拟环境

bash

conda create -n py312 python=3.12 -y
conda activate py312

pip install asyncmy cryptography "elasticsearch[async]>=8,<9" fastapi[standard] huggingface-hub jieba langchain langchain-deepseek langchain-huggingface langgraph loguru omegaconf pyyaml qdrant-client sqlalchemy

Step 1.4 创建配置文件

conf/app_config.yaml：

yaml

logging:
  file:
    enable: true
    level: INFO
    path: logs
    rotation: "10 MB"
    retention: "7 days"
  console:
    enable: true
    level: INFO

db_meta:
  host: localhost; port: 3306; user: atguigu; password: Atguigu.123; database: meta
db_dw:
  host: localhost; port: 3306; user: atguigu; password: Atguigu.123; database: dw
qdrant:
  host: localhost; port: 6333; embedding_size: 1024
embedding:
  host: localhost; port: 8081; model: BAAI/bge-large-zh-v1.5
es:
  host: localhost; port: 9200; index_name: data_agent
llm:
  model_name: deepseek-chat
  api_key: <你的API_KEY>

2. 阶段二：初始化数据库 + Embedding 验证（≈5 min）

bash

# 验证 MySQL 初始化
python -c "
import asyncio
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine
async def check():
    engine = create_async_engine('mysql+asyncmy://atguigu:Atguigu.123@localhost:3306/meta?charset=utf8mb4')
    async with engine.connect() as conn:
        result = await conn.execute(text('show tables;'))
        print('meta tables:', [r[0] for r in result])
    engine2 = create_async_engine('mysql+asyncmy://atguigu:Atguigu.123@localhost:3306/dw?charset=utf8mb4')
    async with engine2.connect() as conn:
        result = await conn.execute(text('show tables;'))
        print('dw tables:', [r[0] for r in result])
asyncio.run(check())
"

# 验证 Embedding 服务
curl http://localhost:8081/embed -X POST -H "Content-Type: application/json" -d '{"inputs":"hello world"}'

3. 阶段三：构建元数据知识库（≈10 min）

Step 3.1 创建 meta_config.yaml

从课程资料中获取 conf/meta_config.yaml，或直接从完整代码复制。

Step 3.2 编写代码并执行

bash

# 确保所有代码文件已就位（从完整代码复制 app/ 目录）
python -m app.scripts.build_meta_knowledge --conf conf/meta_config.yaml

预期输出：

加载配置文件
保存表信息到meta数据库
为字段信息建立向量索引
为字段取值建立全文索引
保存指标信息到meta数据库
为指标信息建立向量索引
元数据知识库构建完成

Step 3.3 验证

bash

# 验证 Qdrant 有无数据
curl http://localhost:6333/collections/data-agent-column | jq .result.status
curl http://localhost:6333/collections/data-agent-metric | jq .result.status

# 验证 ES 有无索引
curl http://localhost:9200/data-agent-value/_count

4. 阶段四：测试智能体工作流（≈5 min）

bash

python -c "
import asyncio
from app.agent.graph import graph
from app.agent.state import DataAgentState
from app.agent.context import DataAgentContext
from app.clients.embedding_client_manager import embedding_client_manager
from app.clients.es_client_manager import es_client_manager
from app.clients.mysql_client_manager import meta_mysql_client_manager, dw_mysql_client_manager
from app.clients.qdrant_client_manager import qdrant_client_manager
from app.repositories.es.value_es_repository import ValueESRepository
from app.repositories.mysql.dw.dw_mysql_repository import DWMySQLRepository
from app.repositories.mysql.meta.meta_mysql_repository import MetaMySQLRepository
from app.repositories.qdrant.column_qdrant_repository import ColumnQdrantRepository
from app.repositories.qdrant.metric_qdrant_repository import MetricQdrantRepository

async def test():
    # 初始化
    embedding_client_manager.init()
    qdrant_client_manager.init()
    es_client_manager.init()
    meta_mysql_client_manager.init()
    dw_mysql_client_manager.init()

    async with meta_mysql_client_manager.session_factory() as meta_session, dw_mysql_client_manager.session_factory() as dw_session:
        context = DataAgentContext(
            embedding_client=embedding_client_manager.client,
            column_qdrant_repository=ColumnQdrantRepository(qdrant_client_manager.client),
            value_es_repository=ValueESRepository(es_client_manager.client),
            metric_qdrant_repository=MetricQdrantRepository(qdrant_client_manager.client),
            meta_mysql_repository=MetaMySQLRepository(meta_session),
            dw_mysql_repository=DWMySQLRepository(dw_session)
        )
        state = DataAgentState(query='统计去年各地区的销售总额')
        async for chunk in graph.astream(input=state, context=context, stream_mode='custom'):
            print(chunk)

    await qdrant_client_manager.close()
    await es_client_manager.close()
    await meta_mysql_client_manager.close()
    await dw_mysql_client_manager.close()

asyncio.run(test())
"

5. 阶段五：启动 API 服务（≈3 min）

bash

# 在 main.py 所在目录执行
fastapi dev main.py

# 或使用 uvicorn
uvicorn main:app --host 0.0.0.0 --port 8000

验证 API

bash

curl -X POST "http://localhost:8000/api/query" \
  -H "Content-Type: application/json" \
  -d '{"query": "统计去年各地区的销售总额"}'

预期返回 SSE 流式数据：

data: {"type": "progress", "step": "抽取关键字", "status": "running"}
data: {"type": "progress", "step": "抽取关键字", "status": "success"}
data: {"type": "progress", "step": "召回字段", "status": "running"}
...
data: {"type": "result", "data": [{"region_name": "华东", "order_amount": 150000}, ...]}

6. 启动前端（可选，≈3 min）

bash

# 进入前端目录
cd data-agent-fronted
npm install
npm run dev
# 浏览器访问 http://localhost:5173

7. 验证清单

编号	验证项	命令	预期结果
1	Docker 全部启动	`docker ps --format "{{.Names}}"`	看到 mysql/es/qdrant/kibana/embedding
2	MySQL 表结构	`python -c "..."` (见阶段二)	看到 meta 库 4 张表 + dw 库 5 张表
3	Qdrant 集合	`curl localhost:6333/collections`	看到 data-agent-column + data-agent-metric
4	ES 索引	`curl localhost:9200/_cat/indices`	看到 data-agent-value
5	元数据构建	阶段三日志	6 步全部成功
6	智能体测试	阶段四 Python 脚本	输出各节点进度 + 最终结果
7	API 接口	`curl POST /api/query`	SSE 流式返回数据
8	前端页面	浏览器访问	看到查询界面，可输入查询

00 快速复现清单：掌柜问数 — Text2SQL 智能体全栈系统 ​

0. 前置检查 ​

1. 阶段一：启动基础设施（≈10 min） ​

Step 1.1 创建项目目录 ​

Step 1.2 启动 Docker 服务 ​

Step 1.3 创建 Python 虚拟环境 ​

Step 1.4 创建配置文件 ​

2. 阶段二：初始化数据库 + Embedding 验证（≈5 min） ​

3. 阶段三：构建元数据知识库（≈10 min） ​

Step 3.1 创建 meta_config.yaml ​

Step 3.2 编写代码并执行 ​

Step 3.3 验证 ​

4. 阶段四：测试智能体工作流（≈5 min） ​

5. 阶段五：启动 API 服务（≈3 min） ​

验证 API ​

6. 启动前端（可选，≈3 min） ​

7. 验证清单 ​

00 快速复现清单：掌柜问数 — Text2SQL 智能体全栈系统

0. 前置检查

1. 阶段一：启动基础设施（≈10 min）

Step 1.1 创建项目目录

Step 1.2 启动 Docker 服务

Step 1.3 创建 Python 虚拟环境

Step 1.4 创建配置文件

2. 阶段二：初始化数据库 + Embedding 验证（≈5 min）

3. 阶段三：构建元数据知识库（≈10 min）

Step 3.1 创建 meta_config.yaml

Step 3.2 编写代码并执行

Step 3.3 验证

4. 阶段四：测试智能体工作流（≈5 min）

5. 阶段五：启动 API 服务（≈3 min）

验证 API

6. 启动前端（可选，≈3 min）

7. 验证清单