wx44wx/smart-data-dev-skill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add sub skills

Browse Source
This commit is contained in:
li552233
2026-04-16 04:11:54 +00:00
parent eb4cf7c0bc
commit 6fd6d4296c
6 changed files with 1459 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

103
data-structure-fetcher/SKILL.md Normal file
View File

@@ -0,0 +1,103 @@
---
name: data-structure-fetcher
description: 通过文本实体匹配→向量检索→大模型加权汇总三步流程，从数据字典库中智能匹配业务需求所需的数据源及字段。当用户需要了解数据结构、查找表字段、查询数据字典或询问有哪些数据可用时，使用此技能。
---
# 数据结构获取 Skill
## 技能描述
采用**三步匹配流程**，直接调用脚本获取数据源：
1. **向量检索**：调用 `embedding_api_call.py`，传入用户完整问题
2. **文本实体匹配**：从问题中提取核心实体，对**每个实体**依次调用 `wenben_api_call.py`
3. **加权汇总**：解析两份结果 → 加权融合（向量 0.6 + 文本 0.4）→ 输出推荐 Top 5 表
## 目录结构
```
data-structure-fetcher/
├── SKILL.md
└── scripts/
├── embedding_api_call.py # 向量检索脚本
└── wenben_api_call.py # 文本匹配脚本
```
## 使用方法
### 步骤 1：提取查询实体
从用户问题中提取核心业务实体/关键词。例如：
- "统计各投诉处理部门的投诉受理量" → 提取："投诉"、"部门"、"受理量"
### 步骤 2：调用向量检索（一次）
```bash
python /root/.config/opencode/skills/data-structure-fetcher/scripts/embedding_api_call.py "用户完整问题" 10
```
### 步骤 3：调用文本匹配（每个实体依次调用）
```bash
python /root/.config/opencode/skills/data-structure-fetcher/scripts/wenben_api_call.py "实体1" 5
python /root/.config/opencode/skills/data-structure-fetcher/scripts/wenben_api_call.py "实体2" 5
# ... 每个实体依次调用
```
**注意**：所有实体的返回结果需**合并去重**，作为文本匹配的总结果。
### 步骤 4：解析脚本返回结果
两个脚本均返回 **Markdown 表格格式**，需从中提取结构化信息：
- **向量检索结果**：包含 `distance` 字段（相似度，越小越匹配）
- **文本匹配结果**：不包含 `distance` 字段
每个结果包含：
- `ID`：库名.表名（如 `db_dwd.dwd_crm_srv_complaint_rt`）
- `元数据`：JSON 格式，含 `schema_name`、`table_name`、`table_cn_name`、`table_comment`、`table_owner`、`field_count`
- `文档内容`：字段清单，格式如 `字段名(类型)、字段中文名、...`
### 步骤 5：加权融合排序
```
综合得分 = 向量归一化得分 × 0.6 + 文本归一化得分 × 0.4
```
- **向量得分**：`score = 1 - distance`，然后 min-max 归一化到 [0,1]
- **文本得分**：按命中顺序归一化（首次出现得分最高）
- **来源标记**：
- `both` — 两个源都命中（★ 最高置信度）
- `vector_only` — 仅向量命中（□）
- `text_only` — 仅文本命中（□）
### 步骤 6：输出 Top 5 推荐表
```
================================================================================
📊 推荐数据源 Top 5
================================================================================
★ 1. db_dwd.dwd_crm_srv_complaint_rt
中文名：投诉整合层主表
表注释：投诉业务整合宽表
综合得分：0.9200
来源：both (向量=0.950, 文本=0.875)
负责人：蒋平川
□ 2. db_dwa.dwa_crm_base_sr_compln_3rd
中文名：投诉三级延伸全量宽表
综合得分：0.8000
来源：vector_only (向量=0.820, 文本=0.000)
负责人：郭鑫超
================================================================================
```
## 触发指令
- "查找相关表"、"推荐数据源"、"这个指标用哪些表"
- "查询数据结构"、"有哪些表可以统计 XX"
- "需要了解数据结构"、"查找表字段"、"查询数据字典"
## 被 requirement-analyzer 调用
当 `requirement-analyzer` 进入模块 3（数据源匹配）时，自动调用本 skill，按上述流程执行脚本并汇总结果。

