feat(docs): reorganize documentation and add orphan tool calls recovery plan

- Move command-security.md and hitl-implementation-plan.md to todo/ directory
- Add new documentation for orphan tool calls recovery with detailed problem analysis and implementation plan
- Document scenarios causing orphan tool calls and propose fix strategy using LangGraph API
This commit is contained in:
YueGuobin committed 2026-03-05 23:07:02 +08:00
1 parent b8cb0bdcb0
commit a7bc68d6e1
4 files changed
+451

No files matched your search

+211
View File
@@ -0,0 +1,211 @@
# 命令安全配置
## 概述
GNS3-Copilot 包含一个命令安全验证系统,用于防止在实验室环境中执行潜在危险或不合适的命令。这有助于:
- **防止安全告警**:`traceroute` 等命令可能触发安全监控系统
- **保护实验室设备**:危险命令(泛洪 ping、debug)可能压垮或崩溃设备
- **保持安全性**:在教学环境中尤其重要
## 默认禁止的命令
### 全局禁止命令(所有工具)
| 模式 | 描述 | 原因 |
|------|------|------|
| `traceroute` | 网络路由跟踪 | 可能触发安全告警并产生大量流量 |
| `tracepath` | 网络路由跟踪 | 类似 traceroute |
| `tracert` | Windows traceroute | 类似 traceroute |
| `ping.*-f` | 泛洪 ping | 可能压垮实验室设备 |
| `ping.*\s+count\s+\d{3,}` | 大量 ping | 过量的 ping 流量 |
| `^debug\s+` | 调试命令 | 可能崩溃或 destabilize 设备 |
| `^test\s+` | 测试命令 | 可能影响设备稳定性 |
### 工具特定限制
**诊断工具**(仅允许诊断/show 命令):
- `conf t` / `configure terminal` - 配置模式
- `end` / `exit` - 配置模式命令
**VPCS 工具**:
- `traceroute` - VPCS 路由跟踪
## 用户配置
### 覆盖默认规则
要允许默认禁止的特定命令,创建 `command_security_override.json` 文件:
```json
{
"allowed_commands": [
"traceroute"
]
}
```
### 添加额外的禁止命令
要添加自己的限制:
```json
{
"forbidden_commands": [
"my_dangerous_command",
"another_pattern.*"
]
}
```
### 完整示例
```json
{
"allowed_commands": [
"traceroute"
],
"forbidden_commands": [
"show running-config"
]
}
```
## 配置文件位置
覆盖文件将在以下位置查找(按顺序):
1. `/etc/gns3-server/command_security_override.json`(系统级)
2. `<项目目录>/command_security_override.json`(项目级)
找到的第一个文件将被使用。
## 安全级别
虽然没有明确的"安全级别",但默认配置设计为适合大多数实验室和教学场景:
- **教学安全**:默认阻止危险命令
- **灵活可配置**:用户可以根据需要覆盖
- **透明可审计**:所有限制都有文档说明
## 使用示例
### 示例 1:允许 Traceroute
**文件:`command_security_override.json`**
```json
{
"allowed_commands": [
"traceroute"
]
}
```
### 示例 2:阻止特定命令
**文件:`command_security_override.json`**
```json
{
"forbidden_commands": [
"show running-config"
]
}
```
### 示例 3:允许所有命令(不推荐)
**文件:`command_security_override.json`**
```json
{
"allowed_commands": [
".*"
]
}
```
**警告**:这将禁用所有安全限制,不推荐。
## 实现细节
### 配置文件
- **默认配置**:`gns3server/agent/gns3_copilot/config/command_security.json`
- **用户覆盖**:`command_security_override.json`(用户创建)
### 组件
1. **配置加载器** (`utils/command_security_config.py`)
- 加载默认和用户配置
- 合并允许/禁止模式
- 提供编译后的正则表达式模式
2. **命令验证器** (`utils/command_validator.py`)
- 根据安全规则验证命令
- 返回清晰的错误消息
- 集成到现有工具中
### 集成位置
安全验证已集成到:
- **诊断工具** (`display_tools_nornir.py`)
- **配置工具** (`config_tools_nornir.py`)
- **VPCS 工具** (`vpcs_tools_telnetlib3.py`)
## 故障排除
### 命令被意外阻止
如果看到错误消息 `"Command not allowed: traceroute"`:
1. 检查命令是否在默认禁止列表中
2. 创建覆盖文件,在 `allowed_commands` 中添加该命令
3. 重启 GNS3 服务器
### 覆盖文件不工作
1. 验证文件位置是否正确
2. 检查 JSON 语法是否有效
3. 检查服务器日志中的加载错误
## 安全考虑
### 为什么禁止这些命令?
- **Traceroute**:生成大量数据包,可能触发 IDS/IPS
- **泛洪 ping**:可能对实验室设备进行 DoS 攻击
- **调试命令**:可能崩溃或 destabilize 生产设备
- **诊断工具中的配置命令**:防止意外的配置更改
### 最佳实践
1. **教学环境**:使用默认配置(安全)
2. **个人实验室**:考虑你真正需要什么
3. **类似生产环境**:保持限制启用
4. **始终审查**:在允许命令之前了解为什么它被阻止
## 未来增强
可能的未来改进(尚未实现):
- [ ] 每项目配置文件
- [ ] 通过 GNS3 Web UI 配置
- [ ] 被阻止命令的审计日志
- [ ] 可配置的安全预设(strict/standard/loose)
- [ ] 基于时间的限制(例如,上课时间禁止危险命令)
## 反馈和问题
如果你:
- 发现默认应该阻止的命令
- 由于合法使用场景需要允许命令
- 有改进安全的建议
请在 GitHub 上提交 issue:https://github.com/yueguobin/gns3-copilot/issues
## 相关文档
- [工具实现](../gns3-copilot/tools_v2/README.md)
- [安全策略](../../SECURITY.md)
- [贡献指南](../../CONTRIBUTING.md)
File diff suppressed because it is too large. Load diff
@@ -0,0 +1,451 @@
# TODO: 修复孤儿 Tool Calls 导致的 Checkpoint 状态不一致
## 问题描述
当 LangGraph agent 在执行过程中异常终止(如服务被强制关闭、进程崩溃等),可能导致 checkpoint 中保存了包含 `tool_calls` 的 `AIMessage`,但没有对应的 `ToolMessage`。这种状态不一致会导致后续对话恢复时出现错误。
### 术语说明
- **孤儿 tool_calls**:`AIMessage` 中包含 `tool_calls` 字段,但消息列表中没有对应的 `ToolMessage`
- **Checkpoint**:LangGraph 用于持久化对话状态的机制
- **状态不一致**:checkpoint 中的消息状态不符合预期的消息对(AIMessage + ToolMessage)
---
## 触发场景
### 场景 1:进程异常终止(主要问题)
```
执行流程:
用户消息 → llm_call → AIMessage(tool_calls) → [Checkpoint 保存]
↓
[进程崩溃/服务关闭]
↓
tool_node 未执行
↓
Checkpoint 中:
- AIMessage (有 tool_calls) ✅
- ToolMessage ❌ 缺失
```
**触发条件:**
- LLM 返回包含 tool_calls 的响应
- Checkpoint 已保存 AIMessage
- 在 tool_node 执行前服务被关闭(kill -9、Ctrl+C、崩溃等)
### 场景 2:达到最大调用次数(已处理)
当前代码通过 `recursion_limit_continue` 函数在 tool_node **执行后**检查剩余步数:
```python
def recursion_limit_continue(state: MessagesState) -> Literal["llm_call", END]:
last_message = state["messages"][-1]
if isinstance(last_message, ToolMessage):
if state["remaining_steps"] < 4:
return END
return "llm_call"
return END
```
**执行流程:**
```
remaining_steps = 5
llm_call → AIMessage(tool_calls) → remaining_steps = 4
↓
should_continue → tool_node(因为有 tool_calls)
↓
tool_node → ToolMessage → remaining_steps = 3
↓
recursion_limit_continue → remaining_steps < 4 → END ✅
```
**结论:** 场景 2 不会产生孤儿 tool_calls,因为 tool_node 总是会执行并生成 ToolMessage。
---
## 修复方案
### 核心思路
在 `stream_chat` 开始时,对于已存在的会话,检测并修复孤儿 tool_calls。
### 修复策略
**策略 A:清除 tool_calls(推荐)**
创建一个新的 `AIMessage`,内容与原消息相同,但不包含 `tool_calls` 字段。
**优点:**
- 简单、干净
- 不会影响后续对话
- 用户可以重新提问
**缺点:**
- 丢失了 LLM 原本的意图(但已经崩溃了,无法恢复)
---
## 实现代码
### 1. 添加修复方法(`agent_service.py`)
```python
async def _fix_orphan_tool_calls(self, graph, config: dict, session_id: str):
"""
检测并修复孤儿 tool_calls(AIMessage有tool_calls但没有对应ToolMessage)。
当进程在 tool_node 执行前崩溃时会产生孤儿 tool_calls。
使用 LangGraph 的 aupdate_state API,安全地创建新的 checkpoint 版本。
"""
try:
# 1. 读取当前状态
state = await graph.aget_state(config)
if not state or not state.values.get("messages"):
return
messages = state.values["messages"]
last_message = messages[-1]
# 2. 检测孤儿 tool_calls
if not (hasattr(last_message, "tool_calls") and last_message.tool_calls):
return
# 检查是否有对应的 ToolMessage
has_tool_message = any(isinstance(m, ToolMessage) for m in messages)
if has_tool_message:
return
log.warning("检测到孤儿 tool_calls: session=%s, 将清除", session_id)
# 3. 创建修复后的消息(不含 tool_calls)
from langchain.messages import AIMessage
fixed_message = AIMessage(
content=last_message.content,
id=getattr(last_message, "id", None)
)
# 4. 使用 LangGraph API 更新状态(创建新 checkpoint)
await graph.aupdate_state(config, {"messages": [fixed_message]})
log.info("孤儿 tool_calls 已修复: session=%s", session_id)
except Exception as e:
log.error("修复孤儿 tool_calls 失败: %s", e, exc_info=True)
```
### 2. 在 `stream_chat` 中调用(`agent_service.py`)
在获取 graph 之后、开始 stream 之前添加修复逻辑:
```python
async def stream_chat(
self,
message: str,
session_id: str,
project_id: Optional[str] = None,
user_id: Optional[str] = None,
jwt_token: Optional[str] = None,
mode: str = "text",
llm_config: Optional[Dict[str, Any]] = None,
) -> AsyncGenerator[Dict[str, Any], None]:
# ... 现有代码 ...
# Get or create chat session
repo = ChatSessionsRepository(self._checkpointer_conn)
session = await repo.get_session_by_thread(session_id)
is_new_session = session is None
if is_new_session:
# Create new session
session = await repo.create_session(...)
log.debug("Created new chat session: thread_id=%s", session_id)
# ... 设置 context variables ...
# Build config
config = {
"configurable": {
"thread_id": session_id,
"project_id": project_id,
},
"metadata": {
"user_id": user_id,
},
}
# Build inputs
inputs = {
"messages": [HumanMessage(content=message, id=str(uuid4()))],
"llm_calls": 0,
"remaining_steps": 20,
"mode": mode,
}
# Get the compiled graph
graph = await self._get_graph()
# 🔧 修复状态:对于已存在的会话,检查并修复孤儿 tool_calls
if not is_new_session:
await self._fix_orphan_tool_calls(graph, config, session_id)
log.debug("LangGraph graph obtained, starting stream")
# ... 继续现有代码 ...
```
### 3. 需要添加的 import
确保 `agent_service.py` 中有以下 import:
```python
from langchain.messages import ToolMessage # 用于检测 ToolMessage 类型
```
---
## 对 Checkpoint 数据库的影响
### LangGraph Checkpoint 机制
LangGraph 的 checkpoint 是**版本化**的,每次状态更新会创建新记录:
```
checkpoints 表结构:
- thread_id
- checkpoint_id (递增的版本号)
- checkpoint (序列化的状态数据)
- metadata
- ...
```
### 安全性分析
| 方面 | 影响 | 说明 |
|------|------|------|
| **原始数据** | 保留不变 | `aupdate_state` 创建新版本,不覆盖历史 |
| **数据库结构** | 完全兼容 | 使用 LangGraph 原生 API,不会破坏结构 |
| **并发安全** | 内置保护 | LangGraph 有锁机制处理并发访问 |
| **存储开销** | 很小 | 只增加一条 checkpoint 记录(约几 KB) |
| **可回滚性** | 支持 | 可回滚到修复前的任何版本 |
### 为什么不直接操作数据库
**❌ 危险方式:**
```python
# 直接修改数据库 - 破坏性强
await conn.execute(
"UPDATE checkpoints SET checkpoint = ? WHERE ...",
[modified_json]
)
```
**问题:**
- 可能破坏序列化格式
- 不创建新版本,覆盖历史
- 可能导致数据库锁定或损坏
- 违反 LangGraph 的设计原则
**✅ 安全方式:**
```python
# 使用 LangGraph 的 aupdate_state
await graph.aupdate_state(config, {"messages": [fixed_message]})
```
---
## 测试方法
### 方法 1:模拟崩溃测试(推荐)
利用强制关闭服务来模拟崩溃场景:
```
步骤:
1. 启动 GNS3 服务
2. 发送一个会触发 tool_calls 的消息(例如查询拓扑)
3. 观察日志,等待看到 AIMessage 返回(有 tool_calls)
4. 在 tool_node 执行完成前,强制关闭服务:
- 方式 1: kill -9 <pid>
- 方式 2: Ctrl+C (如果支持)
5. 重启 GNS3 服务
6. 使用同一个 session_id 继续对话
7. 观察日志,应该看到:
- "检测到孤儿 tool_calls: session=xxx, 将清除"
- "孤儿 tool_calls 已修复: session=xxx"
8. 验证对话可以正常进行
```
### 方法 2:单元测试
直接构造孤儿 tool_calls 状态来测试修复逻辑:
```python
# tests/test_agent_service.py
import pytest
from langchain.messages import AIMessage, HumanMessage, ToolMessage
@pytest.mark.asyncio
async def test_fix_orphan_tool_calls():
"""测试孤儿 tool_calls 修复逻辑"""
from gns3server.agent.gns3_copilot.agent_service import AgentService
# 创建测试用的 agent service
service = AgentService("/tmp/test_project")
await service._get_checkpointer()
graph = await service._get_graph()
config = {"configurable": {"thread_id": "test_session"}}
# 构造孤儿状态:先添加正常消息
await graph.aupdate_state(
config,
{
"messages": [
HumanMessage(content="测试消息", id="msg_1"),
AIMessage(
content="让我帮你查一下",
id="msg_2",
tool_calls=[{
"id": "call_123",
"name": "get_topology",
"args": {"project_id": "test"}
}]
)
# 注意:没有对应的 ToolMessage
],
"llm_calls": 1,
"remaining_steps": 20
}
)
# 调用修复逻辑
await service._fix_orphan_tool_calls(graph, config, "test_session")
# 验证修复结果
state = await graph.aget_state(config)
last_message = state.values["messages"][-1]
# 应该不再有 tool_calls
assert not hasattr(last_message, "tool_calls") or not last_message.tool_calls
assert last_message.content == "让我帮你查一下"
# 清理
await service.close()
@pytest.mark.asyncio
async def test_no_fix_when_normal():
"""测试正常状态不会被误修复"""
from gns3server.agent.gns3_copilot.agent_service import AgentService
service = AgentService("/tmp/test_project")
await service._get_checkpointer()
graph = await service._get_graph()
config = {"configurable": {"thread_id": "test_session_2"}}
# 构造正常状态:有完整的 AIMessage + ToolMessage 对
await graph.aupdate_state(
config,
{
"messages": [
HumanMessage(content="测试消息", id="msg_1"),
AIMessage(
content="让我帮你查一下",
id="msg_2",
tool_calls=[{
"id": "call_123",
"name": "get_topology",
"args": {"project_id": "test"}
}]
),
ToolMessage(
content="拓扑信息:...",
tool_call_id="call_123",
name="get_topology",
id="msg_3"
)
],
"llm_calls": 1,
"remaining_steps": 20
}
)
# 记录原始消息数量
state_before = await graph.aget_state(config)
msg_count_before = len(state_before.values["messages"])
# 调用修复逻辑
await service._fix_orphan_tool_calls(graph, config, "test_session_2")
# 验证状态未改变
state_after = await graph.aget_state(config)
msg_count_after = len(state_after.values["messages"])
assert msg_count_before == msg_count_after # 不应该添加新消息
last_message = state_after.values["messages"][-1]
assert isinstance(last_message, ToolMessage) # 最后一条还是 ToolMessage
# 清理
await service.close()
```
### 方法 3:增强日志和监控
即使无法主动触发,也可以在生产环境验证修复逻辑是否生效:
```python
# 在 _fix_orphan_tool_calls 中添加详细日志
log.warning("检测到孤儿 tool_calls: session=%s", session_id)
log.info("原始消息: tool_calls=%d, content=%s",
len(last_message.tool_calls),
last_message.content[:100])
log.info("修复后: tool_calls=%d",
len(fixed_message.tool_calls) if hasattr(fixed_message, "tool_calls") else 0)
```
---
## 文件修改清单
### 需要修改的文件
1. **`gns3server/agent/gns3_copilot/agent_service.py`**
- 添加 `_fix_orphan_tool_calls` 方法
- 在 `stream_chat` 方法中调用修复逻辑
### 需要添加的测试文件(可选)
2. **`tests/test_agent_service.py`**(新建或添加到现有测试文件)
- `test_fix_orphan_tool_calls()` - 测试孤儿 tool_calls 修复
- `test_no_fix_when_normal()` - 测试正常状态不被误修复
---
## 实施步骤
1. ✅ 创建待办文档(当前文档)
2. ⬜ 在 `agent_service.py` 中添加 `_fix_orphan_tool_calls` 方法
3. ⬜ 在 `stream_chat` 中调用修复逻辑
4. ⬜ 使用模拟崩溃方法测试修复效果
5. ⬜ 添加单元测试(可选)
6. ⬜ 更新相关文档(如有必要)
---
## 相关代码文件
- **主要修改文件**: `gns3server/agent/gns3_copilot/agent_service.py`
- **相关文件**: `gns3server/agent/gns3_copilot/agent/gns3_copilot.py`
- **测试文件**: `tests/test_agent_service.py` (待创建)
---
## 参考文档
- [LangGraph Checkpointer 文档](https://langchain-ai.github.io/langgraph/concepts/low_level/#checkpointer)
- [LangGraph State Management](https://langchain-ai.github.io/langgraph/concepts/low_level/#state)
- [GNS3-Copilot AI Chat API 设计](../ai-chat-api-design.md)
@@ -0,0 +1,215 @@
# GNS3 Copilot Tool Response Format Standard
## 概述
本文档定义了 GNS3 Copilot 工具的标准响应格式,确保所有工具返回统一的数据结构,便于前端处理和美化显示。
## 标准响应格式
### 顶层结构
所有工具应返回以下标准格式:
```python
{
"success": bool, # 整体操作是否成功
"total": int, # 总操作数量
"successful": int, # 成功数量
"failed": int, # 失败数量
"data": list[dict], # 详细结果列表
"error": str, # 全局错误信息(可选,操作完全失败时)
"metadata": dict # 元数据(可选)
}
```
**字段说明**:
| 字段 | 类型 | 必需 | 说明 |
|------|------|------|------|
| `success` | `bool` | 是 | 整体操作是否成功(`failed == 0` 时为 `True`) |
| `total` | `int` | 是 | 处理的项目总数 |
| `successful` | `int` | 是 | 成功的项目数量 |
| `failed` | `int` | 是 | 失败的项目数量 |
| `data` | `list[dict]` | 是 | 每个项目的详细结果 |
| `error` | `str` | 否 | 全局错误消息(当整个操作失败时) |
| `metadata` | `dict` | 否 | 元数据(时间戳、执行时间等) |
### 单个项目格式
`data` 数组中的每个项目应遵循以下格式:
```python
{
"id": str, # 设备/节点/链接 ID
"name": str, # 人类可读的名称
"status": "success" | "failed", # 项目状态
"result": str, # 成功时的结果或输出
"error": str # 失败时的错误信息
}
```
**字段说明**:
| 字段 | 类型 | 必需 | 说明 |
|------|------|------|------|
| `id` | `str` | 是 | 设备/节点/链接的唯一标识符 |
| `name` | `str` | 是 | 人类可读的名称 |
| `status` | `str` | 是 | `"success"` 或 `"failed"` |
| `result` | `str` | 条件 | 状态为 `success` 时的输出 |
| `error` | `str` | 条件 | 状态为 `failed` 时的错误信息 |
## 示例
### 成功响应示例
```python
# 执行多个设备的显示命令
{
"success": True,
"total": 3,
"successful": 2,
"failed": 1,
"data": [
{
"id": "R1",
"name": "Router1",
"status": "success",
"result": "Cisco IOS Software...\nRouter1# show version\n..."
},
{
"id": "R2",
"name": "Router2",
"status": "success",
"result": "Cisco IOS Software...\nRouter2# show version\n..."
},
{
"id": "R3",
"name": "Router3",
"status": "failed",
"error": "Connection refused"
}
],
"metadata": {
"tool_name": "execute_multiple_device_commands",
"execution_time": 5.2
}
}
```
### 完全失败示例
```python
# 整个操作失败(如参数错误)
{
"success": False,
"total": 0,
"successful": 0,
"failed": 0,
"data": [],
"error": "Invalid project_id format",
"metadata": {
"tool_name": "execute_multiple_device_commands"
}
}
```
### 单个设备操作示例
```python
# 操作单个设备
{
"success": True,
"total": 1,
"successful": 1,
"failed": 0,
"data": [
{
"id": "PC1",
"name": "VPCS-1",
"status": "success",
"result": "IP configuration updated: 192.168.1.10/24"
}
],
"metadata": {}
}
```
## 使用标准化函数
在 `gns3server.agent.gns3_copilot.utils` 模块中提供了 `normalize_tool_response` 函数,用于将各种格式转换为标准格式:
```python
from gns3server.agent.gns3_copilot.utils import normalize_tool_response
# 标准化工具响应
normalized = normalize_tool_response(raw_response, tool_name="my_tool")
```
该函数支持:
- 列表格式(`[{...}, {...}]`)
- 字典格式(`{"nodes": [...]}`)
- 字符串格式(自动解析 JSON/Python literal)
- 混合格式(兼容旧工具)
## 兼容性
### 向后兼容
`normalize_tool_response` 函数设计为向后兼容,可以处理现有工具的各种格式:
- `status` / `error` 字段
- `output` / `result` 字段
- `device_name` / `name` 字段
- `total_nodes` / `total` 字段
### 推荐的迁移策略
1. **新工具**:直接返回标准格式
2. **现有工具**:保持不变,使用 `normalize_tool_response` 标准化
3. **前端**:依赖标准格式处理显示
## 前端集成建议
### 渲染逻辑
```javascript
function renderToolResponse(response) {
if (!response.success) {
// 显示全局错误
showError(response.error);
return;
}
// 显示统计摘要
showSummary(response.total, response.successful, response.failed);
// 渲染每个项目
response.data.forEach(item => {
if (item.status === 'success') {
showSuccess(item.name, item.result);
} else {
showError(item.name, item.error);
}
});
}
```
### 状态图标
| 状态 | 图标建议 | 颜色 |
|------|----------|------|
| `success` | ✓ 绿色 | 绿色 |
| `failed` | ✗ 红色 | 红色 |
| `unknown` | ? 灰色 | 灰色 |
## 版本控制
当前标准版本:`v1.0`
格式变更时,应更新 `metadata.version` 字段,前端据此适配。
## 参考
- 实现:`gns3server/agent/gns3_copilot/utils/parse_tool_content.py`
- 消息转换:`gns3server/agent/gns3_copilot/utils/message_converters.py`
- 工具示例:`gns3server/agent/gns3_copilot/tools_v2/`