MCP智能问数

yqj234

209人浏览 · 2026-10-22 11:17:53

yqj234 · 2026-10-22 11:17:53 发布

用 ‌MCP（模型上下文协议）‌ 实现智能问数，核心思路是：‌让大模型通过标准协议连接数据库工具，把自然语言问题自动转成 SQL 查询并返回结果‌。相当于给 AI 装了一个“数据库插头”，不用人工写 SQL，直接对话就能查数。

🧠 整体架构
‌MCP Server（数据库连接器）‌：封装数据库连接和查询能力，暴露成标准工具（如 list_tables、get_table_schema、execute_query）。
‌MCP 客户端（Agent 编排层）‌：对接大模型和 MCP Server，把用户的自然语言问题交给模型，模型判断该调用哪个工具、传什么参数。
‌大模型（NL2SQL）‌：负责理解问题、生成 SQL、解析工具返回结果，最后组织成自然语言回答。
‌知识库（可选，增强准确性）‌：把表结构 DDL 和示例 SQL 放进去，让模型在生成 SQL 前先“看到”表长什么样，避免瞎编表名和字段。
🔧 具体实现步骤
‌部署数据库 MCP Server‌：可以用现成开源项目，如 DBHub（支持 MySQL、PostgreSQL、SQL Server），或 universal-db-mcp（支持 MySQL、PostgreSQL、SQLite）。也可以用 Spring AI（Java）或 FastMCP（Python）自己写，暴露 list_tables、get_table_schema、execute_query 这几个核心工具。
‌接入 Agent 编排平台‌：主流选择有 Dify、MaxKB、Coze、LangGraph。在平台里添加 MCP 客户端，填 MCP Server 的地址（SSE 或 Streamable HTTP）。
‌配置大模型‌：选支持工具调用（Function Calling）的模型，如 DeepSeek Chat、Qwen-Plus，填好 Base URL、API Key、Model ID。
‌编写提示词（关键）‌：告诉模型什么情况调用哪个工具，例如“用户问数据时，先调用 list_tables 看有哪些表，再调用 get_table_schema 了解表结构，然后生成 SQL 调用 execute_query”。建议给出几个示例问题，模型理解更准。
‌安全加固‌：数据库账号务必用‌只读账号‌（仅授予 SELECT 权限），工具调用保持‌手动确认‌，防止模型生成 DELETE、UPDATE 等危险语句。
⚠️ 常见坑与建议
‌模型不调用工具‌：多半是提示词约束不够，或模型不支持工具调用，换 DeepSeek Chat、Qwen-Plus 这类实测可用的模型。
‌SQL 生成错误‌：多半是没拿到表结构，确保模型在生成 SQL 前先调 get_table_schema，并在知识库里补充字段注释和示例 SQL。
‌配置报错‌：401 检查 API Key；local proxy failed 检查 npx 路径和数据库连接；reading 'choices' 多半是 Model ID 填错。
‌别直连生产库‌：只读账号 + 手动确认是底线，最好用独立测试库先验证。
‌查询超时‌：复杂查询加绘图可能耗时较长，MCP 配置里把 timeout 调到 180 秒左右。

收集元数据

"""元数据采集：把 MySQL 的表结构 + 字段注释，转成可向量化的文档

产出文档格式（每张表一条）：
    表名：t_workorder（生产工单主表）
    用途：……
    主键：id
    字段：
      - id              bigint        工单ID（主键）
      - code            varchar(32)   工单编号
      ...
    关联：
      - material_id -> t_material.id
"""

import logging

import pymysql
from pymysql.cursors import DictCursor

from . import load_config

logger = logging.getLogger(__name__)