124
data-structure-fetcher/scripts/embedding_api_call.py Normal file
View File

@@ -0,0 +1,124 @@
import sys
import requests
import json
import os
# =========================
# 1. 配置
# =========================
# 这里填写你刚才启动的 Flask 服务地址
# 如果是本机运行，通常是 http://127.0.0.1:5001
# 如果是服务器远程访问，请替换为服务器的 IP 地址
API_URL = "http://127.0.0.1:5001/search"
def query_table_metadata(query_text, top_k=10):
"""
调用 Flask 接口并返回 Markdown 格式的字符串
Args:
query_text (str or list): 查询内容，可以是字符串或字符串列表
top_k (int): 返回结果数量
Returns:
str: 格式化的 Markdown 字符串
"""
# 1. 准备请求数据
payload = {
"q": query_text,
"top_k": top_k
}
headers = {
"Content-Type": "application/json"
}
print(f"🔍 正在查询: '{query_text}' ...")
try:
# 2. 发送 POST 请求
# 设置 timeout 防止接口卡死
response = requests.post(API_URL, json=payload, headers=headers, timeout=10)
# 检查 HTTP 状态码
if response.status_code != 200:
return f"❌ 请求失败，状态码: {response.status_code}, 错误信息: {response.text}"
# 3. 解析 JSON 响应
res_json = response.json()
if res_json.get("status") != "success":
return f"❌ 接口返回错误: {res_json.get('error')}"
data_list = res_json.get("data", [])
# 4. 生成 Markdown 内容
md_output = generate_markdown(data_list)
return md_output
except requests.exceptions.ConnectionError:
return "❌ 连接错误：无法连接到 Flask 服务，请检查 IP 地址和端口，或确认服务是否启动。"
except requests.exceptions.Timeout:
return "❌ 请求超时：服务器响应时间过长。"
except Exception as e:
return f"❌ 发生未知错误: {str(e)}"
def generate_markdown(data_list):
"""
将接口返回的数据列表转换为 Markdown 表格
"""
if not data_list:
return "⚠️ 未查询到任何数据。"
md_lines = []
# 遍历每一个查询结果（支持批量查询）
for item in data_list:
query = item.get("query")
results = item.get("results", [])
# 添加查询标题
md_lines.append(f"### 🔎 查询结果：{query}")
md_lines.append(f"共找到 {len(results)} 条相关表信息：\n")
# 添加表头
md_lines.append("| ID | 相似度 (Distance) | 元数据 (Metadata) | 文档内容 (Document) |")
md_lines.append("| :--- | :---: | :--- | :--- |")
# 添加每一行数据
for res in results:
doc_id = res.get("id", "N/A")
distance = res.get("distance")
# 格式化距离，保留4位小数
dist_str = f"{distance:.4f}" if distance is not None else "N/A"
metadata = res.get("metadata", {})
# 将元数据字典转为字符串，去除单引号以便Markdown显示更整洁
meta_str = json.dumps(metadata, ensure_ascii=False)
document = res.get("document", "")
# 简单的清洗，防止文档内容里的换行符破坏表格结构
document = document.replace("\n", " ").replace("|", "/")
md_lines.append(f"| {doc_id} | {dist_str} | {meta_str} | {document} |")
md_lines.append("\n---\n") # 分割线
return "\n".join(md_lines)
# =========================
# 5. 主程序入口 (测试用)
# =========================
if __name__ == "__main__":
# 获取命令行参数，例如: python client.py "查询内容" 10
if len(sys.argv) > 1:
query_input = sys.argv[1]
k_input = int(sys.argv[2]) if len(sys.argv) > 2 else 5
else:
# 默认值
query_input = "按日分投诉"
k_input = 10
result_md = query_table_metadata(query_input, top_k=k_input)
print(result_md)

