From edb54bc5769aa62509ca3d8d8876b03f80da2132 Mon Sep 17 00:00:00 2001 From: handsomeAq Date: Mon, 3 Aug 2026 17:26:00 +0800 Subject: [PATCH] =?UTF-8?q?agent1.2=E7=89=88=E6=9C=AC=20=E4=BF=AE=E6=94=B9?= =?UTF-8?q?chat=E4=B8=BA=E6=B5=81=E5=BC=8F=E8=BE=93=E5=87=BA=20=E5=88=A0?= =?UTF-8?q?=E9=99=A4memory=20=E6=96=B0=E5=A2=9E=E8=BF=9B=E5=BA=A6=E6=9F=A5?= =?UTF-8?q?=E8=AF=A2=E8=84=9A=E6=9C=AC=E5=92=8C=E7=8A=B6=E6=80=81=E6=9F=A5?= =?UTF-8?q?=E8=AF=A2=E8=84=9A=E6=9C=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 274 ++++++++++++++++++++---------- agent_status.py | 348 +++++++++++++++++++++++++++++++++++++++ agents/luna/agent.py | 4 +- agents/luna/chat.py | 119 ++++++++++++- agents/my_agent/agent.py | 2 +- agents/my_agent/chat.py | 121 +++++++++++++- agents/qwen/agent.py | 4 +- agents/qwen/chat.py | 119 ++++++++++++- requirements.txt | 19 +++ watch_session.py | 270 ++++++++++++++++++++++++++++++ 10 files changed, 1164 insertions(+), 116 deletions(-) create mode 100644 agent_status.py create mode 100644 requirements.txt create mode 100644 watch_session.py diff --git a/README.md b/README.md index f71bb91..6d72abe 100644 --- a/README.md +++ b/README.md @@ -1,28 +1,42 @@ -# Dev Agent — 全栈开发子 Agent +# 多 Agent 平台 — 全栈开发子 Agent -基于 Google ADK (Agent Development Kit) 构建的全栈开发子 Agent，通过 A2A / MCP / REST API 多种方式调用，支持文件操作、终端命令、会话持久化、上下文压缩、长期记忆。 +基于 Google ADK (Agent Development Kit) 构建的多模型全栈开发 Agent 平台，内置三个独立 Agent，通过 REST API / MCP / CLI 多种方式调用，支持文件操作、终端命令、会话持久化、上下文压缩、长期记忆。 ## 架构总览 ``` 用户 / CodeBuddy（主控） - │ - ├── MCP ──► mcp_dev_agent/server.py ──┐ - │ │ - └── REST ─► api_server.py ◄─────────────┘ - │ - ▼ - App（dev_app） - │ events_compaction_config（LLM 摘要压缩） - ▼ - LlmAgent（dev_agent / 花花） - │ - ┌─────────┼─────────┐ - ▼ ▼ ▼ - 文件系统 终端命令 记忆系统 - MCP run_command preload_memory + │ + ├─ MCP ──► my_agent_server.py ──┐ + ├─ MCP ──► luna_server.py ──────┤ + └─ MCP ──► qwen_server.py ──────┤ + │ + ┌────────────────┘ + ▼ + 各自独立的 API Server（不同端口） + │ + ▼ + App（dev_app） + │ events_compaction_config（LLM 摘要压缩） + ▼ + LlmAgent（root_agent） + │ + ┌──────────┼──────────┐ + ▼ ▼ ▼ + 文件系统 终端命令 记忆系统 + MCP run_command preload_memory ``` +## 三个 Agent + +| Agent 名称 | 目录 | 模型 | API 端口 | 会话数据库 | MCP 服务器名 | +|-----------|------|------|---------|-----------|-------------| +| `my_agent` | `agents/my_agent/` | aq-first-combo | 8001 | `sessions_my.db` | `my-agent` | +| `luna_agent` | `agents/luna/` | gpt-5.6-luna | 8002 | `sessions_luna.db` | `luna-agent` | +| `qwen_agent` | `agents/qwen/` | astron-code-latest | 8003 | `sessions_qwen.db` | `qwen-agent` | + +每个 Agent 完全独立：独立的模型配置、独立的 API Server、独立的会话数据库、独立的 MCP 入口。 + ## 核心能力 | 能力 | 说明 | @@ -30,9 +44,9 @@ | **文件系统操作** | 读/写/列目录/搜索等 14 个工具（MCP: server-filesystem） | | **终端命令执行** | 异步 subprocess，支持编译/构建/测试 | | **网络搜索** | Tavily 搜索 + Fetch 抓取（默认关闭，见下文说明） | -| **SQLite 会话持久化** | 重启不丢，多入口共享 | +| **SQLite 会话持久化** | 重启不丢，每个 Agent 独立数据库 | | **上下文自动压缩** | 每 20 轮 LLM 摘要，长对话不爆 context window | -| **长期记忆框架** | InMemory + 自动存取，可扩展为 Chroma 向量库 | +| **长期记忆框架** | InMemory + 自动存取，可扩展为向量库 | | **REST API** | `/run`、`/run_sse`、会话管理、Swagger UI | | **MCP 接口** | 可直接接入 CodeBuddy / Cursor / Windsurf | | **A2A 协议** | Agent-to-Agent 标准协议（备用方案） | @@ -45,66 +59,98 @@ # 进入项目目录 cd d:/nzy/workspace_python/agent -# 安装依赖（已装可跳过） -pip install -r requirements.txt +# 创建虚拟环境（已创建可跳过） +python -m venv .venv -# 配置环境变量 -# 编辑 my_agent/.env，填入你的 vLLM API Key 等 +# 激活虚拟环境 +.venv\Scripts\activate # Windows +# source .venv/bin/activate # Linux/Mac + +# 安装依赖 +pip install -r requirements.txt ``` -### 2. 启动 API Server +### 2. 配置环境变量 -这是最主要的服务入口，提供 REST API + Swagger UI。 +每个 Agent 有独立的 `.env` 文件，在对应 agent 目录下： + +``` +agents/my_agent/.env # my_agent 配置 +agents/luna/.env # luna_agent 配置 +agents/qwen/.env # qwen_agent 配置 +``` + +主要配置项（详见下方"配置说明"）： +- `VLLM_API_BASE` — vLLM API 端点 +- `VLLM_MODEL` — 模型名称 +- `VLLM_API_KEY` — API Key +- `AGENT_WORKSPACE_DIR` — Agent 可访问的工作目录 + +### 3. 启动 API Server + +每个 Agent 有独立的 API Server，在对应 agent 目录下启动： ```bash -python api_server.py +# 启动 my_agent（端口 8001） +cd agents/my_agent && python api_server.py + +# 启动 luna_agent（端口 8002） +cd agents/luna && python api_server.py + +# 启动 qwen_agent（端口 8003） +cd agents/qwen && python api_server.py ``` 启动后访问： - **Swagger UI**: http://127.0.0.1:8001/docs — 浏览器直接测试接口 -- **健康检查**: http://127.0.0.1:8001/health - **列出 Agent**: http://127.0.0.1:8001/list-apps -### 3. 配置 MCP（CodeBuddy 调用） +### 4. 配置 MCP（CodeBuddy 调用） -在 CodeBuddy 的 MCP 配置中添加： +全局配置文件路径：`~/.codebuddy/.mcp.json` ```json { "mcpServers": { - "dev-agent": { - "command": "python", - "args": ["d:/nzy/workspace_python/agent/mcp_dev_agent/server.py"], - "env": { - "DEV_AGENT_API_URL": "http://127.0.0.1:8001", - "DEV_AGENT_APP_NAME": "dev_agent", - "DEV_AGENT_USER_ID": "codebuddy" - } + "my-agent": { + "type": "stdio", + "command": "D:\\nzy\\workspace_python\\agent\\.venv\\Scripts\\python.exe", + "args": ["d:\\nzy\\workspace_python\\agent\\mcp_dev_agent\\my_agent_server.py"], + "description": "My Agent (aq-first-combo) 全栈开发助手" + }, + "luna-agent": { + "type": "stdio", + "command": "D:\\nzy\\workspace_python\\agent\\.venv\\Scripts\\python.exe", + "args": ["d:\\nzy\\workspace_python\\agent\\mcp_dev_agent\\luna_server.py"], + "description": "Luna Agent (gpt-5.6-luna) 全栈开发助手" + }, + "qwen-agent": { + "type": "stdio", + "command": "D:\\nzy\\workspace_python\\agent\\.venv\\Scripts\\python.exe", + "args": ["d:\\nzy\\workspace_python\\agent\\mcp_dev_agent\\qwen_server.py"], + "description": "Qwen Agent (astron-code-latest) 全栈开发助手" } } } ``` -重启 CodeBuddy 后，即可通过 `run_dev_agent` 工具调用 Dev Agent。 +> **注意**：Windows 路径使用反斜杠 `\\`。MCP Server 通过 venv 的 python.exe 直接启动，不需要手动激活虚拟环境。 + +重启 CodeBuddy 后，三个 MCP 服务器会自动连接，每个提供一个 `run_dev_agent` 工具。 ## 使用方式 -### 方式一：Swagger UI（最直观） - -打开 http://127.0.0.1:8001/docs ，在浏览器里直接测试。 - -**常用接口**： - -- `POST /run` — 同步运行 agent，返回完整事件列表 -- `POST /run_sse` — SSE 流式运行 -- `GET /apps/{app}/users/{user}/sessions/{id}` — 获取会话 -- `POST /apps/{app}/users/{user}/sessions/{id}` — 创建会话 - -### 方式二：命令行对话 +### 方式一：CLI 对话（每个 Agent 独立） ```bash -# 新会话 -python chat.py +# my_agent 对话 +cd agents/my_agent && python chat.py + +# luna_agent 对话 +cd agents/luna && python chat.py + +# qwen_agent 对话 +cd agents/qwen && python chat.py # 指定 session_id 继续对话 python chat.py --session my_session @@ -116,9 +162,19 @@ python chat.py --list python chat.py --delete my_session ``` +### 方式二：Swagger UI + +打开对应端口的 `/docs`，在浏览器里直接测试接口。 + +**常用接口**： +- `POST /run` — 同步运行 agent，返回完整事件列表 +- `POST /run_sse` — SSE 流式运行 +- `GET /apps/{app}/users/{user}/sessions/{id}` — 获取会话 +- `POST /apps/{app}/users/{user}/sessions/{id}` — 创建会话 + ### 方式三：MCP 工具（CodeBuddy / Cursor） -配置好 MCP 后，直接让 IDE 中的 AI 调用 `run_dev_agent` 工具。 +配置好 MCP 后，直接让 IDE 中的 AI 调用对应 Agent 的 `run_dev_agent` 工具。 **工具参数**： @@ -133,7 +189,7 @@ python chat.py --delete my_session curl -X POST http://127.0.0.1:8001/run \ -H "Content-Type: application/json" \ -d '{ - "appName": "dev_agent", + "appName": "my_agent", "userId": "test_user", "sessionId": "test_001", "newMessage": { @@ -147,37 +203,66 @@ curl -X POST http://127.0.0.1:8001/run \ ``` agent/ -├── api_server.py # REST API Server（主入口） -├── chat.py # CLI 对话工具 -├── a2a_server.py # A2A Server（备用，已被 API Server 取代） -├── a2a_client.py # A2A 客户端测试（调试用） -├── test_sse_client.py # SSE 测试（调试用） +├── api_server.py # 旧版根目录 API Server（my_agent，保留兼容） +├── chat.py # 旧版根目录 CLI（my_agent，保留兼容） +├── a2a_server.py # A2A Server（备用） +├── a2a_client.py # A2A 客户端测试 +├── test_sse_client.py # SSE 测试 +├── requirements.txt # Python 依赖 │ -├── my_agent/ +├── agents/ # 所有 Agent 目录 │ ├── __init__.py -│ ├── agent.py # Agent 定义（人设、工具、instruction） -│ ├── app.py # App 容器（上下文压缩配置） -│ └── .env # 环境变量配置 +│ │ +│ ├── my_agent/ # My Agent (aq-first-combo) +│ │ ├── __init__.py +│ │ ├── agent.py # Agent 定义（人设、工具、instruction） +│ │ ├── app.py # App 容器（上下文压缩配置） +│ │ ├── api_server.py # 独立 API Server（端口 8001） +│ │ ├── chat.py # 独立 CLI 对话工具 +│ │ ├── .env # 环境变量配置 +│ │ └── .adk/ # ADK 会话数据 +│ │ +│ ├── luna/ # Luna Agent (gpt-5.6-luna) +│ │ ├── __init__.py +│ │ ├── agent.py +│ │ ├── app.py +│ │ ├── api_server.py # 独立 API Server（端口 8002） +│ │ ├── chat.py # 独立 CLI 对话工具 +│ │ └── .env +│ │ +│ └── qwen/ # Qwen Agent (astron-code-latest) +│ ├── __init__.py +│ ├── agent.py +│ ├── app.py +│ ├── api_server.py # 独立 API Server（端口 8003） +│ ├── chat.py # 独立 CLI 对话工具 +│ └── .env │ -├── mcp_dev_agent/ # Dev Agent MCP Server（CodeBuddy 入口） -│ ├── server.py -│ └── README.md +├── mcp_dev_agent/ # MCP Server（CodeBuddy 入口） +│ ├── server.py # 通用 MCP Server 逻辑（FastMCP） +│ ├── my_agent_server.py # my_agent MCP 入口（端口 8001） +│ ├── luna_server.py # luna_agent MCP 入口（端口 8002） +│ └── qwen_server.py # qwen_agent MCP 入口（端口 8003） │ -├── mcp_server/ # 旧版 MCP Server（已废弃，保留参考） +├── mcp_server/ # 旧版任务队列 MCP Server（保留参考） │ └── ... │ -├── mcp_tools/ # 备用 MCP 工具（已废弃，保留参考） +├── mcp_tools/ # 备用 MCP 工具（保留参考） │ └── command_executor/ │ ├── data/ # 数据目录（运行时生成） -│ └── sessions.db # SQLite 会话数据库 +│ ├── sessions_my.db # my_agent 会话数据库 +│ ├── sessions_luna.db # luna_agent 会话数据库 +│ └── sessions_qwen.db # qwen_agent 会话数据库 │ └── PLAN.md # 项目计划文档 ``` ## 配置说明 -### 环境变量（my_agent/.env） +### 环境变量（Agent .env） + +每个 Agent 目录下的 `.env` 文件： ```env # vLLM API 配置 @@ -197,31 +282,31 @@ PYTHONUTF8=1 ### API Server 配置 -通过环境变量或直接修改 `agents/my_agent/api_server.py`： +通过环境变量或直接修改对应 `api_server.py`： | 变量 | 默认值 | 说明 | |------|--------|------| | `API_SERVER_HOST` | `0.0.0.0` | 监听地址 | -| `API_SERVER_PORT` | `8001` | 监听端口 | +| `API_SERVER_PORT` | `8001/8002/8003` | 监听端口（各 Agent 不同） | ### MCP Server 配置 -通过环境变量配置： +每个 `*_server.py` 入口脚本顶部硬编码了对应的 API 地址和 Agent 名称： -| 变量 | 默认值 | 说明 | -|------|--------|------| -| `DEV_AGENT_API_URL` | `http://127.0.0.1:8001` | API Server 地址 | -| `DEV_AGENT_APP_NAME` | `dev_agent` | Agent 名称 | -| `DEV_AGENT_USER_ID` | `codebuddy` | 用户 ID（会话隔离用） | +| Agent | API URL | App Name | +|-------|---------|----------| +| my_agent | `http://127.0.0.1:8001` | `my_agent` | +| luna_agent | `http://127.0.0.1:8002` | `luna_agent` | +| qwen_agent | `http://127.0.0.1:8003` | `qwen_agent` | ## 会话与记忆 ### 会话持久化 -所有会话存储在 `data/sessions.db`（SQLite），重启服务不丢失。 +每个 Agent 的会话存储在独立的 SQLite 数据库中（`data/sessions_*.db`），重启服务不丢失。 - **同入口续聊**：同一个 session_id 下次接着聊 -- **跨入口共享**：API Server、CLI、MCP 都用同一个数据库 +- **跨入口共享**：同一个 Agent 的 API Server、CLI、MCP 共用同一个数据库 ### 上下文压缩 @@ -231,7 +316,7 @@ PYTHONUTF8=1 - Token 超阈值紧急压缩（默认 50k） - 原始事件完整保留（可回溯） -配置在 `agents/my_agent/app.py` 的 `EventsCompactionConfig`。 +配置在各 Agent 的 `app.py` 的 `EventsCompactionConfig`。 ### 长期记忆 @@ -259,9 +344,9 @@ read_file、read_text_file、read_media_file、read_multiple_files、write_file ### 网络搜索工具（默认关闭） -Tavily 搜索 + Fetch 抓取默认注释掉了，因为 vLLM 端点有 413（请求体过大）限制。如需启用： +Tavily 搜索 + Fetch 抓取默认注释掉了，因为大响应内容可能导致请求体过大。如需启用： -1. 取消 `agents/my_agent/agent.py` 中 `fetch_mcp` 和 `tavily_mcp` 的注释 +1. 取消对应 `agent.py` 中 `fetch_mcp` 和 `tavily_mcp` 的注释 2. 配置 `TAVILY_API_KEY` 环境变量 ## 工作流程 @@ -286,11 +371,11 @@ Tavily 搜索 + Fetch 抓取默认注释掉了，因为 vLLM 端点有 413（请 ### 本地开发 ```bash -# 终端 1：启动 API Server -python api_server.py +# 终端 1：启动 my_agent API Server +cd agents/my_agent && python api_server.py # 终端 2（可选）：用 CLI 测试 -python chat.py +cd agents/my_agent && python chat.py # 或者直接用 Swagger UI：http://127.0.0.1:8001/docs ``` @@ -308,16 +393,19 @@ python chat.py |------|------|------| | Agent 框架 | Google ADK | 2.5.0 | | LLM 接入 | LiteLLM + vLLM (OpenAI 兼容) | 1.80.0 | -| MCP | Model Context Protocol SDK | 1.29.0 | +| MCP | FastMCP (Model Context Protocol SDK) | 1.29.0 | | HTTP 服务 | FastAPI + Uvicorn | - | | 会话存储 | SQLite | - | | A2A 协议 | a2a-sdk | 1.1.2 | ## 常见问题 -### Q: 启动后 MCP 工具连不上？ +### Q: MCP 服务器连不上？ -A: 第一次启动 npx 需要下载 MCP 包，可能需要 30 秒到 1 分钟。如果超时，检查网络连接。 +A: 请检查： +1. 对应 Agent 的 API Server 是否已启动（`agents/my_agent/api_server.py` 等） +2. `.mcp.json` 中的 python.exe 路径和脚本路径是否正确（Windows 使用反斜杠） +3. 端口是否被占用（`netstat -ano | findstr 8001`） ### Q: 调用时报 413 Request Entity Too Large？ @@ -325,11 +413,15 @@ A: vLLM 端点的 nginx 限制了请求体大小。当前已暂时关闭 Tavily ### Q: 会话数据存在哪？ -A: `data/sessions.db`，SQLite 格式，可用任何 SQLite 工具打开查看。 +A: `data/sessions_*.db`，每个 Agent 有独立的 SQLite 数据库文件。 ### Q: 怎么重置会话？ -A: 用 CLI 的 `python chat.py --delete `，或直接调用 DELETE 会话 API，或直接删除 `data/sessions.db` 文件。 +A: 用 CLI 的 `python chat.py --delete `，或直接调用 DELETE 会话 API，或直接删除对应的 `.db` 文件。 + +### Q: 三个 Agent 有什么区别？ + +A: 区别只在使用的模型不同（aq-first-combo / gpt-5.6-luna / astron-code-latest），工具集和能力完全一致。可以根据任务特点选择合适的模型。 ## 许可证 diff --git a/agent_status.py b/agent_status.py new file mode 100644 index 0000000..d7a048b --- /dev/null +++ b/agent_status.py @@ -0,0 +1,348 @@ +""" +全局 Agent 状态监控脚本 +列出三个 agent 所有会话的实时状态，一眼看出哪些在跑、哪些卡住了。 + +用法： + python agent_status.py # 查看所有 agent 的所有会话 + python agent_status.py --watch # 实时刷新模式（每 3 秒刷新一次） + python agent_status.py -w -i 2 # 实时刷新，间隔 2 秒 + python agent_status.py --agent my_agent # 只看指定 agent + python agent_status.py --limit 20 # 每个 agent 最多显示 20 条 + python agent_status.py -s my_session # 查看特定会话的详细状态 +""" +import argparse +import json +import os +import sys +import time +from datetime import datetime + +import httpx + + +# 三个 agent 的 API 地址 +AGENTS = { + "my_agent": "http://127.0.0.1:8001", + "luna_agent": "http://127.0.0.1:8002", + "qwen_agent": "http://127.0.0.1:8003", +} + +USER_ID = "codebuddy" + + +def check_api_alive(url: str) -> bool: + """检查 API Server 是否存活""" + try: + resp = httpx.get(f"{url}/health", timeout=2.0) + return resp.status_code == 200 + except Exception: + return False + + +def list_sessions(api_url: str, app_name: str, user_id: str) -> list[dict]: + """获取所有会话列表""" + try: + resp = httpx.get( + f"{api_url}/apps/{app_name}/users/{user_id}/sessions", + timeout=5.0, + ) + if resp.status_code == 200: + data = resp.json() + # 响应可能是 list 或 {sessions: [...]} + if isinstance(data, list): + return data + return data.get("sessions", []) + return [] + except Exception: + return [] + + +def get_session_detail(api_url: str, app_name: str, user_id: str, session_id: str) -> dict | None: + """获取会话详情""" + try: + resp = httpx.get( + f"{api_url}/apps/{app_name}/users/{user_id}/sessions/{session_id}", + timeout=5.0, + ) + if resp.status_code == 200: + return resp.json() + return None + except Exception: + return None + + +def format_time(timestamp: float) -> str: + """格式化时间戳""" + if not timestamp: + return "?" + try: + return datetime.fromtimestamp(timestamp).strftime("%m-%d %H:%M:%S") + except Exception: + return str(timestamp)[:19] + + +def get_last_event_summary(events: list[dict]) -> tuple[str, str, str, str]: + """ + 从事件列表提取最后一条事件的摘要信息。 + 返回: (角色, 作者, 状态描述, 内容摘要) + """ + if not events: + return ("-", "-", "(空会话)", "") + + last = events[-1] + content = last.get("content", {}) + role = content.get("role", "?") + author = last.get("author", "") + parts = content.get("parts", []) + + # 判断状态 + status = "" + summary = "" + + role_cn = { + "user": "用户输入", + "model": "模型回复", + "function": "工具调用", + }.get(role, role) + + for part in parts: + if "thought" in part and part.get("thought"): + status = "💭 思考中" + text = part.get("text", "") + summary = text[:60].replace("\n", " ") + break + elif "functionCall" in part: + call = part["functionCall"] + status = f"📞 调用中: {call.get('name', '?')}" + args = call.get("args", {}) + # 显示关键参数 + if "path" in args: + summary = f"path: {args['path'][:50]}" + elif "command" in args: + summary = f"cmd: {args['command'][:50]}" + else: + args_str = json.dumps(args, ensure_ascii=False)[:60] + summary = args_str + break + elif "functionResponse" in part: + resp = part["functionResponse"] + status = f"✅ 工具返回: {resp.get('name', '?')}" + resp_content = resp.get("content", []) + text = "" + for c in resp_content: + if isinstance(c, dict) and c.get("type") == "text": + text += c.get("text", "") + elif isinstance(c, str): + text += c + summary = text[:80].replace("\n", " ") + if resp.get("isError"): + status = f"❌ 工具错误: {resp.get('name', '?')}" + break + elif "text" in part: + status = f"💬 {role_cn}" + summary = part["text"][:80].replace("\n", " ") + break + + if not status: + status = f"📨 {role_cn}" + + return (role, author, status, summary) + + +def print_status_table(agent_names: list[str], limit: int): + """打印所有 agent 的会话状态表格""" + total_sessions = 0 + active_count = 0 + + for agent_name in agent_names: + api_url = AGENTS[agent_name] + alive = check_api_alive(api_url) + + print(f"\n{'═' * 80}") + status_icon = "🟢" if alive else "🔴" + print(f"{status_icon} {agent_name} ({api_url})") + print(f"{'─' * 80}") + + if not alive: + print(" ⚠️ API Server 未启动或无法连接") + continue + + sessions = list_sessions(api_url, agent_name, USER_ID) + total_sessions += len(sessions) + + if not sessions: + print(" （暂无会话）") + continue + + # 按更新时间倒序（字段名可能是 lastUpdateTime 或 last_update_time） + sessions.sort( + key=lambda s: s.get("lastUpdateTime") or s.get("last_update_time", 0), + reverse=True, + ) + + # 只显示 limit 条 + shown = sessions[:limit] + hidden_count = len(sessions) - limit + + print(f" {'#':>3s} {'最后更新时间':<18s} {'事件数':>5s} {'状态'}") + print(f" {'─' * 76}") + + for idx, sess in enumerate(shown, 1): + sid = sess.get("id", "?") + last_time = sess.get("lastUpdateTime") or sess.get("last_update_time", 0) + time_str = format_time(last_time) + + # 取详情获取最后事件 + detail = get_session_detail(api_url, agent_name, USER_ID, sid) + events = detail.get("events", []) if detail else [] + event_count = len(events) + _, _, status, _ = get_last_event_summary(events) + + # 判断是否活跃（5分钟内有更新） + now_ts = time.time() + is_active = (now_ts - last_time) < 300 + if is_active and event_count > 0: + active_count += 1 + active_icon = "⚡" + else: + active_icon = " " + + sid_short = sid if len(sid) <= 12 else sid[:10] + ".." + print(f" {active_icon}{idx:>2d}. {time_str} {event_count:>5d} {status[:50]}") + print(f" id: {sid}") + + if hidden_count > 0: + print(f"\n ... 还有 {hidden_count} 个会话未显示（共 {len(sessions)} 个）") + + print(f"\n{'═' * 80}") + print(f" 总计: {total_sessions} 个会话 | 活跃中（5分钟内有更新）: {active_count}") + print(f"{'═' * 80}\n") + + +def watch_mode(agent_names: list[str], interval: float, limit: int): + """实时刷新模式""" + print(f"\n🔄 实时监控模式（每 {interval} 秒刷新，Ctrl+C 退出）\n") + try: + while True: + # 清屏 + if os.name == "nt": + os.system("cls") + else: + os.system("clear") + print_status_table(agent_names, limit) + print(f" 最后刷新: {datetime.now().strftime('%H:%M:%S')} | Ctrl+C 退出") + time.sleep(interval) + except KeyboardInterrupt: + print("\n👋 已退出监控。") + + +def show_session_detail(agent_name: str, session_id: str): + """查看特定会话的详细状态""" + api_url = AGENTS.get(agent_name, "") + if not api_url: + print(f"未知 agent: {agent_name}") + return + + alive = check_api_alive(api_url) + if not alive: + print(f"⚠️ {agent_name} API Server 未启动（{api_url}）") + return + + detail = get_session_detail(api_url, agent_name, USER_ID, session_id) + if not detail: + print(f"会话 [{session_id}] 不存在") + return + + events = detail.get("events", []) + event_count = len(events) + last_time = detail.get("lastUpdateTime") or detail.get("last_update_time", 0) + # 会话详情里没有 create_time，从第一个事件的时间戳估算 + create_time = detail.get("create_time", 0) + if not create_time and events: + create_time = events[0].get("timestamp", 0) + + print(f"\n{'═' * 80}") + print(f"📋 会话详情") + print(f"{'═' * 80}") + print(f" Agent: {agent_name}") + print(f" Session: {session_id}") + print(f" 创建时间: {format_time(create_time)}") + print(f" 更新时间: {format_time(last_time)}") + print(f" 事件数: {event_count}") + + if events: + duration = last_time - create_time if create_time and last_time else 0 + if duration > 0: + mins = int(duration // 60) + secs = int(duration % 60) + print(f" 运行时长: {mins}分{secs}秒") + + # 最后 5 条事件 + print(f"\n{'─' * 80}") + print(f" 最后 5 条事件：") + print(f"{'─' * 80}") + + for i, event in enumerate(events[-5:], max(1, event_count - 4)): + role, author, status, summary = get_last_event_summary([event]) + ts = event.get("timestamp", 0) + t_str = format_time(ts).split()[-1] if " " in format_time(ts) else format_time(ts) + print(f"\n #{i} [{t_str}] {status}") + if summary: + print(f" {summary[:100]}") + + print(f"\n{'═' * 80}\n") + + +def main(): + global USER_ID + parser = argparse.ArgumentParser(description="全局 Agent 状态监控工具") + parser.add_argument("--agent", "-a", default=None, + help="只查看指定 agent（默认查看所有）") + parser.add_argument("--watch", "-w", action="store_true", + help="实时刷新模式") + parser.add_argument("--interval", "-i", type=float, default=3.0, + help="刷新间隔秒数（默认 3.0）") + parser.add_argument("--limit", "-l", type=int, default=10, + help="每个 agent 最多显示的会话数（默认 10）") + parser.add_argument("--session", "-s", default=None, + help="查看特定会话的详细状态") + parser.add_argument("--user", "-u", default="codebuddy", + help="用户 ID（默认 codebuddy）") + args = parser.parse_args() + + global USER_ID + USER_ID = args.user + + # 确定要查看的 agent 列表 + if args.agent: + agent_name = args.agent + # 支持别名 + aliases = { + "my": "my_agent", "default": "my_agent", "aq": "my_agent", + "luna": "luna_agent", "gpt": "luna_agent", + "qwen": "qwen_agent", "astron": "qwen_agent", + } + if agent_name in aliases: + agent_name = aliases[agent_name] + if agent_name not in AGENTS: + print(f"未知 agent: {args.agent}") + print(f"可用: {list(AGENTS.keys())}") + sys.exit(1) + agent_names = [agent_name] + else: + agent_names = list(AGENTS.keys()) + + # 查看单个会话详情 + if args.session: + show_session_detail(agent_names[0], args.session) + return + + # 实时刷新模式 + if args.watch: + watch_mode(agent_names, args.interval, args.limit) + else: + print_status_table(agent_names, args.limit) + + +if __name__ == "__main__": + main() diff --git a/agents/luna/agent.py b/agents/luna/agent.py index 46d7c9d..3edddc6 100644 --- a/agents/luna/agent.py +++ b/agents/luna/agent.py @@ -71,7 +71,7 @@ filesystem_mcp = McpToolset( # --- 终端命令执行工具（Python 原生，绕开 MCP 通信问题）--- -async def run_command(command: str, cwd: str = None, timeout: int = 300000) -> str: +async def run_command(command: str, cwd: str = None, timeout: int = 300000000) -> str: """ 在终端中执行一条命令，返回输出结果。 @@ -195,6 +195,6 @@ root_agent = LlmAgent( "**验证结果**：编译/测试是否通过，如有警告需列出\n" "**需要主控关注**：如有需要上报的问题，详细说明" ), - tools=[filesystem_mcp, run_command_tool, preload_memory], + tools=[filesystem_mcp, run_command_tool], after_agent_callback=auto_save_memory_callback, ) diff --git a/agents/luna/chat.py b/agents/luna/chat.py index 48d5ac6..7e0f005 100644 --- a/agents/luna/chat.py +++ b/agents/luna/chat.py @@ -34,6 +34,7 @@ from google.adk.runners import Runner from google.adk.sessions.sqlite_session_service import SqliteSessionService from google.adk.memory.in_memory_memory_service import InMemoryMemoryService from google.adk.artifacts.in_memory_artifact_service import InMemoryArtifactService +from google.adk.agents.run_config import RunConfig, StreamingMode from google.genai import types from agents.luna.app import dev_app @@ -136,19 +137,125 @@ async def chat(session_id: str | None = None): print("Luna: ", end="", flush=True) - try: + async def _agent_task(): + """运行 agent 并流式输出，返回是否完成""" + displayed_text = "" + thought_printed = False + run_config = RunConfig(streaming_mode=StreamingMode.SSE) async for event in runner.run_async( user_id=USER_ID, session_id=session_id, new_message=types.Content(parts=[types.Part(text=user_input)]), + run_config=run_config, ): - if event.is_final_response(): - # 最终回复 - for part in event.content.parts: - if hasattr(part, "text") and part.text: - print(part.text, end="", flush=True) + if not event.content or not event.content.parts: + continue + + parts = event.content.parts + + # 1. 思考内容（thought parts）——灰色流式显示 + thought_parts = [ + p.text for p in parts + if hasattr(p, "text") and p.text + and getattr(p, "thought", False) + ] + if thought_parts: + thought_text = "".join(thought_parts) + if not thought_printed: + print("\n\033[90m（思考中…", end="", flush=True) + thought_printed_outer[0] = True + thought_displayed_outer[0] = 0 + if len(thought_text) > thought_displayed_outer[0]: + print(thought_text[thought_displayed_outer[0]:], end="", flush=True) + thought_displayed_outer[0] = len(thought_text) + + # 2. 正式文本——增量显示 + text_parts = [ + p.text for p in parts + if hasattr(p, "text") and p.text + and not getattr(p, "thought", False) + ] + if text_parts: + text = "".join(text_parts) + if len(text) > len(displayed_text): + if thought_printed_outer[0]: + print("\033[0m\nLuna: ", end="", flush=True) + thought_printed_outer[0] = False + new_text = text[len(displayed_text):] + print(new_text, end="", flush=True) + displayed_text = text + + # 3. 工具调用提示 + fcalls = event.get_function_calls() + if fcalls and not event.partial: + if thought_printed_outer[0]: + print("\033[0m", end="", flush=True) + thought_printed_outer[0] = False + for fc in fcalls: + args_str = str(fc.args)[:80] + print(f"\n\033[36m🔧 调用工具: {fc.name}({args_str})\033[0m") + print("Luna: ", end="", flush=True) + + # 4. 最终响应 + if event.is_final_response() and not event.partial: + if thought_printed_outer[0]: + print("\033[0m", end="", flush=True) + thought_printed_outer[0] = False print() + return True + return False + + thought_printed_outer = [False] + thought_displayed_outer = [0] + + # 启动 agent 任务 + 按键监听 + task = asyncio.create_task(_agent_task()) + + async def _keyboard_listener(): + """监听按键，检测到中断键时取消 agent 任务""" + if sys.platform != "win32": + return + import msvcrt + while not task.done(): + await asyncio.sleep(0.05) + if msvcrt.kbhit(): + ch = msvcrt.getwch() + # 支持的中断键: Ctrl+C (0x03), Esc (0x1b), q/Q + if ch in ("\x03", "\x1b", "q", "Q"): + task.cancel() + return + # 功能键/方向键是两个字节的，跳过第二个 + if ch in ("\xe0", "\x00"): + try: + msvcrt.getwch() + except Exception: + pass + + try: + kb_task = asyncio.create_task(_keyboard_listener()) + await task + kb_task.cancel() + try: + await kb_task + except asyncio.CancelledError: + pass + except asyncio.CancelledError: + if thought_printed_outer[0]: + print("\033[0m", end="") + print("\n\033[33m[已中断] 按回车继续输入新指令\033[0m") + # 清空可能残留的输入缓冲 + if sys.platform == "win32": + import msvcrt + while msvcrt.kbhit(): + msvcrt.getwch() + try: + input() + except EOFError: + pass + continue except Exception as e: + if thought_printed_outer[0]: + print("\033[0m", end="") print(f"\n[出错] {e}") print() diff --git a/agents/my_agent/agent.py b/agents/my_agent/agent.py index 4679d78..8a74e5c 100644 --- a/agents/my_agent/agent.py +++ b/agents/my_agent/agent.py @@ -195,6 +195,6 @@ root_agent = LlmAgent( "**验证结果**：编译/测试是否通过，如有警告需列出\n" "**需要主控关注**：如有需要上报的问题，详细说明" ), - tools=[filesystem_mcp, run_command_tool, preload_memory], + tools=[filesystem_mcp, run_command_tool], after_agent_callback=auto_save_memory_callback, ) diff --git a/agents/my_agent/chat.py b/agents/my_agent/chat.py index 0d1ac32..dfd23de 100644 --- a/agents/my_agent/chat.py +++ b/agents/my_agent/chat.py @@ -34,6 +34,7 @@ from google.adk.runners import Runner from google.adk.sessions.sqlite_session_service import SqliteSessionService from google.adk.memory.in_memory_memory_service import InMemoryMemoryService from google.adk.artifacts.in_memory_artifact_service import InMemoryArtifactService +from google.adk.agents.run_config import RunConfig, StreamingMode from google.genai import types from agents.my_agent.app import dev_app @@ -135,21 +136,125 @@ async def chat(session_id: str | None = None): print("花花: ", end="", flush=True) - try: - full_response = "" + async def _agent_task(): + """运行 agent 并流式输出，返回是否完成""" + displayed_text = "" + thought_printed = False + run_config = RunConfig(streaming_mode=StreamingMode.SSE) async for event in runner.run_async( user_id=USER_ID, session_id=session_id, new_message=types.Content(parts=[types.Part(text=user_input)]), + run_config=run_config, ): - if event.is_final_response(): - # 最终回复 - for part in event.content.parts: - if hasattr(part, "text") and part.text: - print(part.text, end="", flush=True) - full_response += part.text + if not event.content or not event.content.parts: + continue + + parts = event.content.parts + + # 1. 思考内容（thought parts）——灰色流式显示 + thought_parts = [ + p.text for p in parts + if hasattr(p, "text") and p.text + and getattr(p, "thought", False) + ] + if thought_parts: + thought_text = "".join(thought_parts) + if not thought_printed: + print("\n\033[90m（思考中…", end="", flush=True) + thought_printed_outer[0] = True + thought_displayed_outer[0] = 0 + if len(thought_text) > thought_displayed_outer[0]: + print(thought_text[thought_displayed_outer[0]:], end="", flush=True) + thought_displayed_outer[0] = len(thought_text) + + # 2. 正式文本——增量显示 + text_parts = [ + p.text for p in parts + if hasattr(p, "text") and p.text + and not getattr(p, "thought", False) + ] + if text_parts: + text = "".join(text_parts) + if len(text) > len(displayed_text): + if thought_printed_outer[0]: + print("\033[0m\n花花: ", end="", flush=True) + thought_printed_outer[0] = False + new_text = text[len(displayed_text):] + print(new_text, end="", flush=True) + displayed_text = text + + # 3. 工具调用提示 + fcalls = event.get_function_calls() + if fcalls and not event.partial: + if thought_printed_outer[0]: + print("\033[0m", end="", flush=True) + thought_printed_outer[0] = False + for fc in fcalls: + args_str = str(fc.args)[:80] + print(f"\n\033[36m🔧 调用工具: {fc.name}({args_str})\033[0m") + print("花花: ", end="", flush=True) + + # 4. 最终响应 + if event.is_final_response() and not event.partial: + if thought_printed_outer[0]: + print("\033[0m", end="", flush=True) + thought_printed_outer[0] = False print() + return True + return False + + thought_printed_outer = [False] + thought_displayed_outer = [0] + + # 启动 agent 任务 + 按键监听 + task = asyncio.create_task(_agent_task()) + + async def _keyboard_listener(): + """监听按键，检测到中断键时取消 agent 任务""" + if sys.platform != "win32": + return + import msvcrt + while not task.done(): + await asyncio.sleep(0.05) + if msvcrt.kbhit(): + ch = msvcrt.getwch() + # 支持的中断键: Ctrl+C (0x03), Esc (0x1b), q/Q + if ch in ("\x03", "\x1b", "q", "Q"): + task.cancel() + return + # 功能键/方向键是两个字节的，跳过第二个 + if ch in ("\xe0", "\x00"): + try: + msvcrt.getwch() + except Exception: + pass + + try: + kb_task = asyncio.create_task(_keyboard_listener()) + await task + kb_task.cancel() + try: + await kb_task + except asyncio.CancelledError: + pass + except asyncio.CancelledError: + if thought_printed_outer[0]: + print("\033[0m", end="") + print("\n\033[33m[已中断] 按回车继续输入新指令\033[0m") + # 清空可能残留的输入缓冲 + if sys.platform == "win32": + import msvcrt + while msvcrt.kbhit(): + msvcrt.getwch() + try: + input() + except EOFError: + pass + continue except Exception as e: + if thought_printed_outer[0]: + print("\033[0m", end="") print(f"\n[出错] {e}") print() diff --git a/agents/qwen/agent.py b/agents/qwen/agent.py index 75a35fb..f1dc1d6 100644 --- a/agents/qwen/agent.py +++ b/agents/qwen/agent.py @@ -71,7 +71,7 @@ filesystem_mcp = McpToolset( # --- 终端命令执行工具（Python 原生，绕开 MCP 通信问题）--- -async def run_command(command: str, cwd: str = None, timeout: int = 300000) -> str: +async def run_command(command: str, cwd: str = None, timeout: int = 300000000) -> str: """ 在终端中执行一条命令，返回输出结果。 @@ -195,6 +195,6 @@ root_agent = LlmAgent( "**验证结果**：编译/测试是否通过，如有警告需列出\n" "**需要主控关注**：如有需要上报的问题，详细说明" ), - tools=[filesystem_mcp, run_command_tool, preload_memory], + tools=[filesystem_mcp, run_command_tool], after_agent_callback=auto_save_memory_callback, ) diff --git a/agents/qwen/chat.py b/agents/qwen/chat.py index e775f4e..35c2b08 100644 --- a/agents/qwen/chat.py +++ b/agents/qwen/chat.py @@ -34,6 +34,7 @@ from google.adk.runners import Runner from google.adk.sessions.sqlite_session_service import SqliteSessionService from google.adk.memory.in_memory_memory_service import InMemoryMemoryService from google.adk.artifacts.in_memory_artifact_service import InMemoryArtifactService +from google.adk.agents.run_config import RunConfig, StreamingMode from google.genai import types from agents.qwen.app import dev_app @@ -136,19 +137,125 @@ async def chat(session_id: str | None = None): print("Qwen: ", end="", flush=True) - try: + async def _agent_task(): + """运行 agent 并流式输出，返回是否完成""" + displayed_text = "" + thought_printed = False + run_config = RunConfig(streaming_mode=StreamingMode.SSE) async for event in runner.run_async( user_id=USER_ID, session_id=session_id, new_message=types.Content(parts=[types.Part(text=user_input)]), + run_config=run_config, ): - if event.is_final_response(): - # 最终回复 - for part in event.content.parts: - if hasattr(part, "text") and part.text: - print(part.text, end="", flush=True) + if not event.content or not event.content.parts: + continue + + parts = event.content.parts + + # 1. 思考内容（thought parts）——灰色流式显示 + thought_parts = [ + p.text for p in parts + if hasattr(p, "text") and p.text + and getattr(p, "thought", False) + ] + if thought_parts: + thought_text = "".join(thought_parts) + if not thought_printed: + print("\n\033[90m（思考中…", end="", flush=True) + thought_printed_outer[0] = True + thought_displayed_outer[0] = 0 + if len(thought_text) > thought_displayed_outer[0]: + print(thought_text[thought_displayed_outer[0]:], end="", flush=True) + thought_displayed_outer[0] = len(thought_text) + + # 2. 正式文本——增量显示 + text_parts = [ + p.text for p in parts + if hasattr(p, "text") and p.text + and not getattr(p, "thought", False) + ] + if text_parts: + text = "".join(text_parts) + if len(text) > len(displayed_text): + if thought_printed_outer[0]: + print("\033[0m\nQwen: ", end="", flush=True) + thought_printed_outer[0] = False + new_text = text[len(displayed_text):] + print(new_text, end="", flush=True) + displayed_text = text + + # 3. 工具调用提示 + fcalls = event.get_function_calls() + if fcalls and not event.partial: + if thought_printed_outer[0]: + print("\033[0m", end="", flush=True) + thought_printed_outer[0] = False + for fc in fcalls: + args_str = str(fc.args)[:80] + print(f"\n\033[36m🔧 调用工具: {fc.name}({args_str})\033[0m") + print("Qwen: ", end="", flush=True) + + # 4. 最终响应 + if event.is_final_response() and not event.partial: + if thought_printed_outer[0]: + print("\033[0m", end="", flush=True) + thought_printed_outer[0] = False print() + return True + return False + + thought_printed_outer = [False] + thought_displayed_outer = [0] + + # 启动 agent 任务 + 按键监听 + task = asyncio.create_task(_agent_task()) + + async def _keyboard_listener(): + """监听按键，检测到中断键时取消 agent 任务""" + if sys.platform != "win32": + return + import msvcrt + while not task.done(): + await asyncio.sleep(0.05) + if msvcrt.kbhit(): + ch = msvcrt.getwch() + # 支持的中断键: Ctrl+C (0x03), Esc (0x1b), q/Q + if ch in ("\x03", "\x1b", "q", "Q"): + task.cancel() + return + # 功能键/方向键是两个字节的，跳过第二个 + if ch in ("\xe0", "\x00"): + try: + msvcrt.getwch() + except Exception: + pass + + try: + kb_task = asyncio.create_task(_keyboard_listener()) + await task + kb_task.cancel() + try: + await kb_task + except asyncio.CancelledError: + pass + except asyncio.CancelledError: + if thought_printed_outer[0]: + print("\033[0m", end="") + print("\n\033[33m[已中断] 按回车继续输入新指令\033[0m") + # 清空可能残留的输入缓冲 + if sys.platform == "win32": + import msvcrt + while msvcrt.kbhit(): + msvcrt.getwch() + try: + input() + except EOFError: + pass + continue except Exception as e: + if thought_printed_outer[0]: + print("\033[0m", end="") print(f"\n[出错] {e}") print() diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..44dc5fe --- /dev/null +++ b/requirements.txt @@ -0,0 +1,19 @@ +# 核心框架 +google-adk>=2.5.0 +google-genai>=2.14.0 +litellm>=1.80.0 + +# MCP +mcp>=1.29.0 +mcp-types>=2.0.0 +httpx>=0.28.0 + +# Web 服务 +fastapi>=0.140.0 +uvicorn>=0.51.0 + +# 配置 +python-dotenv>=1.2.0 + +# A2A（可选） +a2a-sdk>=1.1.2 diff --git a/watch_session.py b/watch_session.py new file mode 100644 index 0000000..5468a68 --- /dev/null +++ b/watch_session.py @@ -0,0 +1,270 @@ +""" +Session 实时监控脚本 +输入 session_id，实时打印该会话中的所有新事件（用户输入、模型回复、工具调用等）。 + +用法： + python watch_session.py --session + python watch_session.py -s --agent my_agent + python watch_session.py -s test_001 --poll 2.0 + +支持的参数： + --session / -s : 会话 ID（必填） + --agent / -a : Agent 名称，默认 my_agent（可选：my_agent / luna_agent / qwen_agent） + --user / -u : 用户 ID，默认 codebuddy + --poll / -p : 轮询间隔（秒），默认 1.5 + --url : API Server 地址，默认根据 agent 自动选择 +""" +import argparse +import json +import sys +import time +from datetime import datetime + +import httpx + + +# Agent 对应的默认 API 地址 +AGENT_URLS = { + "my_agent": "http://127.0.0.1:8001", + "luna_agent": "http://127.0.0.1:8002", + "qwen_agent": "http://127.0.0.1:8003", +} + +# Agent 名称别名 +AGENT_ALIASES = { + "my": "my_agent", + "default": "my_agent", + "aq": "my_agent", + "luna": "luna_agent", + "gpt": "luna_agent", + "qwen": "qwen_agent", + "astron": "qwen_agent", +} + + +def resolve_agent(name: str) -> str: + """解析 agent 名称""" + name = name.strip().lower() + if name in AGENT_URLS: + return name + if name in AGENT_ALIASES: + return AGENT_ALIASES[name] + for full_name in AGENT_URLS: + if name in full_name: + return full_name + raise ValueError( + f"未知的 agent: {name}\n" + f"可用: {list(AGENT_URLS.keys())}\n" + f"别名: {list(AGENT_ALIASES.keys())}" + ) + + +def get_api_url(agent_name: str, custom_url: str | None) -> str: + """获取 API 地址""" + if custom_url: + return custom_url.rstrip("/") + return AGENT_URLS[agent_name] + + +def fetch_session(api_url: str, app_name: str, user_id: str, session_id: str) -> dict | None: + """获取会话数据""" + try: + resp = httpx.get( + f"{api_url}/apps/{app_name}/users/{user_id}/sessions/{session_id}", + timeout=10.0, + ) + if resp.status_code == 200: + return resp.json() + if resp.status_code == 404: + return None + print(f"[警告] 获取会话失败 (HTTP {resp.status_code}): {resp.text[:200]}") + return None + except Exception as e: + print(f"[警告] 连接 API Server 失败: {e}") + return None + + +def format_event(event: dict, index: int) -> str: + """格式化单个事件为可读字符串""" + content = event.get("content", {}) + role = content.get("role", "?") + author = event.get("author", "") + parts = content.get("parts", []) + timestamp = event.get("timestamp", 0) + + time_str = "" + if timestamp: + try: + time_str = datetime.fromtimestamp(timestamp).strftime("%H:%M:%S") + except Exception: + time_str = str(timestamp) + + role_label = { + "user": "👤 用户", + "model": "🤖 模型", + "function": "🔧 工具", + }.get(role, f"❓ {role}") + + author_str = f" [{author}]" if author else "" + header = f"\n{'─' * 60}\n[{time_str}] {role_label}{author_str} #{index}\n{'─' * 60}" + + lines = [header] + + for part in parts: + if "text" in part: + text = part["text"] + # thoughts 单独标注 + if part.get("thought"): + lines.append(f"💭 [思考中]\n{text}\n") + else: + lines.append(f"{text}\n") + + elif "functionCall" in part: + call = part["functionCall"] + args_str = json.dumps(call.get("args", {}), ensure_ascii=False, indent=2) + # 太长就截断 + if len(args_str) > 500: + args_str = args_str[:500] + f"\n... (共 {len(args_str)} 字符，已截断)" + lines.append(f"📞 调用工具: {call.get('name', '?')}\n{args_str}\n") + + elif "functionResponse" in part: + resp = part["functionResponse"] + resp_name = resp.get("name", "?") + resp_content = resp.get("content", []) + # 提取文本内容 + text_parts = [] + for c in resp_content: + if isinstance(c, dict) and c.get("type") == "text": + text_parts.append(c.get("text", "")) + elif isinstance(c, str): + text_parts.append(c) + result_text = "\n".join(text_parts) if text_parts else str(resp_content) + # 太长就截断 + if len(result_text) > 800: + result_text = result_text[:800] + f"\n... (共 {len(result_text)} 字符，已截断)" + lines.append(f"✅ 工具返回: {resp_name}\n{result_text}\n") + + elif "code" in part: + code = part["code"] + lines.append(f"📝 代码片段:\n```\n{code}\n```\n") + + elif "executableCode" in part: + ec = part["executableCode"] + lines.append(f"💻 可执行代码 ({ec.get('language', '?')}):\n```\n{ec.get('code', '')[:500]}\n```\n") + + else: + part_types = list(part.keys()) + lines.append(f"[其他内容] 类型: {part_types}\n") + + return "\n".join(lines) + + +def extract_events(session_data: dict) -> list[dict]: + """从会话数据中提取事件列表""" + return session_data.get("events", []) or [] + + +def watch_session( + api_url: str, + app_name: str, + user_id: str, + session_id: str, + poll_interval: float, +): + """实时监控会话""" + print(f"🔍 开始监控会话") + print(f" Agent: {app_name}") + print(f" API: {api_url}") + print(f" 用户: {user_id}") + print(f" 会话ID: {session_id}") + print(f" 轮询间隔: {poll_interval}s") + print(f" 按 Ctrl+C 退出\n") + + last_event_count = 0 + + # 首次获取，如果有历史事件，问要不要回放 + session = fetch_session(api_url, app_name, user_id, session_id) + if session is None: + print(f"会话 [{session_id}] 不存在，请检查 session_id 和 agent 是否正确。") + print(f"提示: 确认 {app_name} 的 API Server 是否已启动（{api_url}）") + return + + events = extract_events(session) + existing_count = len(events) + + if existing_count > 0: + print(f"📜 该会话已有 {existing_count} 条历史事件。") + try: + choice = input("是否打印历史事件？(y/n，默认 n): ").strip().lower() + except (EOFError, KeyboardInterrupt): + print("\n已退出。") + return + + if choice in ("y", "yes"): + for i, event in enumerate(events, 1): + print(format_event(event, i)) + last_event_count = existing_count + print(f"\n✅ 历史事件回放完毕，共 {existing_count} 条。") + print(f" 现在开始监控新事件...\n") + else: + last_event_count = existing_count + print(f" 跳过历史，从第 {existing_count + 1} 条开始监控新事件...\n") + else: + print("📭 该会话目前没有事件，等待新事件...\n") + + # 开始轮询 + try: + while True: + time.sleep(poll_interval) + session = fetch_session(api_url, app_name, user_id, session_id) + if session is None: + continue + + events = extract_events(session) + current_count = len(events) + + if current_count > last_event_count: + # 有新事件 + for i in range(last_event_count, current_count): + print(format_event(events[i], i + 1)) + last_event_count = current_count + + # 检测是否结束（最后一条是 model role 的 final 事件） + # 这里不自动退出，继续轮询，因为可能有多轮对话 + + except KeyboardInterrupt: + print(f"\n\n👋 已停止监控。共检测到 {last_event_count} 条事件。") + + +def main(): + parser = argparse.ArgumentParser(description="Session 实时监控工具") + parser.add_argument("--session", "-s", required=True, help="会话 ID") + parser.add_argument("--agent", "-a", default="my_agent", + help="Agent 名称（默认 my_agent）") + parser.add_argument("--user", "-u", default="codebuddy", + help="用户 ID（默认 codebuddy）") + parser.add_argument("--poll", "-p", type=float, default=1.5, + help="轮询间隔秒数（默认 1.5）") + parser.add_argument("--url", default=None, + help="自定义 API Server 地址（覆盖默认）") + args = parser.parse_args() + + try: + agent_name = resolve_agent(args.agent) + except ValueError as e: + print(str(e)) + sys.exit(1) + + api_url = get_api_url(agent_name, args.url) + + watch_session( + api_url=api_url, + app_name=agent_name, + user_id=args.user, + session_id=args.session, + poll_interval=args.poll, + ) + + +if __name__ == "__main__": + main()