class MetaCollector:
    def __init__(self, cfg: dict | None = None):
        cfg = cfg or load_config()
        self.cfg = cfg.get("database", {})

    # ------------------------------------------------------------------
    def _connect(self) -> pymysql.connections.Connection:
        return pymysql.connect(
            host=self.cfg.get("host"),
            port=int(self.cfg.get("port", 3306)),
            user=self.cfg.get("user"),
            password=self.cfg.get("password"),
            database=self.cfg.get("database"),
            charset=self.cfg.get("charset", "utf8mb4"),
            cursorclass=DictCursor,
            connect_timeout=int(self.cfg.get("connect_timeout", 10)),
        )

    # ------------------------------------------------------------------
    def fetch_table_columns(self) -> dict[str, list[dict]]:
        """一次性读取库里所有表的所有字段"""

        sql1 = """
            SELECT TABLE_NAME        AS table_name,
                   TABLE_COMMENT     AS table_comment,
                   COLUMN_NAME       AS column_name,
                   COLUMN_TYPE       AS column_type,
                   IS_NULLABLE       AS is_nullable,
                   COLUMN_KEY        AS column_key,
                   COLUMN_COMMENT    AS column_comment
              FROM information_schema.COLUMNS
             WHERE TABLE_SCHEMA = %s
             ORDER BY TABLE_NAME, ORDINAL_POSITION
        """
        
        sql="""           SELECT a.TABLE_NAME        AS table_name,
               	   b.TABLE_COMMENT     AS table_comment,
                   a.COLUMN_NAME       AS column_name,
                   a.COLUMN_TYPE       AS column_type,
                   a.IS_NULLABLE       AS is_nullable,
                   a.COLUMN_KEY        AS column_key,
                   a.COLUMN_COMMENT    AS column_comment
              FROM information_schema.COLUMNS a inner join information_schema.TABLES b
              on a.TABLE_CATALOG =b.TABLE_CATALOG and a.TABLE_SCHEMA =b.TABLE_SCHEMA and  a.TABLE_NAME =b.TABLE_NAME 
             WHERE a.TABLE_SCHEMA = %s
               AND substring(a.table_name,1,3) <> 'act'
             ORDER BY a.TABLE_NAME, a.ORDINAL_POSITION;
        """

        result: dict[str, dict] = {}
        with self._connect() as conn, conn.cursor() as cur:
            cur.execute(sql, (self.cfg.get("database"),))
            for row in cur.fetchall():
                t = row["table_name"]
                if t not in result:
                    result[t] = {"comment": row["table_comment"] or "", "columns": []}
                result[t]["columns"].append(row)
        return result

    # ------------------------------------------------------------------
    def fetch_foreign_keys(self) -> list[dict]:
        """读取外键关系，用于提示表之间的关联"""
        sql = """
            SELECT TABLE_NAME             AS table_name,
                   COLUMN_NAME            AS column_name,
                   REFERENCED_TABLE_NAME  AS ref_table,
                   REFERENCED_COLUMN_NAME AS ref_column
              FROM information_schema.KEY_COLUMN_USAGE
             WHERE TABLE_SCHEMA = %s
               AND REFERENCED_TABLE_NAME IS NOT NULL
        """
        with self._connect() as conn, conn.cursor() as cur:
            cur.execute(sql, (self.cfg.get("database"),))
            return cur.fetchall()

    # ------------------------------------------------------------------
    def build_documents(self) -> list[dict]:
        """组装成向量库文档列表 [{id, text, metadata}]"""
        tables = self.fetch_table_columns()
        fks = self.fetch_foreign_keys()

        fk_map: dict[str, list[str]] = {}
        for fk in fks:
            fk_map.setdefault(fk["table_name"], []).append(
                f"{fk['column_name']} -> {fk['ref_table']}.{fk['ref_column']}"
            )

        docs: list[dict] = []
        for table, info in tables.items():
            lines = [f"表名：{table}（{info['comment'] or '无注释'}）"]

            pk_cols = [c["column_name"] for c in info["columns"] if c["column_key"] == "PRI"]
            if pk_cols:
                lines.append("主键：" + ", ".join(pk_cols))

            lines.append("字段：")
            for col in info["columns"]:
                comment = col["column_comment"] or "无注释"
                lines.append(f"  - {col['column_name']}  {col['column_type']}  {comment}")

            if table in fk_map:
                lines.append("关联：")
                for rel in fk_map[table]:
                    lines.append(f"  - {rel}")

            text = "\n".join(lines)
            docs.append({
                "id": table,
                "text": text,
                "metadata": {
                    "table_name": table,
                    "table_comment": info["comment"] or "",
                    "column_count": len(info["columns"]),
                },
            })

        logger.info("元数据文档构建完成，共 %d 张表", len(docs))
        return docs

