--- name: ai-chat description: AI Chat 测试与管理。发送消息测试 AI Chat 工具调用链路，管理工具开关和 Provider 配置，支持 local/staging 环境切换。 arguments: [args] --- # AI Chat Skill 测试和管理 Coolbuy PaaS AI Chat 服务的 Claude Code skill。 ## Quick Reference | 命令 | 用途 | |------|------| | `/ai-chat send ` | 发送消息到 AI Chat，实时显示工具调用 + AI 回复 | | `/ai-chat env [local\|staging]` | 切换/查看目标环境（默认 local） | | `/ai-chat tools [category]` | 列出当前环境已注册的工具 | | `/ai-chat config` | 查看 AI 配置（Provider、工具开关等） | | `/ai-chat history` | 显示本次会话的历史消息 | --- ## Environment Config 两套环境，通过 `/ai-chat env` 切换： | 环境 | Auth URL | AI URL | 登录账号 | |------|----------|--------|----------| | **local** (默认) | `http://localhost:7089` | `http://localhost:7092` | lining_admin / admin123 | | **staging** | `http://39.105.150.219:7089` | `http://39.105.150.219:7092` | lining_admin / admin123 | ### 状态文件 环境状态保存在 `/tmp/ai-chat-state.json`，格式： ```json { "env": "local", "token": "eyJ...", "token_env": "local", "history": [] } ``` --- ## Commands ### /ai-chat env **切换或查看当前环境。** 用法： - `/ai-chat env` — 显示当前环境 - `/ai-chat env local` — 切换到本地环境 - `/ai-chat env staging` — 切换到 staging 环境 实现步骤： 1. 读取 `/tmp/ai-chat-state.json`（不存在则默认 `{"env":"local","history":[]}`） 2. 如果提供了参数，更新 `env` 字段并清空 `token`（环境变了 token 失效） 3. 写回状态文件 4. 输出当前环境信息表格 --- ### /ai-chat send **发送消息到 AI Chat 并实时显示流式响应。** 用法：`/ai-chat send ` 实现步骤： #### Step 1: 读取状态 ```bash # 读取状态文件 cat /tmp/ai-chat-state.json 2>/dev/null || echo '{"env":"local","history":[]}' ``` 确定环境变量： - **local**: `AUTH_URL=http://localhost:7089`, `AI_URL=http://localhost:7092` - **staging**: `AUTH_URL=http://39.105.150.219:7089`, `AI_URL=http://39.105.150.219:7092` #### Step 2: 获取 Token 如果状态文件中没有 token 或 `token_env` 与当前 `env` 不匹配，执行登录： ```bash curl -s -X POST "$AUTH_URL/api/v1/auth/login" \ -H "Content-Type: application/json" \ -d '{"username":"lining_admin","password":"admin123"}' ``` 从响应中提取 `access_token`： ```bash # 响应格式 # {"access_token":"eyJ...","refresh_token":"...","token_type":"Bearer","expires_in":7200,"user_info":{...}} ``` 用 `python3 -c "import json,sys; print(json.load(sys.stdin)['access_token'])"` 提取 token。 将 token 和 token_env 保存到状态文件。 #### Step 3: 构造请求并发送 SSE 流 ```bash curl -s -N -X POST "$AI_URL/api/v1/ai/chat/stream" \ -H "Authorization: Bearer $TOKEN" \ -H "Content-Type: application/json" \ -d "{\"message\":\"$MSG\",\"history\":$HISTORY}" 2>&1 ``` **重要**: `history` 字段传入之前的会话历史（从状态文件读取），实现多轮对话。 #### Step 4: 解析 SSE 事件 用 Python 脚本解析 SSE 流（比 bash while read 更可靠）： ```python #!/usr/bin/env python3 """解析 AI Chat SSE 流并格式化输出""" import sys, json full_content = "" tool_calls = [] for line in sys.stdin: line = line.strip() if not line.startswith("data:"): continue data_str = line[5:].strip() if not data_str: continue try: event = json.loads(data_str) except json.JSONDecodeError: continue evt_type = event.get("type", "") if evt_type == "content": chunk = event.get("content", "") full_content += chunk # 实时输出内容片段 sys.stdout.write(chunk) sys.stdout.flush() elif evt_type == "tool_call": tc = event.get("tool_call", {}) tool_name = tc.get("name", "unknown") tool_args = tc.get("arguments", {}) tool_id = tc.get("id", "") tool_calls.append({"id": tool_id, "name": tool_name}) # 输出工具调用标记 args_str = json.dumps(tool_args, ensure_ascii=False) if len(args_str) > 200: args_str = args_str[:200] + "..." print(f"\n🔧 Tool Call: {tool_name}", file=sys.stderr) print(f" Args: {args_str}", file=sys.stderr) elif evt_type == "tool_result": tr = event.get("tool_result", {}) tool_name = tr.get("name", "unknown") content = tr.get("content", "") is_error = tr.get("is_error", False) # 截断长结果 if len(content) > 500: content = content[:500] + f"... ({len(content)} chars total)" status = "❌ Error" if is_error else "✅ Result" print(f" {status} [{tool_name}]: {content}", file=sys.stderr) elif evt_type == "done": usage = event.get("usage") or {} prompt_t = usage.get("prompt_tokens", 0) completion_t = usage.get("completion_tokens", 0) total_t = usage.get("total_tokens", 0) print(f"\n\n--- Done ---", file=sys.stderr) if total_t > 0: print(f"Tokens: {prompt_t} prompt + {completion_t} completion = {total_t} total", file=sys.stderr) if tool_calls: print(f"Tool calls: {len(tool_calls)} ({', '.join(tc['name'] for tc in tool_calls)})", file=sys.stderr) elif evt_type == "error": err = event.get("error", "unknown error") print(f"\n❌ Error: {err}", file=sys.stderr) # 输出换行 print() # 将 full_content 输出到 fd 3 用于状态更新（如果 fd 3 打开） try: with open("/tmp/ai-chat-response.txt", "w") as f: f.write(full_content) except: pass ``` #### Step 5: 更新会话历史 发送完成后，将用户消息和 AI 回复追加到状态文件的 `history` 数组中： ```json [ {"role": "user", "content": "<用户消息>"}, {"role": "assistant", "content": ""} ] ``` #### 完整 bash 执行流程 ```bash #!/usr/bin/env bash set -euo pipefail MSG="$1" STATE_FILE="/tmp/ai-chat-state.json" # 1. 读取状态 if [ -f "$STATE_FILE" ]; then STATE=$(cat "$STATE_FILE") else STATE='{"env":"local","history":[]}' fi ENV=$(echo "$STATE" | python3 -c "import json,sys; print(json.load(sys.stdin).get('env','local'))") TOKEN=$(echo "$STATE" | python3 -c "import json,sys; print(json.load(sys.stdin).get('token',''))") TOKEN_ENV=$(echo "$STATE" | python3 -c "import json,sys; print(json.load(sys.stdin).get('token_env',''))") HISTORY=$(echo "$STATE" | python3 -c "import json,sys; print(json.dumps(json.load(sys.stdin).get('history',[])))") # 2. 确定 URL if [ "$ENV" = "staging" ]; then AUTH_URL="http://39.105.150.219:7089" AI_URL="http://39.105.150.219:7092" else AUTH_URL="http://localhost:7089" AI_URL="http://localhost:7092" fi # 3. 获取 token（如果需要） if [ -z "$TOKEN" ] || [ "$TOKEN_ENV" != "$ENV" ]; then echo "🔐 Logging in to $ENV environment..." LOGIN_RESP=$(curl -s -X POST "$AUTH_URL/api/v1/auth/login" \ -H "Content-Type: application/json" \ -d '{"username":"lining_admin","password":"admin123"}') TOKEN=$(echo "$LOGIN_RESP" | python3 -c "import json,sys; d=json.load(sys.stdin); print(d.get('access_token',d.get('data',{}).get('access_token','')))") if [ -z "$TOKEN" ]; then echo "❌ Login failed: $LOGIN_RESP" exit 1 fi echo "✅ Login successful" # 更新状态中的 token STATE=$(echo "$STATE" | python3 -c " import json,sys s=json.load(sys.stdin) s['token']='$TOKEN' s['token_env']='$ENV' print(json.dumps(s,ensure_ascii=False)) ") fi # 4. 发送 SSE 请求并解析 echo "" echo "📤 Sending to $ENV: $MSG" echo "---" # 转义消息中的特殊字符 MSG_JSON=$(python3 -c "import json; print(json.dumps('$MSG'))") curl -s -N -X POST "$AI_URL/api/v1/ai/chat/stream" \ -H "Authorization: Bearer $TOKEN" \ -H "Content-Type: application/json" \ -d "{\"message\":$MSG_JSON,\"history\":$HISTORY}" 2>&1 | python3 -c " import sys, json full_content = '' tool_calls = [] for line in sys.stdin: line = line.strip() if not line.startswith('data:'): continue data_str = line[5:].strip() if not data_str: continue try: event = json.loads(data_str) except json.JSONDecodeError: continue evt_type = event.get('type', '') if evt_type == 'content': chunk = event.get('content', '') full_content += chunk sys.stdout.write(chunk) sys.stdout.flush() elif evt_type == 'tool_call': tc = event.get('tool_call', {}) tool_name = tc.get('name', 'unknown') tool_args = tc.get('arguments', {}) tool_calls.append({'name': tool_name}) args_str = json.dumps(tool_args, ensure_ascii=False) if len(args_str) > 200: args_str = args_str[:200] + '...' print(f'\n🔧 Tool Call: {tool_name}', file=sys.stderr) print(f' Args: {args_str}', file=sys.stderr) elif evt_type == 'tool_result': tr = event.get('tool_result', {}) tool_name = tr.get('name', 'unknown') content = tr.get('content', '') is_error = tr.get('is_error', False) if len(content) > 500: content = content[:500] + f'... ({len(content)} chars total)' status = '❌' if is_error else '✅' print(f' {status} [{tool_name}]: {content}', file=sys.stderr) elif evt_type == 'done': usage = event.get('usage') or {} pt = usage.get('prompt_tokens', 0) ct = usage.get('completion_tokens', 0) tt = usage.get('total_tokens', 0) print(f'\n\n--- Done ---', file=sys.stderr) if tt > 0: print(f'Tokens: {pt} prompt + {ct} completion = {tt} total', file=sys.stderr) if tool_calls: names = ', '.join(tc['name'] for tc in tool_calls) print(f'Tool calls: {len(tool_calls)} ({names})', file=sys.stderr) elif evt_type == 'error': err = event.get('error', 'unknown error') print(f'\n❌ Error: {err}', file=sys.stderr) print() with open('/tmp/ai-chat-response.txt', 'w') as f: f.write(full_content) " # 5. 更新历史 RESPONSE=$(cat /tmp/ai-chat-response.txt 2>/dev/null || echo "") python3 -c " import json state_file = '$STATE_FILE' try: with open(state_file) as f: state = json.load(f) except: state = {'env': 'local', 'history': []} msg = $MSG_JSON resp = '''$RESPONSE''' state['history'].append({'role': 'user', 'content': msg}) if resp: state['history'].append({'role': 'assistant', 'content': resp}) state['token'] = '$TOKEN' state['token_env'] = '$ENV' with open(state_file, 'w') as f: json.dump(state, f, ensure_ascii=False, indent=2) " echo "" echo "💬 History: $(echo "$STATE" | python3 -c "import json,sys; h=json.load(sys.stdin).get('history',[]); print(len(h)//2 + 1)") messages in session" ``` **重要注意事项**： - 上面的脚本是逻辑参考，**不要**原样执行。Claude Code 应按步骤逐一执行 bash 命令。 - 消息中的引号和特殊字符需要用 python3 json.dumps 转义。 - 如果 token 过期（401 响应），自动重新登录。 - SSE 超时设置 `--max-time 120`。 --- ### /ai-chat tools **列出当前环境已注册的工具。** 用法： - `/ai-chat tools` — 列出所有工具 - `/ai-chat tools ` — 按分类过滤 实现步骤： 1. 读取状态获取环境和 token（必要时先登录） 2. 发送请求： ```bash # 列出所有工具 curl -s "$AI_URL/api/v1/ai/tools" \ -H "Authorization: Bearer $TOKEN" # 按分类过滤 curl -s "$AI_URL/api/v1/ai/tools?category=order" \ -H "Authorization: Bearer $TOKEN" ``` 3. 响应格式： ```json { "tools": [ { "name": "list_orders", "description": "查询订单列表", "category": "order", "enabled": true, "parameters": {...} } ] } ``` 4. 按 category 分组，输出表格： ``` 📋 AI Tools (142 total) order (15 tools) ├── list_orders 查询订单列表 ├── get_order_detail 获取订单详情 └── ... product (12 tools) ├── list_products 查询商品列表 └── ... ``` 已知工具分类：order, product, sku, inventory, task, brand, requirement, customer, dashboard, distribution, finance, discount, channel, approval, organization, feature_gap --- ### /ai-chat config **查看 AI 服务配置。** 实现步骤： 1. 根据当前环境读取对应配置文件： - **local**: 读取 `ai-service/api/etc/ai-api-local.yaml` - **staging**: SSH 到 staging 读取 `/opt/coolbuy/configs/ai-api.yaml` 2. 显示关键配置项： - AI Provider 和 Model - 各工具分类的开关状态 - API 端口 3. 输出格式： ``` ⚙️ AI Config (local) Provider: deepseek Model: deepseek-chat Port: 7092 Tool Categories: ✅ order ✅ product ✅ sku ✅ inventory ✅ task ✅ brand ✅ requirement ✅ customer ✅ dashboard ✅ distribution ✅ finance ✅ discount ✅ channel ✅ approval ✅ organization ✅ feature_gap ``` --- ### /ai-chat history **显示当前会话的历史消息。** 实现步骤： 1. 读取 `/tmp/ai-chat-state.json` 的 `history` 数组 2. 按时间顺序显示： ``` 💬 Chat History (3 messages) [1] 👤 User: 你好 🤖 AI: 你好！我是 AI 助手... [2] 👤 User: 搜索组织架构找到大客户部 🤖 AI: 我来帮你搜索... (used: search_organizations) [3] 👤 User: 最近的订单 🤖 AI: 以下是最近的订单列表... ``` 3. 如果需要清空历史：`/ai-chat history clear` - 删除状态文件中的 history 数组，重置为空 --- ## SSE Event Format Reference AI Chat SSE 流使用 `event: message` + `data: {json}` 格式： | type | 数据字段 | 说明 | |------|---------|------| | `content` | `content: ""` | 增量文本内容 | | `tool_call` | `tool_call: {id, name, arguments}` | AI 请求调用工具 | | `tool_result` | `tool_result: {tool_call_id, name, content, is_error}` | 工具执行结果 | | `done` | `usage: {prompt_tokens, completion_tokens, total_tokens}` (可能为 null), `finish_reason` | 流结束 | | `error` | `error: ""` | 错误 | --- ## Troubleshooting ### Token 过期 (401) 如果请求返回 401，删除缓存 token 重新登录： ```bash # 清除 token 强制重新登录 python3 -c " import json with open('/tmp/ai-chat-state.json') as f: s=json.load(f) s['token']='' with open('/tmp/ai-chat-state.json','w') as f: json.dump(s,f) " ``` ### 连接失败 - **local**: 确认本地服务已启动 (`./scripts/start_dev.sh`) - **staging**: 确认 staging 服务运行中 (`ssh coolbuy-staging "docker ps | grep ai-service"`) ### SSE 流中断 - 检查 AI 服务日志 - local: 查看终端输出 - staging: `ssh coolbuy-staging "docker logs coolbuy-ai-service --tail 50"` ### 消息中包含特殊字符 务必用 `python3 -c "import json; print(json.dumps(msg))"` 转义消息内容，避免 JSON 解析失败。