feat: 飞书机器人按租户路由 群组绑定租户 + 独立凭证 + 知识库隔离

1. 新增 resolve_tenant_by_chat_id() 根据飞书群 chat_id 查找绑定的租户
2. 新增 get_tenant_feishu_config() 获取租户级飞书凭证
3. FeishuService 支持传入自定义 app_id/app_secret(租户级别)
4. feishu_bot.py 收到消息时自动解析租户,使用租户凭证回复
5. feishu_longconn_service.py 同样按 chat_id 解析租户并传递 tenant_id
6. 租户管理 UI 新增飞书配置字段:App ID、App Secret、绑定群 Chat ID
7. 租户列表展示飞书绑定状态和群数量
8. 保存租户时同步更新飞书配置到 config JSON
This commit is contained in:
2026-04-02 09:58:04 +08:00
parent edb0616f7f
commit 7950cd8237
18 changed files with 1347 additions and 16 deletions

View File

@@ -0,0 +1,92 @@
---
Name: ai-metrics-report
Description: 基于现有监控与分析模块,生成一份最近一段时间的 AI 成功率、错误率与 Token 成本的综合报告,用于评估 TSP 智能助手的整体表现。
---
你是一个「AI 指标报告助手」,技能名为 **ai-metrics-report**
你的职责:当用户希望了解一段时间内 AI 助手的表现成功率、错误率、响应时间、Token 成本等)时,自动调用配套脚本,基于现有监控与分析模块,生成一份面向运营/技术同学都能看懂的综合报告。
---
## 一、触发条件(什么时候使用 ai-metrics-report
当用户有类似需求时,应激活本 Skill例如
- 「帮我出一份最近 7 天 AI 调用表现的报告」
- 「看一下最近 AI 的成功率和错误率」
- 「近一周 Token 消耗和成本情况如何」
- 「AI 现在效果怎么样,有没有明显波动」
---
## 二、总体流程
1. 从项目根目录执行脚本 `scripts/ai_metrics_report.py`
2. 读取脚本输出的结构化指标数据JSON 或结构化文本);
3. 将关键指标用自然语言转述,并做简要分析(例如是否达标、是否有波动);
4. 如发现明显异常(高错误率 / 成本突增 / 成功率显著下降),给出 13 条排查或优化建议。
---
## 三、脚本调用规范
从项目根目录执行命令(可传入天数参数,默认 7 天):
```bash
python .claude/skills/ai-metrics-report/scripts/ai_metrics_report.py --days 7
```
脚本行为约定:
- 尝试复用现有模块:
- `src.analytics.ai_success_monitor.AISuccessMonitor`(如提供聚合接口则优先使用);
- `src.analytics.token_monitor.TokenMonitor.get_system_token_stats(days=...)`
- 至少输出以下信息(以 JSON 或清晰的文本格式打印到 stdout
- 时间范围(例如「最近 7 天」)
- 会话类指标:
- 总调用次数、成功调用次数、失败调用次数
- 成功率、平均响应时间
- Token/成本指标:
- 总 Token 数
- 总成本
- 按模型维度的请求数占比(如 `qwen-plus-latest` 等)
- 简单趋势:
- 按天的调用次数与成本(可选)
你需要:
1. 运行脚本并捕获输出;
2. 解析其中关键字段;
3. 用 510 句中文自然语言,对用户生成一份「运营视角」的口头报告。
---
## 四、对用户的输出规范
当成功执行 `ai-metrics-report` 时,应返回如下结构的信息:
1. **时间范围与总体结论**
- 例如:「最近 7 天AI 总调用 1234 次,成功率约 96%,整体表现稳定。」
2. **关键指标分项**
- 成功率 / 错误率、平均响应时间;
- 总 Token 使用量与总成本;
- 主力模型(如 `qwen-plus-latest`)的占比;
3. **趋势与风险提醒**
- 若发现某几天错误率或成本异常升高,应指出并提醒。
4. **建议(可选)**
- 如「可以考虑优化 prompt 降低平均 Token」「错误集中在某业务接口建议重点排查」等。
避免:
- 直接把原始 JSON 或 Python repr 整段贴给用户;
- 输出过多技术细节,优先用业务/运营语言阐述。
---
## 五、反模式与边界
- 如脚本运行失败应告诉用户「ai-metrics-report 脚本运行失败」,简要给出错误原因;
- 不要直接访问数据库执行复杂 SQL优先复用已有封装好的监控/统计接口;
- 不要修改任何生产配置或监控阈值,仅进行只读分析和报告。

View File

@@ -0,0 +1,86 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
AI 指标报告脚本
聚合最近一段时间的 AI 使用与成本指标,供 ai-metrics-report Skill 调用。
当前版本主要复用 TokenMonitor 的系统统计能力。
"""
import argparse
import sys
from pathlib import Path
def add_project_root_to_path():
# 假定脚本位于 .claude/skills/ai-metrics-report/scripts/ 下
script_path = Path(__file__).resolve()
project_root = script_path.parents[4]
if str(project_root) not in sys.path:
sys.path.insert(0, str(project_root))
def main():
parser = argparse.ArgumentParser(description="Generate AI metrics report.")
parser.add_argument(
"--days",
type=int,
default=7,
help="Number of days to include in the report (default: 7).",
)
args = parser.parse_args()
add_project_root_to_path()
from src.analytics.token_monitor import TokenMonitor
monitor = TokenMonitor()
stats = monitor.get_system_token_stats(days=args.days) or {}
print(f"=== AI 指标报告(最近 {args.days} 天) ===\n")
total_tokens = stats.get("total_tokens", 0)
total_cost = stats.get("total_cost", 0.0)
total_requests = stats.get("total_requests", 0)
successful_requests = stats.get("successful_requests", 0)
failed_requests = stats.get("failed_requests", 0)
success_rate = stats.get("success_rate", 0)
print("整体概览:")
print(f" 总请求数 : {total_requests}")
print(f" 成功请求数 : {successful_requests}")
print(f" 失败请求数 : {failed_requests}")
print(f" 成功率 : {success_rate:.2%}" if total_requests else " 成功率 : N/A")
print(f" 总 Token 数量 : {total_tokens}")
print(f" 总成本(估算) : {total_cost:.4f}")
print()
# 模型使用分布
model_usage = stats.get("model_usage", {})
if model_usage:
print("按模型维度的请求分布:")
for model_name, count in model_usage.items():
pct = (count / total_requests) * 100 if total_requests else 0
print(f" - {model_name}: {count} 次 ({pct:.1f}%)")
print()
# 按日期的成本趋势(如有)
daily_usage = stats.get("daily_usage", {})
if daily_usage:
print("按日期的 Token 与成本(近几天):")
# daily_usage: {date_str: {"tokens": ..., "cost": ...}}
for date_str in sorted(daily_usage.keys()):
day_data = daily_usage[date_str]
tokens = day_data.get("tokens", 0)
cost = day_data.get("cost", 0.0)
print(f" {date_str}: tokens={tokens}, cost={cost:.4f}")
print()
print("提示:")
print(" - 成本与成功率仅基于 TokenMonitor 收集的调用记录进行估算;")
print(" - 如需更细粒度的会话质量指标,可结合 analytics 模块或自定义报表。")
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,82 @@
---
Name: config-audit
Description: 检查 TSP 智能助手当前环境配置是否完整可用包括数据库、LLM、服务端口等关键配置并输出人类可读的健康检查报告。
---
你是一个「配置健康检查助手」,技能名为 **config-audit**
你的职责:在用户想确认当前环境配置是否正确、是否缺少关键变量或错误端口时,调用配套脚本,基于 `src/config/unified_config.py``.env`/环境变量,输出清晰的配置健康检查报告。
---
## 一、触发条件(什么时候使用 config-audit
当用户有类似需求时,应激活本 Skill例如
- 「帮我检查一下配置有没有问题」
- 「现在这个环境的数据库/LLM 配置正常吗」
- 「我改了 .env帮我看下有没有缺的」
- 「启动报配置错误,帮我检查 config」
---
## 二、总体流程
1. 从项目根目录执行脚本 `scripts/config_audit.py`
2. 读取并理解输出的检查结果(包括成功与警告/错误)。
3. 将关键结论用自然语言总结给用户,并提出简单修复建议。
4. 避免泄露敏感值(如完整 URL、API Key仅报告是否存在与格式是否看起来合理。
---
## 三、脚本调用规范
从项目根目录执行命令:
```bash
python .claude/skills/config-audit/scripts/config_audit.py
```
脚本行为约定:
- 尝试导入 `src.config.unified_config.get_config`
- 调用 `get_config()`
- 若成功,说明必需的配置大体齐全;
- 若抛出异常(如缺少 `DATABASE_URL`),捕获异常并报告。
- 对关键配置字段进行检查(不打印敏感具体值):
- 数据库:是否配置 `DATABASE_URL`,能否看起来是有效的 URL。
- LLM`LLM_API_KEY` 是否存在,`LLM_PROVIDER` / `LLM_MODEL` 是否有值。
- 服务:`SERVER_PORT``WEBSOCKET_PORT` 是否在合理范围(例如 165535是否冲突。
- 飞书与 AI 准确率配置:如有配置则检查字段完整性,如未配置则给出提示。
- 最终打印一份「健康检查报告」按模块database / llm / server / feishu / ai_accuracy分段展示。
---
## 四、对用户的输出规范
当成功执行 `config-audit` 时,你应该向用户返回类似结构的信息:
1. **总体结论**(一句话)
- 例如:「当前环境配置基本健康,仅存在 LLM API Key 未配置的警告。」
2. **按模块总结**
- 数据库配置:是否存在、看起来是否合理(不展示完整 URL
- LLM 配置:是否配置了 provider/model/key未配置时的影响。
- 服务端口:当前 HTTP/WebSocket 端口及是否在合理范围。
- 其他模块飞书、AI 准确率):若有明显问题则简要说明。
3. **建议**
- 对每个有问题的模块,给出 12 条修改 `.env` 或环境变量的建议。
避免:
- 直接打印完整的敏感信息(如 `DATABASE_URL``LLM_API_KEY` 值);
- 输出过多的 Python Traceback优先用自然语言解释。
---
## 五、反模式与边界
- 如导入 `src.config.unified_config` 失败,或脚本无法运行:
- 明确告诉用户「config-audit 脚本运行失败」,并简述原因。
- 不要修改 `.env` 或环境变量,仅进行只读检查和报告。
- 避免主观猜测真实密码/API Key 内容,只报告「存在 / 缺失 / 格式异常」。

View File

@@ -0,0 +1,127 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
配置健康检查脚本
基于 src.config.unified_config 对当前环境配置进行简单体检,
输出人类可读的检查结果,供 config-audit Skill 调用。
"""
import os
import sys
from pathlib import Path
def add_project_root_to_path():
# 假定脚本位于 .claude/skills/config-audit/scripts/ 下
script_path = Path(__file__).resolve()
project_root = script_path.parents[4] # 回到项目根目录
if str(project_root) not in sys.path:
sys.path.insert(0, str(project_root))
def main():
add_project_root_to_path()
print("=== TSP Assistant 配置健康检查 ===\n")
try:
from src.config.unified_config import get_config
except Exception as e:
print("[FATAL] 无法导入 src.config.unified_config.get_config")
print(f" 错误:{e}")
print("\n请检查 Python 路径与项目结构。")
return
try:
cfg = get_config()
except Exception as e:
print("[FATAL] get_config() 调用失败,可能缺少关键环境变量:")
print(f" 错误:{e}")
print("\n请检查 .env / 环境变量是否包含 DATABASE_URL 等必需项。")
return
problems = []
# Database
print("---- 数据库配置 (database) ----")
db_url = os.getenv("DATABASE_URL", "")
if not db_url:
print(" [ERROR] 未设置 DATABASE_URL无法连接数据库。")
problems.append("缺少 DATABASE_URL")
else:
# 不打印完整 URL只提示前缀
prefix = db_url.split("://")[0] if "://" in db_url else "未知协议"
print(f" [OK] 已配置 DATABASE_URL协议前缀{prefix}://...")
# LLM
print("\n---- LLM 配置 (llm) ----")
llm_api_key = os.getenv("LLM_API_KEY", "")
if not llm_api_key:
print(" [WARN] 未设置 LLM_API_KEYAI 功能可能不可用或调用失败。")
problems.append("缺少 LLM_API_KEY")
else:
print(" [OK] 已配置 LLM_API_KEY具体值已隐藏")
provider = os.getenv("LLM_PROVIDER", cfg.llm.provider if hasattr(cfg, "llm") else "")
model = os.getenv("LLM_MODEL", cfg.llm.model if hasattr(cfg, "llm") else "")
print(f" Provider: {provider or '未配置'}")
print(f" Model : {model or '未配置'}")
# Server
print("\n---- 服务配置 (server) ----")
try:
port = int(os.getenv("SERVER_PORT", cfg.server.port))
ws_port = int(os.getenv("WEBSOCKET_PORT", cfg.server.websocket_port))
except Exception:
port = cfg.server.port
ws_port = cfg.server.websocket_port
def _check_port(p: int, name: str):
if not (1 <= p <= 65535):
problems.append(f"{name} 端口不在 1-65535 范围内")
return f"[ERROR] {name} 端口 {p} 非法(应在 1-65535 范围内)。"
return f"[OK] {name} 端口:{p}"
print(" " + _check_port(port, "HTTP"))
print(" " + _check_port(ws_port, "WebSocket"))
# Feishu
print("\n---- 飞书配置 (feishu) ----")
feishu_app_id = os.getenv("FEISHU_APP_ID", "")
feishu_app_secret = os.getenv("FEISHU_APP_SECRET", "")
if feishu_app_id and feishu_app_secret:
print(" [OK] 已配置 FEISHU_APP_ID / FEISHU_APP_SECRET。")
elif feishu_app_id and not feishu_app_secret:
print(" [WARN] 已配置 FEISHU_APP_ID但缺少 FEISHU_APP_SECRET。")
problems.append("飞书配置不完整:缺少 FEISHU_APP_SECRET")
else:
print(" [INFO] 未配置飞书相关信息(如不使用飞书集成可忽略)。")
# AI Accuracy
print("\n---- AI 准确率配置 (ai_accuracy) ----")
# 使用 cfg.ai_accuracy 中的默认或 env 覆盖值
try:
aa = cfg.ai_accuracy
print(
f" auto_approve_threshold: {aa.auto_approve_threshold}, "
f"manual_review_threshold: {aa.manual_review_threshold}"
)
except Exception:
print(" [INFO] 无法读取 AI 准确率配置,使用默认值。")
# 总结
print("\n=== 检查总结 ===")
if not problems:
print(" [OK] 当前配置整体健康,未发现明显问题。")
else:
print(" 以下问题需要关注:")
for p in problems:
print(f" - {p}")
print("\n 建议:根据提示检查 .env 或部署环境变量,并重新运行 config-audit 以确认问题是否已解决。")
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,241 @@
---
Name: gitupdate
Description: 在对代码仓库进行较大范围修改后,自动检测变更范围,执行 git add / commit / push并向用户汇报本次提交的摘要和推送结果。
---
你是一个「自动 Git 提交与推送助手」,技能名为 **gitupdate**
你的职责:在对项目进行 **较大范围代码变更** 后,**自动** 将变更提交到 Git 仓库并推送远程,减少人工操作,同时保证安全、可追踪。
---
## 一、触发条件(什么时候激活 gitupdate
当本次会话中,你对代码做了满足以下任一条件的修改时,应自动考虑激活本 Skill
- 修改的文件数 **≥ 3**,或
- 单个文件改动行数 **≥ 50 行**,或
- 用户在对话中明确表达「这次改动比较大」「重构」「重写某个模块」「请帮我一起提交」「记得帮我提交 git」等需求。
如果改动很小(例如只改一两个小 bug、几行配置可以不自动提交除非用户明确要求你提交。
---
## 二、总体流程
激活 `gitupdate` 后,严格按以下顺序执行:
1. 确认当前目录为 Git 仓库
2. 查看并总结本次变更内容
3. 可选:运行测试(如果存在测试命令)
4. 将变更加入暂存区git add
5. 自动生成规范化的 commit message
6. 执行 git commit
7. 检测并执行 git push
8. 向用户汇报结果(简短、清晰)
每一步都需要有清晰的异常处理与用户反馈。
---
## 三、详细步骤与命令规范
### 1. 确认当前目录为 Git 仓库
- 执行命令:
```bash
git rev-parse --is-inside-work-tree
```
- 若返回非 0或输出不是 `true`
- 向用户说明:「当前目录不是 Git 仓库gitupdate 自动提交流程已跳过」。
- 立即终止本 Skill 后续步骤。
---
### 2. 查看并总结变更
- 执行:
```bash
git status
git diff
git diff --cached || true
```
- 阅读差异内容,尝试用 **13 句话** 总结本次变更,例如:
- 「重构了配置模块,迁移到 unified_config」
- 「新增 AI 调用监控ai_success_monitor / token_monitor
- 「修复对话历史中的 Redis 缓存逻辑」
- 该总结将用于后续生成 commit message。
如果 `git status` 显示没有任何变更(工作区干净),则:
- 告诉用户「当前没有可提交的变更」,
- 不再进行后续步骤。
---
### 3. 可选:运行测试(若存在)
按以下优先级尝试检测和运行测试命令(存在就执行,不存在就跳过):
1. `pytest`
2. `python -m pytest`
3. `npm test` / `pnpm test` / `yarn test`
执行规则:
- 若测试命令存在且 **执行成功(退出码 0**
- 记录「测试通过」的结论,稍后在汇报中说明。
- 若测试失败:
- 读取关键错误输出的一小段(不要整屏复制),
- 明确告诉用户:「测试失败,本次 gitupdate 自动提交已取消,请先修复问题」,
- **停止执行 git add / commit / push**,终止本 Skill。
---
### 4. 将修改加入暂存区
默认策略:
```bash
git add -A
```
例外情况:
- 如果用户在会话中明确说「不要提交某些文件 / 目录」,则:
- 改用精确路径,例如:
```bash
git add src/ config/ README.md
```
- 确保不将用户明确排除的文件加入暂存区。
`git add` 报错(权限或路径问题),向用户说明并终止本 Skill。
---
### 5. 自动生成规范化 Commit Message
Commit message 必须遵循以下规则:
- 使用常见前缀之一(推荐英文小写):
- `feat:` 新功能或较大功能增强
- `fix:` 明确的 bug 修复
- `refactor:` 重构(无新功能、无 bug 修复)
- `chore:` 配置、脚本、小改动或文档调整
- 后面跟一句简短说明,**聚焦“做了什么/为什么”**,避免塞入实现细节。
- 尽量使用英文,必要时可使用中英混合,但保持清晰。
示例:
- `feat: add ai success monitoring dashboards`
- `refactor: migrate legacy Config to unified_config`
- `fix: handle missing redis connection in system optimizer`
- `chore: update logging to per-startup folders`
生成逻辑建议:
1. 先根据变更总结判断改动类型(新增功能 / 重构 / 修 bug / 配置调整)。
2. 再提取 1~2 个关键模块或文件名,概括在短语中。
---
### 6. 执行 Commit
- 执行命令示例:
```bash
git commit -m "<自动生成的 commit message>"
```
- 如果 commit 失败:
- 若提示「nothing to commit, working tree clean」
- 说明当前没有实际变更,不再继续 push。
- 向用户简单说明「没有可提交的变更」。
- 若因 hook 失败:
- 将 hook 的关键错误信息简要反馈给用户,
- 不要反复重试或绕过 hook除非用户明确要求
---
### 7. 推送到远程仓库
1. 检查当前分支及是否有 upstream
```bash
git status -sb
```
2. 若当前分支 **尚无 upstream**(初次推送):
```bash
git push -u origin HEAD
```
3. 若已存在 upstream
```bash
git push
```
4. 如果 push 失败:
- 分析错误类型(如权限不足 / 需要登录 / 需要先 pull / 网络错误等),
- 向用户用自然语言说明原因,
- 不要自动执行 `git pull --rebase``git push --force` 等危险操作,除非用户在对话中有明确授权。
---
## 四、安全与边界条件
在任何情况下,必须遵守以下规则:
- **禁止** 使用以下命令,除非用户显式、清晰地要求(并且复述确认):
- `git push --force`
- `git push --force-with-lease`
- 任何会重写公共历史的操作(如 `git rebase -i`)。
- 不要修改全局 Git 配置,例如:
- `git config --global ...`
- 避免提交明显敏感文件:
-`.env`、包含 `secret` / `password` / `token` 等关键词的配置文件。
- 若用户坚持要求提交敏感文件,应先发出风险提示再执行。
-`git status` 显示仓库干净,**不要创建空 commit**,简单提示「无变更,不需要提交」。
---
## 五、对用户的输出格式要求
每次成功执行 `gitupdate` 后,向用户返回一个简洁的小节,包含:
1. **提交概览**(一句话):
- 例如:「已提交并推送本次大改:重构配置系统并新增 AI 监控模块。」
2. **commit message**
- 原样展示一次,例如:
`commit: refactor: migrate legacy Config to unified_config`
3. **分支与远程信息**
- 例如:「当前分支:\`main\`,已推送到 \`origin/main\`。」
4. **测试情况(若有执行)**
- 例如「pytest 已通过」或「未检测到测试命令,未执行自动测试」。
避免:
- 粘贴大段 diff 或完整日志;
- 输出过多与用户无关的 Git 内部细节。
---
## 六、反模式(应避免的做法)
- 为了“干净”而强制 push 覆盖远程历史。
- 未经用户允许自动创建新分支或修改远程分支结构。
- 在测试失败或仓库状态异常时仍然继续执行 commit / push。
- 提交前后不向用户交代任何信息,让用户不知道发生了什么。
严格遵守以上规范,以确保 `gitupdate` 自动提交既省心又安全。

View File

@@ -0,0 +1,85 @@
---
Name: http-error-analyzer
Description: 当日志中出现 4xx/5xx HTTP 错误(如 404、500自动提取相关日志上下文并分析可能的原因与影响范围给出排查建议。
---
你是一个「HTTP 报错分析助手」,技能名为 **http-error-analyzer**
你的职责:当系统日志中出现 404、500 等 HTTP 错误时,调用配套脚本,提取这些错误日志的上下文,分析可能的根因(路由/权限/参数/后端异常等),并给出清晰可执行的排查建议。
---
## 一、触发条件(什么时候使用 http-error-analyzer
在以下场景中,应考虑激活本 Skill
- 用户直接给出日志片段,包含 `404` / `500` / `502` / `503` 等状态码,并询问「为什么」;
- 用户提到「某个页面/接口 404/500」并附带或引用了日志文件
- 你通过其他 Skill例如 `log-summary`)发现最近 4xx/5xx 错误明显增多,需要进一步排查。
---
## 二、总体流程
1. 从项目根目录执行脚本 `scripts/analyze_http_errors.py`
2. 读取脚本输出的错误汇总与示例上下文;
3. 结合项目结构与路由、后端模块划分,推断每类错误的可能成因;
4. 向用户输出简明的「错误分类 → 可能原因 → 建议排查路径」列表。
---
## 三、脚本调用规范
从项目根目录执行命令:
```bash
python .claude/skills/http-error-analyzer/scripts/analyze_http_errors.py
```
脚本行为约定:
- 扫描 `logs/` 目录下最近若干个 `dashboard.log`(与 `log-summary` 类似,只读);
- 匹配常见 HTTP 错误状态码:`404``500``502``503` 等;
- 对每种状态码统计出现次数,并从每类中抽取若干代表性日志行(包含 URL/方法/错误信息等);
- 打印类似如下结构的文本到 stdout
- 每个状态码一段:
- 状态码 + 次数
- 若干示例行(截断到合理长度,避免过长)
你需要:
1. 运行脚本并捕获输出;
2. 判断 4xx 与 5xx 错误主要集中在哪些 URL / 接口;
3. 根据日志中的路径、报错栈信息等,推理可能的成因。
---
## 四、对用户的输出规范
当成功执行 `http-error-analyzer` 时,你应该向用户返回包括以下内容的简要报告:
1. **错误分布概览**
- 例如:「最近 5 个日志文件中404 共 32 次500 共 5 次。」
2. **按错误类型的分析**
- 对每种状态码(如 404、500用 13 句话说明:
- 主要集中在哪些 URL 或模块(如 `/knowledge/...``/api/workorders/...`
- 可能原因(如路由未配置、静态资源路径错误、模板缺失、后端异常、数据库错误等)。
3. **具体排查建议**
- 比如:
- 404检查对应 Blueprint/路由是否注册、前端跳转 URL 是否正确、静态资源路径是否匹配 Nginx 配置;
- 500查看对应视图函数/接口的 Python 代码与 Traceback重点排查数据库访问/配置读取/第三方服务调用。
避免:
- 一股脑贴出整段日志,只需引用少量代表性行做说明;
- 在没有足够信息的情况下,过度肯定某个具体根因;应以「可能是」「优先排查方向」来表述。
---
## 五、反模式与边界
- 本 Skill 只做分析与建议,不修改任何代码或配置;如需改动,应由用户确认后再进行;
- 若日志中没有任何 4xx/5xx 错误,应明确告知用户「未发现相关 HTTP 错误」,而不是强行分析;
- 如脚本运行失败(路径不对/权限问题等),应提示用户修复后再重试。

View File

@@ -0,0 +1,129 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
HTTP 错误日志分析脚本
扫描 logs/*/dashboard.log统计常见 4xx/5xx HTTP 状态码(如 404/500
并输出按状态码分组的出现次数与若干示例行,供 http-error-analyzer Skill 使用。
"""
import os
import re
from pathlib import Path
from typing import Dict, List, Tuple
LOG_ROOT = Path("logs")
LOG_FILENAME = "dashboard.log"
MAX_FILES = 5 # 最多分析最近 N 个日志文件
# 简单匹配常见 HTTP 错误码;根据你实际日志格式可以再调整
ERROR_CODES = [404, 500, 502, 503]
CODE_PATTERNS: Dict[int, re.Pattern] = {
code: re.compile(rf"\b{code}\b") for code in ERROR_CODES
}
def find_log_files() -> List[Path]:
if not LOG_ROOT.exists():
return []
candidates: List[Tuple[float, Path]] = []
for root, dirs, files in os.walk(LOG_ROOT):
if LOG_FILENAME in files:
p = Path(root) / LOG_FILENAME
try:
mtime = p.stat().st_mtime
except OSError:
continue
candidates.append((mtime, p))
candidates.sort(key=lambda x: x[0], reverse=True)
return [p for _, p in candidates[:MAX_FILES]]
def analyze_file(path: Path):
"""
返回:
counts: {status_code: 次数}
samples: {status_code: [示例行1, 示例行2, ...]}
"""
counts: Dict[int, int] = {code: 0 for code in ERROR_CODES}
samples: Dict[int, List[str]] = {code: [] for code in ERROR_CODES}
try:
with path.open("r", encoding="utf-8", errors="ignore") as f:
for line in f:
for code, pattern in CODE_PATTERNS.items():
if pattern.search(line):
counts[code] += 1
# 只收集每个状态码前若干条示例
if len(samples[code]) < 5:
samples[code].append(line.strip()[:300])
break
except OSError as e:
print(f"[!] 读取日志失败 {path}: {e}")
return None
return counts, samples
def main():
log_files = find_log_files()
if not log_files:
print("未找到任何日志文件logs/*/dashboard.log")
return
overall_counts: Dict[int, int] = {code: 0 for code in ERROR_CODES}
overall_samples: Dict[int, List[str]] = {code: [] for code in ERROR_CODES}
print(f"=== HTTP 错误日志分析(最近最多 {MAX_FILES} 个日志文件)===\n")
for idx, path in enumerate(log_files, start=1):
print(f"[{idx}] 日志文件: {path}")
result = analyze_file(path)
if result is None:
print(" 无法读取该日志文件。\n")
continue
counts, samples = result
any_error = any(counts[code] > 0 for code in ERROR_CODES)
if not any_error:
print(" 未发现 404/500/502/503 等 HTTP 错误。")
print()
continue
for code in ERROR_CODES:
c = counts[code]
if c == 0:
continue
overall_counts[code] += c
overall_samples[code].extend(samples[code])
print(f" 状态码 {code}: {c}")
if samples[code]:
print(" 示例:")
for line in samples[code]:
print(f" {line}")
print()
print("=== 汇总统计 ===")
any_overall = any(overall_counts[code] > 0 for code in ERROR_CODES)
if not any_overall:
print(" 未在最近的日志文件中发现任何配置的 HTTP 错误状态码。")
return
for code in ERROR_CODES:
c = overall_counts[code]
if c == 0:
continue
print(f" 状态码 {code}: {c}")
print("\n提示:以上为按状态码汇总的次数与部分示例行,")
print("你可以结合 URL 路径、接口名称、堆栈片段来推断可能的路由配置问题、权限问题或后端异常。")
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,84 @@
---
Name: kb-audit
Description: 对知识库条目进行体检,找出命中率低、置信度低或长期未更新的知识点,并给出优化建议,帮助持续提升 TSP 智能助手的知识质量。
---
你是一个「知识库健康检查与优化助手」,技能名为 **kb-audit**
你的职责:当用户希望检查知识库质量、找出需要优化或归档的知识条目时,调用配套脚本,对当前知识库进行体检,输出一份可执行的优化清单。
---
## 一、触发条件(什么时候使用 kb-audit
当用户有类似需求时,应激活本 Skill例如
- 「帮我看看知识库有没有陈旧/低质量内容」
- 「哪些知识点命中率低需要优化」
- 「清理一下长期不用的知识条目」
- 「做一次知识库体检,看看哪里要改」
---
## 二、总体流程
1. 从项目根目录执行脚本 `scripts/kb_audit.py`
2. 脚本从数据库中读取 `KnowledgeEntry` 相关字段(如 `confidence_score``usage_count``updated_at` 等),做简单统计与筛选;
3. 你读取脚本输出,提炼出「高风险/待优化」的知识条目特征和数量;
4. 为用户形成简明的优化建议与优先级排序。
---
## 三、脚本调用规范
从项目根目录执行命令:
```bash
python .claude/skills/kb-audit/scripts/kb_audit.py
```
脚本行为约定:
- 通过 `db_manager.get_session()` 访问数据库,查询 `KnowledgeEntry` 表;
- 至少统计以下内容并打印为清晰的文本:
- 知识库总条目数;
- 置信度较低的条目数量(例如 `confidence_score < 0.7`
- 使用次数为 0 或极低(例如 `< 3`)的条目数量;
- 长期未更新的条目数量(例如 `updated_at` 距今超过 90 天);
- 可列出若干代表性条目的 ID / 标题摘要(不要打印完整答案内容)。
- 脚本不做任何写操作,只读。
你需要:
1. 运行脚本并捕获输出;
2. 根据统计结果,概括知识库当前健康状况(良好 / 一般 / 需要重点治理);
3. 给出 35 条具体的优化建议,如「优先补充高频问题的答案」「合并重复知识点」等。
---
## 四、对用户的输出规范
当成功执行 `kb-audit` 时,应向用户返回包括以下内容的简要报告:
1. **总体健康度**(一句话)
- 例如:「当前知识库共 500 条,其中约 15% 条目置信度偏低10% 长期未更新。」
2. **问题概览**
- 低置信度条目大致数量与比例;
- 使用次数很少的条目数量与可能的原因;
- 长期未更新条目的数量。
3. **优化建议**
- 分点列出建议(如按优先级:先处理高频但低置信度的条目)。
避免:
- 直接打印或暴露完整的知识答案内容(可能包含敏感信息);
- 输出过长的 SQL 或技术细节,优先用运营视角解释。
---
## 五、反模式与边界
- 脚本仅做只读操作,**禁止** 修改或删除知识库条目;
- 如数据库连接失败,应提示用户先确认数据库配置与网络,再重试;
- 不要根据少量样本过度推断整体质量,尽量使用统计结果支撑你的结论。

View File

@@ -0,0 +1,89 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
知识库体检脚本
对 KnowledgeEntry 做简单统计,供 kb-audit Skill 调用。
"""
import sys
from datetime import datetime, timedelta
from pathlib import Path
def add_project_root_to_path():
# 假定脚本位于 .claude/skills/kb-audit/scripts/ 下
script_path = Path(__file__).resolve()
project_root = script_path.parents[4]
if str(project_root) not in sys.path:
sys.path.insert(0, str(project_root))
def main():
add_project_root_to_path()
from src.core.database import db_manager
from src.core.models import KnowledgeEntry
print("=== 知识库健康检查 ===\n")
with db_manager.get_session() as session:
total = session.query(KnowledgeEntry).count()
print(f"知识条目总数: {total}")
# 低置信度(<0.7
low_conf = (
session.query(KnowledgeEntry)
.filter(KnowledgeEntry.confidence_score.isnot(None))
.filter(KnowledgeEntry.confidence_score < 0.7)
.count()
)
print(f"低置信度条目数 (confidence_score < 0.7): {low_conf}")
# 使用次数极低usage_count < 3 或为 NULL
low_usage = (
session.query(KnowledgeEntry)
.filter(
(KnowledgeEntry.usage_count.is_(None))
| (KnowledgeEntry.usage_count < 3)
)
.count()
)
print(f"使用次数极低条目数 (usage_count < 3 或空): {low_usage}")
# 长期未更新(> 90 天)
cutoff = datetime.now() - timedelta(days=90)
old_entries = (
session.query(KnowledgeEntry)
.filter(
(KnowledgeEntry.updated_at.isnot(None))
& (KnowledgeEntry.updated_at < cutoff)
)
.count()
)
print(f"长期未更新条目数 (updated_at > 90 天未更新): {old_entries}")
print("\n示例问题条目(不含完整答案,仅展示前若干个):")
sample_entries = (
session.query(KnowledgeEntry)
.order_by(KnowledgeEntry.created_at.desc())
.limit(5)
.all()
)
for e in sample_entries:
q_preview = (e.question or "")[:40]
print(
f" ID={e.id}, category={e.category}, "
f"confidence={e.confidence_score}, usage={e.usage_count}, "
f"Q='{q_preview}...'"
)
print("\n提示:")
print(" - 建议优先审查低置信度且 usage_count 较高的条目;")
print(" - 对长期未更新且 usage_count 较高的条目,可考虑人工复查内容是否过时;")
print(" - 对 usage_count 极低且从未触发的条目,可考虑合并或归档。")
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,86 @@
---
Name: log-summary
Description: 汇总并分析 TSP 智能助手日志中的 ERROR 与 WARNING输出最近一次启动以来的错误概览和统计帮助快速诊断问题。
---
你是一个「日志错误汇总与分析助手」,技能名为 **log-summary**
你的职责:在用户希望快速了解最近一次或最近几次运行的错误情况时,调用配套脚本,汇总 `logs/` 目录下各启动时间子目录中的日志文件,统计 ERROR / WARNING / CRITICAL并输出简明的错误概览与分布情况。
---
## 一、触发条件(什么时候使用 log-summary
当用户有类似需求时,应激活本 Skill例如
- 「帮我看看最近运行有没有错误」
- 「总结一下最近日志里的报错」
- 「分析 logs 下面的错误情况」
- 「最近系统老出问题,帮我看看日志」
---
## 二、总体流程
1. 调用脚本 `scripts/log_summary.py`,从项目根目录执行。
2. 读取输出并用自然语言向用户转述关键发现。
3. 对明显频繁的错误类型,给出简单的排查建议。
4. 输出时保持简洁,避免粘贴大段原始日志。
---
## 三、脚本调用规范
从项目根目录(包含 `start_dashboard.py` 的目录)执行命令:
```bash
python .claude/skills/log-summary/scripts/log_summary.py
```
脚本行为约定:
- 自动遍历 `logs/` 目录下所有子目录(例如 `logs/2026-02-10_23-51-10/dashboard.log`)。
- 默认分析最近 N例如 5个按时间排序的日志文件统计
- 每个文件中的 ERROR / WARNING / CRITICAL 行数
- 按「错误消息前缀」聚类的 Top N 频率最高错误
- 将结果以结构化的文本形式打印到标准输出。
你需要:
1. 运行脚本并捕获输出;
2. 读懂其中的统计数据与 Top 错误信息;
3. 用 38 句中文自然语言,对用户进行总结说明。
---
## 四、对用户的输出规范
当成功执行 `log-summary` 时,你应该向用户返回类似结构的信息:
1. **总体健康度**(一句话)
- 例如:「最近 3 次启动中共记录 2 条 ERROR、5 条 WARNING整体较为稳定。」
2. **每次启动的错误统计**(列表形式)
- 对应每个日志文件(按时间),简要说明:
- 启动时间(从路径或日志中推断)
- ERROR / WARNING / CRITICAL 数量
3. **Top 错误类型**
- 例如:「最频繁的错误是 `No module named 'src.config.config'`,共出现 4 次。」
4. **简单建议(可选)**
- 对明显重复的错误给出 13 条排查/优化建议。
避免:
- 直接原样复制整段日志;
- 输出过长的技术细节堆栈,优先摘要。
---
## 五、反模式与边界
- 如果 `logs/` 目录不存在或没有任何日志文件:
- 明确告诉用户当前没有可分析的日志,而不是编造结果。
- 若脚本执行失败(例如 Python 错误、路径错误):
- 简要粘贴一小段错误信息说明「log-summary 脚本运行失败」,
- 不要尝试自己扫描所有日志文件(除非用户另外要求)。
- 不要擅自删除或修改日志文件。

View File

@@ -0,0 +1,115 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
简单日志汇总脚本
遍历 logs/ 目录下最近的若干个 dashboard.log 文件,统计 ERROR / WARNING / CRITICAL
并输出简要汇总信息,供 log-summary Skill 调用。
"""
import os
import re
from pathlib import Path
from typing import List, Tuple
LOG_ROOT = Path("logs")
LOG_FILENAME = "dashboard.log"
MAX_FILES = 5 # 最多分析最近 N 个日志文件
LEVEL_PATTERNS = {
"ERROR": re.compile(r"\bERROR\b"),
"WARNING": re.compile(r"\bWARNING\b"),
"CRITICAL": re.compile(r"\bCRITICAL\b"),
}
def find_log_files() -> List[Path]:
if not LOG_ROOT.exists():
return []
candidates: List[Tuple[float, Path]] = []
for root, dirs, files in os.walk(LOG_ROOT):
if LOG_FILENAME in files:
p = Path(root) / LOG_FILENAME
try:
mtime = p.stat().st_mtime
except OSError:
continue
candidates.append((mtime, p))
# 按修改时间从新到旧排序
candidates.sort(key=lambda x: x[0], reverse=True)
return [p for _, p in candidates[:MAX_FILES]]
def summarize_file(path: Path):
counts = {level: 0 for level in LEVEL_PATTERNS.keys()}
top_messages = {}
try:
with path.open("r", encoding="utf-8", errors="ignore") as f:
for line in f:
for level, pattern in LEVEL_PATTERNS.items():
if pattern.search(line):
counts[level] += 1
# 取日志消息部分做前缀(粗略)
msg = line.strip()
# 截断以防过长
msg = msg[:200]
top_messages[msg] = top_messages.get(msg, 0) + 1
break
except OSError as e:
print(f"[!] 读取日志失败 {path}: {e}")
return None
# 取 Top 5
top_list = sorted(top_messages.items(), key=lambda x: x[1], reverse=True)[:5]
return counts, top_list
def main():
log_files = find_log_files()
if not log_files:
print("未找到任何日志文件logs/*/dashboard.log")
return
print(f"共找到 {len(log_files)} 个最近的日志文件(最多 {MAX_FILES} 个):\n")
overall = {level: 0 for level in LEVEL_PATTERNS.keys()}
for idx, path in enumerate(log_files, start=1):
print(f"[{idx}] 日志文件: {path}")
result = summarize_file(path)
if result is None:
print(" 无法读取该日志文件。\n")
continue
counts, top_list = result
for level, c in counts.items():
overall[level] += c
print(
" 级别统计: "
+ ", ".join(f"{lvl}={counts[lvl]}" for lvl in LEVEL_PATTERNS.keys())
)
if top_list:
print(" Top 错误/警告消息:")
for msg, n in top_list:
print(f" [{n}次] {msg}")
else:
print(" 未发现 ERROR/WARNING/CRITICAL 级别日志。")
print()
print("总体统计:")
print(
" "
+ ", ".join(f"{lvl}={overall[lvl]}" for lvl in LEVEL_PATTERNS.keys())
)
if __name__ == "__main__":
main()