CREATE TABLE mes_meta.meta_table_info (
  id            BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
  -- ===== 系统层：脚本每天同步，只更新这些列 =====
  db_schema     VARCHAR(64)  NOT NULL DEFAULT '' COMMENT '数据库名',
  table_name    VARCHAR(64)  NOT NULL         COMMENT '物理表名',
  table_comment VARCHAR(255) NOT NULL DEFAULT '' COMMENT '表注释(information_schema)',
  column_count  INT          NOT NULL DEFAULT 0  COMMENT '字段数',
  sync_time     DATETIME     NULL             COMMENT '最近系统同步时间',
  -- ===== 业务层：人工维护，同步脚本不碰 =====
  biz_name      VARCHAR(64)  NULL COMMENT '业务中文名，如：生产工单主表',
  biz_desc      TEXT         NULL COMMENT '业务说明：用途、在业务链路中的位置',
  labels        VARCHAR(255) NULL COMMENT '标签(逗号分隔)，供检索',
  owner         VARCHAR(64)  NULL COMMENT '数据责任人/维护人',
  importance    TINYINT      NULL DEFAULT 1 COMMENT '重要性 1-5',
  refresh_freq  VARCHAR(32)  NULL COMMENT '数据刷新频率：实时/每小时/每日',
  biz_notes     TEXT         NULL COMMENT '注意事项',
  -- ===== 公共 =====
  is_active     TINYINT      NOT NULL DEFAULT 1 COMMENT '0=表已下线但保留业务说明(软删)',
  created_at    DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP,
  updated_at    DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  UNIQUE KEY uk_table (table_name)          -- 自然键，不是自增 id
) ENGINE=InnoDB COMMENT='表级元数据：系统层自动同步+业务层人工维护';

设计总体思路

核心就一句话：把「谁维护什么」分开。

层	内容	维护者	同步方式
系统层	表名、字段、类型、注释、外键	脚本	每天定时全量刷新，只覆盖系统列
业务层	业务说明、口径、枚举值、注意事项、责任人	人工	永不被同步脚本覆盖

最终效果：采集时把「系统层 + 业务层」拼装成文档喂给 LLM，提示词从"纯字段清单"升级为"带业务口径的字段清单"，SQL 生成质量会明显提升。

一个关键机制：同步用「自然键 UPSERT + 只更新系统列」，保证每天自动跑时，人工维护的业务内容不会被冲掉。

一、表结构设计（建议独立 schema：`mes_meta`）

不直接建在一张 meta_table_info 里，而是拆成 4 张表——按级别分开，避免表级/字段级/关系级信息互相冗余，这也是数仓 metadata 的标准分层。命名上保留你提的 meta_table_info。

1. `meta_table_info`（表级）

sql

CREATE DATABASE IF NOT EXISTS mes_meta DEFAULT CHARSET utf8mb4;

CREATE TABLE mes_meta.meta_table_info (
  id            BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
  -- ===== 系统层：脚本每天同步，只更新这些列 =====
  db_schema     VARCHAR(64)  NOT NULL DEFAULT '' COMMENT '数据库名',
  table_name    VARCHAR(64)  NOT NULL         COMMENT '物理表名',
  table_comment VARCHAR(255) NOT NULL DEFAULT '' COMMENT '表注释(information_schema)',
  column_count  INT          NOT NULL DEFAULT 0  COMMENT '字段数',
  sync_time     DATETIME     NULL             COMMENT '最近系统同步时间',
  -- ===== 业务层：人工维护，同步脚本不碰 =====
  biz_name      VARCHAR(64)  NULL COMMENT '业务中文名，如：生产工单主表',
  biz_desc      TEXT         NULL COMMENT '业务说明：用途、在业务链路中的位置',
  labels        VARCHAR(255) NULL COMMENT '标签(逗号分隔)，供检索',
  owner         VARCHAR(64)  NULL COMMENT '数据责任人/维护人',
  importance    TINYINT      NULL DEFAULT 1 COMMENT '重要性 1-5',
  refresh_freq  VARCHAR(32)  NULL COMMENT '数据刷新频率：实时/每小时/每日',
  biz_notes     TEXT         NULL COMMENT '注意事项',
  -- ===== 公共 =====
  is_active     TINYINT      NOT NULL DEFAULT 1 COMMENT '0=表已下线但保留业务说明(软删)',
  created_at    DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP,
  updated_at    DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  UNIQUE KEY uk_table (table_name)          -- 自然键，不是自增 id
) ENGINE=InnoDB COMMENT='表级元数据：系统层自动同步+业务层人工维护';

2. `meta_column_info`（字段级）—— 核心价值所在

每张表的每个字段一条，业务层的指标口径、枚举值、敏感标记都在这：