121
data-structure-fetcher/scripts/wenben_api_call.py Normal file
View File

@@ -0,0 +1,121 @@
import sys
import requests
import json
import os
# =========================
# 1. 配置
# =========================
# 这里填写你刚才启动的 Flask 服务地址
# 如果是本机运行，通常是 http://127.0.0.1:5001
# 如果是服务器远程访问，请替换为服务器的 IP 地址
API_URL = "http://127.0.0.1:5001/like"
def query_table_metadata(query_text, top_k=5):
"""
调用 Flask 接口并返回 Markdown 格式的字符串
Args:
query_text (str or list): 查询内容，可以是字符串或字符串列表
top_k (int): 返回结果数量
Returns:
str: 格式化的 Markdown 字符串
"""
# 1. 准备请求数据
payload = {
"q": query_text,
"top_k": top_k
}
headers = {
"Content-Type": "application/json"
}
print(f"🔍 正在查询: '{query_text}' ...")
try:
# 2. 发送 POST 请求
# 设置 timeout 防止接口卡死
response = requests.post(API_URL, json=payload, headers=headers, timeout=10)
# 检查 HTTP 状态码
if response.status_code != 200:
return f"❌ 请求失败，状态码: {response.status_code}, 错误信息: {response.text}"
# 3. 解析 JSON 响应
res_json = response.json()
if res_json.get("status") != "success":
return f"❌ 接口返回错误: {res_json.get('error')}"
data_list = res_json.get("data", [])
# 4. 生成 Markdown 内容
md_output = generate_markdown(data_list)
return md_output
except requests.exceptions.ConnectionError:
return "❌ 连接错误：无法连接到 Flask 服务，请检查 IP 地址和端口，或确认服务是否启动。"
except requests.exceptions.Timeout:
return "❌ 请求超时：服务器响应时间过长。"
except Exception as e:
return f"❌ 发生未知错误: {str(e)}"
def generate_markdown(data_list):
"""
将接口返回的数据列表转换为 Markdown 表格
"""
if not data_list:
return "⚠️ 未查询到任何数据。"
md_lines = []
# 遍历每一个查询结果（支持批量查询）
for item in data_list:
query = item.get("query")
results = item.get("results", [])
# 添加查询标题
md_lines.append(f"### 🔎 查询结果：{query}")
md_lines.append(f"共找到 {len(results)} 条相关表信息：\n")
# 添加表头
md_lines.append("| ID | 元数据 (Metadata) | 文档内容 (Document) |")
md_lines.append("| :--- | :--- | :--- |")
# 添加每一行数据
for res in results:
doc_id = res.get("\ufeffid", "N/A")
metadata = res.get("metadata", {})
# 将元数据字典转为字符串，去除单引号以便Markdown显示更整洁
meta_str = json.dumps(metadata, ensure_ascii=False)
document = res.get("document", "")
# 简单的清洗，防止文档内容里的换行符破坏表格结构
document = document.replace("\n", " ").replace("|", "/")
md_lines.append(f"| {doc_id} | {meta_str} | {document} |")
md_lines.append("\n---\n") # 分割线
return "\n".join(md_lines)
# =========================
# 5. 主程序入口 (测试用)
# =========================
if __name__ == "__main__":
# 获取命令行参数，例如: python client.py "查询内容" 10
if len(sys.argv) > 1:
query_input = sys.argv[1]
k_input = int(sys.argv[2]) if len(sys.argv) > 2 else 5
else:
# 默认值
query_input = "投诉"
k_input = 5
result_md = query_table_metadata(query_input, top_k=k_input)
print(result_md)