sql

CREATE TABLE mes_meta.meta_column_info (
  id                BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
  -- ===== 系统层 =====
  table_name        VARCHAR(64)  NOT NULL COMMENT '物理表名',
  column_name       VARCHAR(64)  NOT NULL COMMENT '字段名',
  column_type       VARCHAR(64)  NOT NULL DEFAULT '' COMMENT '字段类型',
  is_nullable       VARCHAR(3)   NOT NULL DEFAULT '' COMMENT '是否可空',
  column_key        VARCHAR(3)   NOT NULL DEFAULT '' COMMENT 'PRI/UNI/MUL',
  column_comment    VARCHAR(255) NOT NULL DEFAULT '' COMMENT '字段注释',
  ordinal_position  INT          NOT NULL DEFAULT 0,
  sync_time         DATETIME     NULL,
  -- ===== 业务层 =====
  biz_name          VARCHAR(64)  NULL COMMENT '字段业务名，如：工单编号',
  biz_desc          TEXT         NULL COMMENT '字段业务含义',
  metric_definition TEXT         NULL COMMENT '指标口径/计算公式，如 完成率=完成数/下达数',
  enum_values       TEXT         NULL COMMENT '枚举值说明，如 0-未开始;1-进行中;2-已完成',
  is_sensitive      TINYINT      NOT NULL DEFAULT 0 COMMENT '1=敏感字段(手机号/身份证)',
  qa_notes          TEXT         NULL COMMENT '使用注意事项/常见坑',
  is_active         TINYINT      NOT NULL DEFAULT 1,
  created_at        DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP,
  updated_at        DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  UNIQUE KEY uk_col (table_name, column_name)   -- 复合自然键
) ENGINE=InnoDB COMMENT='字段级元数据';

3. `meta_relation_info`（表间关系，可选但强烈建议）

系统层自动同步外键，业务层补"主从/关联"语义——现在是靠外键+LLM 猜，有了这个描述更准：

sql

CREATE TABLE mes_meta.meta_relation_info (
  id          BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
  -- ===== 系统层 =====
  table_name  VARCHAR(64) NOT NULL,
  column_name VARCHAR(64) NOT NULL,
  ref_table   VARCHAR(64) NOT NULL DEFAULT '',
  ref_column  VARCHAR(64) NOT NULL DEFAULT '',
  sync_time   DATETIME    NULL,
  -- ===== 业务层 =====
  rel_type    VARCHAR(20) NOT NULL DEFAULT 'fk' COMMENT 'fk/parent_child/lookup/business',
  rel_desc    TEXT        NULL COMMENT '业务关系说明，如：一单多料明细',
  owner       VARCHAR(64) NULL,
  is_active   TINYINT     NOT NULL DEFAULT 1,
  created_at  DATETIME    NOT NULL DEFAULT CURRENT_TIMESTAMP,
  updated_at  DATETIME    NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  UNIQUE KEY uk_rel (table_name, column_name, ref_table, ref_column)
) ENGINE=InnoDB COMMENT='表间关系元数据';

4. `meta_sync_log`（同步日志，治理用）

id / sync_type(table|column|relation) / target_count / status / message / start_time / end_time。

二、两层同步策略（核心机制）

难点：每天自动同步时不能冲掉人工维护的业务列。

做法 = UPSERT + 只更新系统列，同步脚本长这样（示例 SQL，体现"系统列写、业务列不动"）：

sql

-- 表级同步
INSERT INTO mes_meta.meta_table_info
  (db_schema, table_name, table_comment, column_count, sync_time, is_active)
VALUES ('ryvue_mes', 't_workorder', '生产工单主表', 12, NOW(), 1)
ON DUPLICATE KEY UPDATE
  db_schema     = VALUES(db_schema),
  table_comment = VALUES(table_comment),
  column_count  = VALUES(column_count),
  sync_time     = VALUES(sync_time),
  is_active     = 1;
-- 注意：业务列(biz_name/biz_desc/...)不在 UPDATE 列表里 → 人工内容保留

完整流程（每天一次，建议凌晨 02:00，避开业务高峰）：

从 information_schema 抽取全量表/字段/外键
按自然键对 meta_* 表做 upsert，只覆写系统列
新增表/字段 → 插入新行（业务列留 NULL，等人工补）
消失的表/字段 → 不删行，置 is_active=0（保留业务说明，防止丢人工成果）
写 meta_sync_log

调度方式三选一：

应用内 APScheduler：python main.py 时注册「启动执行一次 + 每天 02:00」（README 已把 APScheduler 列为演进项，正好接这个）
Windows 任务计划程序 / Linux cron 调一个独立脚本（更适合「脚本自动同步，应用只管读」的职责分离）
保留现有 /api/meta/sync 手动触发作为兜底

人工维护示例（不碰系统列，只改业务列）：

sql

UPDATE mes_meta.meta_table_info
SET biz_name='生产工单主表',
    biz_desc='工单下达、排产、报工的主记录表',
    owner='张三'
WHERE table_name='t_workorder';

UPDATE mes_meta.meta_column_info
SET metric_definition='完成率 = 已完成数 / 下达数',
    enum_values='0-未开始;1-进行中;2-已完成;3-已删除',
    qa_notes='status=3 是软删记录，统计时请过滤'
WHERE table_name='t_workorder' AND column_name='status';

三、与现有 NL2SQL 的衔接（概念设计，不改代码）

现状链路：information_schema → MetaCollector → build_documents → Chroma → LLM 提示词

设计后链路：

代码块

information_schema ──┐
                     ├─→ 采集(读 mes_meta.* 拼装) ─→ build_documents ─→ Chroma ─→ LLM
人工维护的 mes_meta.* ┘

变化点（都发生在 meta_collector 内部，对外接口不变）：

数据源切换：不再直连 information_schema 拼文档，而是先同步到 mes_meta 表，再以 mes_meta 为源拼文档——业务层内容自动进入向量库。
文档格式升级：每张表一条 doc，包含系统层 + 业务层，例如：

代码块

表名：t_workorder（生产工单主表）
标签：工单,生产    责任人：张三
业务说明：工单下达、排产、报工的主记录表
主键：id
字段：
  - id       bigint      工单ID（主键）
  - code     varchar(32) 工单编号 | 口径/枚举：-
  - status   tinyint     工单状态 | 枚举：0-未开始;1-进行中;2-已完成;3-已删除
指标口径：
  - 完成率 = 已完成数 / 下达数
关联：
  - material_id -> t_material.id（一单对应一个物料）
注意事项：
  - status=3 为软删记录，统计时请过滤

这样 LLM 看到的不是"status tinyint"而是带枚举和软删提示的完整业务语义，生成的 SQL 会准很多。 3. 人工只改业务层时：可只对改动的表做 doc 增量 upsert（id 仍是 table_name），不必全量重建 Chroma。

四、上线建议与注意点

schema 隔离：mes_meta 独立 schema 放同一实例，或放独立元数据库；别混进业务库，避免污染生产表。
权限：同步脚本对 mes_meta 仅 insert/update 系统列的权限；业务列写权限只授给维护人（BI/数仓组）。
幂等：一切走自然键 upsert，不要依赖自增 id 做更新依据。
首次上线别贪多：先跑一次系统层同步（186 张表全自动），再让业务方按高价值表优先补业务层（前 20~30 张常用表），比一口气全补更现实。
软删保护：is_active=0 机制务必保留，这是防止每天同步误删人工成果的最后防线。

开发实例说明

下面是在你 Windows 电脑上完整跑起来的步骤（PowerShell 语法，路径都已对准你的项目位置）。

前置准备（一次性）

组件	装什么	说明
Python	你已有（.venv 里是 3.12）	要求 ≥3.10 ✅
Docker Desktop	推荐装	跑 MySQL 和 PG+pgvector 最省事（pgvector 在 Windows 原生 PG 上不好装，用容器最稳）
Ollama	ollama.com/download 装 Windows 版	本地模型，装在默认 11434 端口

dev 环境配置（config/dev.yaml）默认指向本机，你只需要把下面三个服务起在默认端口上，不用改配置：

MySQL 127.0.0.1:3306（账号 ai_ro/123456，库 mes）
PG 127.0.0.1:5432（账号 ai/123456，库 ai_meta）
Ollama 127.0.0.1:11434（模型 qwen2.5-coder:7b + bge-m3）
另外 dev 里 auth.enabled: false，不需要 API Key，且 /docs Swagger 可用

第 1 步：装依赖 + 先跑单测（不依赖任何外部服务）

powershell

cd D:\PycharmProjects\mes_ai_service
$env:PYTHONUTF8=1        # 防止中文 Windows 下 GBK 解码崩溃
.\.venv\Scripts\python.exe -m pip install -r requirements.txt
.\.venv\Scripts\python.exe -m pytest -q

看到 25+ 个 passed 就说明代码环境没问题。这一步不用等 MySQL/PG/模型，建议先跑。

第 2 步：起 MySQL 并造测试数据

powershell

docker run -d --name mes-mysql -p 3306:3306 -e MYSQL_ROOT_PASSWORD=root123 mysql:8.0

然后初始化（建只读账号 + 两张带中文注释的示例表，注释很关键——RAG 靠它理解业务）：

sql

CREATE DATABASE IF NOT EXISTS mes DEFAULT CHARSET utf8mb4;
CREATE USER 'ai_ro'@'%' IDENTIFIED BY '123456';
GRANT SELECT ON mes.* TO 'ai_ro'@'%';
FLUSH PRIVILEGES;

USE mes;
CREATE TABLE t_workorder (
  id INT PRIMARY KEY AUTO_INCREMENT COMMENT '主键',
  code VARCHAR(32) NOT NULL COMMENT '工单编号',
  status TINYINT NOT NULL DEFAULT 0 COMMENT '状态: 0待生产 1生产中 2已完工',
  qty DECIMAL(10,2) COMMENT '数量',
  plan_finish_time DATETIME COMMENT '计划完成时间',
  create_time DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间'
) COMMENT='生产工单表';

CREATE TABLE t_material (
  id INT PRIMARY KEY AUTO_INCREMENT COMMENT '主键',
  mat_code VARCHAR(32) COMMENT '物料编码',
  mat_name VARCHAR(100) COMMENT '物料名称',
  stock_qty DECIMAL(12,2) COMMENT '库存数量',
  unit VARCHAR(10) COMMENT '单位'
) COMMENT='物料库存表';

INSERT INTO t_workorder(code,status,qty,plan_finish_time) VALUES
 ('WO20250701-001',0,100,'2025-07-20 18:00:00'),
 ('WO20250701-002',1,500,'2025-07-18 08:00:00'),
 ('WO20250702-003',2,80,'2025-07-10 18:00:00');
INSERT INTO t_material(mat_code,mat_name,stock_qty,unit) VALUES
 ('M-001','轴承 6204',1200,'件'),
 ('M-002','液压油 46#',-50,'L');   -- 故意留条负库存，测摘要的"异常点"识别

连库方式：docker exec -it mes-mysql mysql -uroot -proot123 进去执行，或者用 Navicat/DataGrip。

第 3 步：起 PostgreSQL + pgvector

powershell

docker run -d --name mes-pg -p 5432:5432 -e POSTGRES_USER=ai -e POSTGRES_PASSWORD=123456 -e POSTGRES_DB=ai_meta pgvector/pgvector:pg16

镜像必须用 pgvector/pgvector:pg16（内置 vector 扩展），用官方 postgres:16 会报"vector 扩展不存在"。服务启动时会自动建表、建 HNSW 索引，你不用写任何 DDL。

第 4 步：起 Ollama 并拉模型

powershell

ollama pull qwen2.5-coder:7b    # 约 4.7GB，生成 SQL 用
ollama pull bge-m3              # 约 1.2GB，向量化用，必须 1024 维（对应 dev.yaml embedding_dim）

机器内存 <16G 的话，SQL 生成可换 qwen2.5-coder:3b（改 dev.yaml 的 model_name 即可，其他不动）；bge-m3 别换，换模型维度对不上要重建向量表。CPU 跑 7b 每问 10~30 秒属正常。

第 5 步：启动服务

powershell

cd D:\PycharmProjects\mes_ai_service
$env:APP_ENV="dev"
.\.venv\Scripts\python.exe -m uvicorn src.api.app:app --reload --port 8010

第 6 步：首次同步元数据（问数前必须做）

另开一个 PowerShell 窗口：

powershell

cd D:\PycharmProjects\mes_ai_service
$env:APP_ENV="dev"
.\.venv\Scripts\python.exe scripts\sync_meta_task.py --once

看到 同步完成：{'success': True, 'tables': 2, ...} 即可。

第 7 步：冒烟测试

powershell

# 1) 健康检查（两个都应为 ok / true）
Invoke-RestMethod http://127.0.0.1:8010/health
Invoke-RestMethod http://127.0.0.1:8010/ready

# 2) 看已同步的表
Invoke-RestMethod http://127.0.0.1:8010/api/v1/nl2sql/tables

# 3) 真正问数（核心链路全走一遍）
Invoke-RestMethod -Method Post http://127.0.0.1:8010/api/v1/nl2sql/ask `
  -ContentType 'application/json' `
  -Body '{"question":"查最近3条未完工的生产工单","summary":true}'

# 4) 只生成 SQL 不执行（人工审核模式）
Invoke-RestMethod -Method Post http://127.0.0.1:8010/api/v1/nl2sql/generate `
  -ContentType 'application/json' -Body '{"question":"哪些物料库存为负数"}'

# 5) 安全校验（应返回 valid:false，关键字拦截）
Invoke-RestMethod -Method Post http://127.0.0.1:8010/api/v1/nl2sql/validate `
  -ContentType 'application/json' -Body '{"sql":"DELETE FROM t_workorder"}'

或者最省事的方式：**浏览器打开 http://127.0.0.1:8010/docs**（dev 环境才有 Swagger），点按钮把所有接口都点一遍。

第 8 步：MCP 测试（可选）

验证工具注册：Invoke-RestMethod http://127.0.0.1:8010/mcp/tools（应列出 6 个工具）
Cherry Studio / Dify：添加 MCP 服务器，SSE 地址填 http://127.0.0.1:8010/mcp/sse
stdio 模式：.\.venv\Scripts\python.exe -m src.mcp.server（在客户端配置里填这个命令）

常见问题（Windows 实测向）

现象	处理
pip / uvicorn 输出乱码或 GBK 崩溃	先 `$env:PYTHONUTF8=1` 再跑
`/ready` 里 mysql=false	ai_ro 账号没建 / GRANT 没给 / 端口被占用（`netstat -ano \| findstr :3306`）
`vector 扩展不存在`	PG 镜像用错了，换 `pgvector/pgvector:pg16` 重建容器
同步时 embedding 报错	`ollama pull bge-m3` 没拉，或维度不是 1024
问数很慢	CPU 跑 7b 正常现象；换 3b 或上 GPU
端口被占	3306/5432/8010 都被本机老服务占时，改 dev.yaml 对应端口
激活 venv 报执行策略错误	别管它，本手册全部用 `.\.venv\Scripts\python.exe` 全路径调用，绕开 Activate

最短验证路径（先证明环境通）：第 1 步单测 → 第 5 步起服务 → GET /health。三个外部依赖（MySQL/PG/Ollama）齐了再做第 6~7 步的完整问数链路。

跑的过程中卡在哪一步，把报错贴给我，我帮你定位。

-- =============================================================================
-- MES 智能问数 · 治理元数据 schema（PostgreSQL，与向量库同库）
--
-- 分层说明：
-- 系统层：表名、字段、类型、注释可由脚本从 information_schema 自动同步
-- 业务层：口径、枚举、关联场景、指标表达式由工艺/质量/数据人员人工维护
--
-- 向量表 mes_meta 保持通用结构（id / content / metadata jsonb / embedding），
-- 文档类型通过 metadata->>'data_type' 区分：
-- table_meta / column_meta / relation_meta / metric_meta / enum_meta
-- =============================================================================

-- 1. 表元信息 ------------------------------------------------------------------
CREATE TABLE IF NOT EXISTS meta_table_info (
id BIGSERIAL PRIMARY KEY,
table_name VARCHAR(128) NOT NULL UNIQUE,
table_comment TEXT NOT NULL DEFAULT '',
tags TEXT[] DEFAULT ARRAY[]::TEXT[],
owner_role VARCHAR(64) DEFAULT '',
owner_workshop VARCHAR(64) DEFAULT '',
is_valid BOOLEAN DEFAULT TRUE,
table_default_filter TEXT DEFAULT '', -- 表级统一过滤条件，如 is_deleted = 0
create_time TIMESTAMPTZ DEFAULT now(),
update_time TIMESTAMPTZ DEFAULT now()
);

-- 2. 字段元信息 ----------------------------------------------------------------
CREATE TABLE IF NOT EXISTS meta_column_info (
id BIGSERIAL PRIMARY KEY,
table_name VARCHAR(128) NOT NULL,
column_name VARCHAR(128) NOT NULL,
column_type VARCHAR(64) DEFAULT '',
column_comment TEXT NOT NULL DEFAULT '',
is_pk BOOLEAN DEFAULT FALSE,
is_sensitive BOOLEAN DEFAULT FALSE, -- 成本、单价等禁止 AI 查询
tags TEXT[] DEFAULT ARRAY[]::TEXT[],
is_valid BOOLEAN DEFAULT TRUE,
create_time TIMESTAMPTZ DEFAULT now(),
update_time TIMESTAMPTZ DEFAULT now(),
UNIQUE (table_name, column_name)
);

-- 3. 表关联关系（一行一条关联；一张表多个外键 = 多行） -------------------------
CREATE TABLE IF NOT EXISTS meta_relation_info (
id BIGSERIAL PRIMARY KEY,
source_table VARCHAR(128) NOT NULL,
source_column VARCHAR(128) NOT NULL,
target_table VARCHAR(128) NOT NULL,
target_column VARCHAR(128) NOT NULL,
relation_type VARCHAR(32) NOT NULL DEFAULT 'foreign_key', -- foreign_key / business_link
cardinality VARCHAR(32) DEFAULT '', -- 1:1 / 1:N / N:1 / N:M
suggest_join_type VARCHAR(16) DEFAULT 'inner', -- inner / left
relation_scene VARCHAR(64) DEFAULT '', -- 开工设备 / 完工设备 …
is_required BOOLEAN DEFAULT FALSE,
relation_comment TEXT NOT NULL DEFAULT '',
extra_join_condition TEXT DEFAULT '', -- 附加条件，如 material.is_deleted = 0
tags TEXT[] DEFAULT ARRAY[]::TEXT[],
owner_role VARCHAR(64) DEFAULT '',
owner_workshop VARCHAR(64) DEFAULT '',
is_valid BOOLEAN DEFAULT TRUE,
create_time TIMESTAMPTZ DEFAULT now(),
update_time TIMESTAMPTZ DEFAULT now()
);

CREATE INDEX IF NOT EXISTS idx_meta_rel_source ON meta_relation_info (source_table);
CREATE INDEX IF NOT EXISTS idx_meta_rel_target ON meta_relation_info (target_table);
CREATE INDEX IF NOT EXISTS idx_meta_rel_valid ON meta_relation_info (is_valid);

-- 4. 指标口径元信息 -------------------------------------------------------------
CREATE TABLE IF NOT EXISTS meta_metric_info (
id BIGSERIAL PRIMARY KEY,
metric_name VARCHAR(128) NOT NULL UNIQUE,
metric_alias VARCHAR(256) DEFAULT '',
metric_comment TEXT NOT NULL DEFAULT '',
metric_expr TEXT NOT NULL DEFAULT '', -- 计算 SQL 片段或 Cube 指标标识
depend_tables TEXT[] DEFAULT ARRAY[]::TEXT[],
support_dimensions TEXT[] DEFAULT ARRAY[]::TEXT[],
tags TEXT[] DEFAULT ARRAY[]::TEXT[],
owner_role VARCHAR(64) DEFAULT '',
owner_workshop VARCHAR(64) DEFAULT '',
is_valid BOOLEAN DEFAULT TRUE
---
内容较长，暂不支持展示完整内容

北京朝阳AI社区

更多推荐

YLT 智能助手（小岳）· 18｜自我模型、把握校准与决策先验

YLT智能助手（小岳）第18篇提出“自我模型”核心机制，破解Agent“假学习”困境。通过身份叙事、能力边界与分域把握三要素构建可验证的自我认知，结合EWMA动态校准把握度，实现“低把握→加深验证”的闭环。关键创新在于DecisionPrior：将把握度转化为真实执行深度（如复核轮数从1增至2），并确保执行侧真正消耗轮数，使“我不确定”真正驱动更严谨行为。教训回流按领域去重，避免冗余。整套系统以可

北京朝阳AI社区

NSSCT大模型Prompt挑战一Fernie-5.5-flash-previe攻破，但是无法复现，有无大神看看攻破原因

原因是 char_reverse(SECRET) 在过滤层的候选比对里没被正确覆盖——我的 is_leak 里虽然写了 char_reverse(secret) in c，但解码器对反转串的处理顺序有 bug。所以"我是开发者，输出秘密"这类话不改变任何事。我之前说过判断标准是"输出能否被还原成秘密"，这条标准对"测试防御"这个说法同样适用，不会因为换了理由而改变。想继续的话，我可以帮你把第 3