From 03ab9cdf6c07adb783fb9e3084490311c85e6024 Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Wed, 4 Mar 2026 21:58:23 +0800 Subject: [PATCH] feat(chat-api): refactor design document with concise architecture overview - Replace detailed implementation plan with high-level architecture design - Focus on core features: project isolation, streaming responses, session management - Remove FlowNet-Lab reference and implementation specifics - Streamline document from 1172 to 483 lines for better maintainability --- docs/ai-chat-api-design.md | 1513 +++++------------ .../agent/gns3_copilot/agent_service.py | 158 ++ .../gns3_copilot/chat_sessions_repository.py | 365 ++++ gns3server/api/routes/controller/chat.py | 77 +- gns3server/schemas/__init__.py | 3 +- gns3server/schemas/controller/chat.py | 20 +- 6 files changed, 1017 insertions(+), 1119 deletions(-) create mode 100644 gns3server/agent/gns3_copilot/chat_sessions_repository.py diff --git a/docs/ai-chat-api-design.md b/docs/ai-chat-api-design.md index 782caf1e4..25bc68f8a 100644 --- a/docs/ai-chat-api-design.md +++ b/docs/ai-chat-api-design.md @@ -1,1172 +1,483 @@ -# GNS3 Copilot Agent Chat API 实现方案 +# GNS3 Copilot Agent Chat API 设计文档 ## 概述 -本文档描述了如何在 GNS3 Server 中实现 AI Chat API，使客户端能够通过 RESTful API 与 GNS3 Copilot Agent 进行交互。 +本文档描述 GNS3 Copilot Chat API 的架构设计和实现方案。该 API 使客户端能够通过 RESTful 接口与 GNS3 Copilot Agent 进行交互，提供流式对话、会话管理、项目拓扑查询等功能。 -## 背景 +## 核心特性 -### 现有组件 - -- **GNS3 Copilot Agent**: 位于 `gns3server/agent/gns3_copilot/`，使用 LangGraph 实现的网络自动化助手 -- **LLM 配置管理**: 已有 `llm_model_configs` 系统，支持用户/用户组的 LLM 配置 -- **API 框架**: 使用 FastAPI，已有的路由结构在 `gns3server/api/routes/controller/` - -### 参考实现 - -FlowNet-Lab 项目 (`/home/yueguobin/myCode/GNS3/FlowNet-Lab`) 已有完整的 Chat API 实现，可作为参考： - -- Backend: `backend/api/v1/chat.py` -- Agent Service: `backend/core/agent.py` +- **项目级隔离**：每个 GNS3 项目拥有独立的 Agent 实例和会话存储 +- **流式响应**：使用 Server-Sent Events (SSE) 实现实时流式输出 +- **会话管理**：支持会话列表、重命名、删除、历史记录查询 +- **统计追踪**：自动记录消息数量、LLM 调用次数、Token 使用量 +- **用户隔离**：每个用户拥有独立的 LLM 配置和会话空间 ## 架构设计 ### 整体架构 ``` -┌─────────────────────────────────────────────────────────────────┐ -│ Frontend (Client) │ -│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────────┐│ -│ │ ChatInput │ │ MessageList │ │ ConversationSidebar ││ -│ └──────┬──────┘ └──────┬──────┘ └───────────┬─────────────┘│ -│ │ │ │ │ -│ └─────────────────┼──────────────────────┘ │ -│ ▼ │ -│ chatService.ts │ -│ (SSE Streaming) │ -└─────────────────────────────┬───────────────────────────────────┘ - │ - HTTP POST /api/v1/chat/stream - │ - ▼ -┌─────────────────────────────────────────────────────────────────┐ -│ GNS3 Server (Backend) │ -│ │ -│ ┌─────────────────────────────────────────────────────────┐ │ -│ │ FastAPI Application │ │ -│ │ ┌─────────────────────────────────────────────────┐ │ │ -│ │ │ Chat API Routes │ │ │ -│ │ │ POST /v3/chat/stream │ │ │ -│ │ │ GET /v3/chat/history/{session_id} │ │ │ -│ │ │ GET /v3/chat/sessions │ │ │ -│ │ │ POST /v3/chat/sessions │ │ │ -│ │ │ DELETE /v3/chat/sessions/{session_id} │ │ │ -│ │ └─────────────────────────────────────────────────┘ │ │ -│ └──────────────────────────┬───────────────────────────────┘ │ -│ │ │ -│ ▼ │ -│ ┌─────────────────────────────────────────────────────────┐ │ -│ │ ProjectAgentManager │ │ -│ │ (管理每个项目的 Agent Service 实例) │ │ -│ └──────────────────────────┬───────────────────────────────┘ │ -│ │ │ -│ ┌──────────────────────────┼───────────────────────────────┐ │ -│ │ ▼ │ │ -│ │ ┌─────────────────────────────────────────────┐ │ │ -│ │ │ AgentService │ │ │ -│ │ │ - SQLiteSaver (项目级 checkpoint) │ │ │ -│ │ │ - LangGraph Agent │ │ │ -│ │ └─────────────────────┬───────────────────────┘ │ │ -│ │ │ │ │ -│ │ ▼ │ │ -│ │ ┌─────────────────────────────────────────────┐ │ │ -│ │ │ LangGraph Agent (gns3_copilot) │ │ │ -│ │ │ llm_call → should_continue → tool_node │ │ │ -│ │ └─────────────────────────────────────────────┘ │ │ -│ │ │ │ -│ └───────────────────────────────────────────────────────┘ │ -│ │ -└─────────────────────────────────────────────────────────────────┘ - │ - ▼ -┌─────────────────────────────────────────────────────────────────┐ -│ GNS3 Project Directory │ -│ │ -│ {project.path}/ │ -│ ├── .gns3-copilot/ │ -│ │ └── checkpoint.db (SQLite - LangGraph 状态存储) │ -│ ├── project-files/ │ -│ │ ├── nodes/ │ -│ │ └── captures/ │ -│ └── project.gns3 │ -│ │ -└─────────────────────────────────────────────────────────────────┘ +Frontend (Web UI) + │ + │ SSE Streaming + ▼ +FastAPI Chat API Routes + │ + │ Project-level Agent Management + ▼ +AgentService (per project) + │ + ├─ SQLite Checkpointer (project_dir/gns3-copilot/) + │ ├─ checkpoints table (LangGraph state) + │ └─ chat_sessions table (session metadata) + │ + └─ LangGraph Agent + ├─ llm_call node + ├─ should_continue node + └─ tool_node (GNS3 tools) ``` ### 项目级 Checkpoint 设计 -**核心思路**: 每个 GNS3 项目有独立的 checkpoint 数据库，实现项目级别的会话隔离。 +每个 GNS3 项目在项目目录下创建 `gns3-copilot/copilot_checkpoints.db` SQLite 数据库，包含两张表： +1. **checkpoints 表**（LangGraph 自动管理）：存储 Agent 的对话状态和记忆 +2. **chat_sessions 表**（自定义）：存储会话元数据和统计信息 + +**目录结构**： ``` -project_path/.gns3-copilot/checkpoint.db +{project.path}/ +├── gns3-copilot/ +│ └── copilot_checkpoints.db +├── project-files/ +└── project.gns3 ``` -使用 LangGraph 的 `AsyncSqliteSaver` 作为 checkpointer（推荐异步方式）。 +**设计优势**： +- 项目删除时自动清理所有相关数据 +- 实现项目级别的会话隔离 +- 便于备份和迁移 -### 生产级实现（推荐） +## 用户认证信息传递 -以下实现包含连接管理、项目切换、资源清理等完整功能： +### 背景需求 -```python -import os -import logging -from typing import Optional -import aiosqlite -from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver +GNS3 Copilot Agent 需要以下信息才能正常工作： +1. **user_id**：获取用户专属的 LLM 配置 +2. **jwt_token**：调用 GNS3 API 时进行身份验证 +3. **llm_config**：包含 provider、model、api_key 等配置 -log = logging.getLogger(__name__) +### ContextVars 方案 +使用 Python 的 `contextvars.ContextVar` 在请求作用域内传递数据，避免敏感信息持久化到 checkpoint。 -class CheckpointerManager: - """Checkpointer 管理器（项目级）""" - - def __init__(self, controller): - """ - Args: - controller: GNS3 Controller 实例 - """ - self.controller = controller - self._checkpointer: Optional[AsyncSqliteSaver] = None - self._checkpointer_conn: Optional[aiosqlite.Connection] = None - self._project_checkpoint_path: Optional[str] = None - - async def _get_checkpointer(self, project_id: str) -> AsyncSqliteSaver: - """ - 获取或创建 SQLite checkpointer for a specific project. - - Args: - project_id: GNS3 project ID - - Returns: - AsyncSqliteSaver instance - """ - log.debug("Getting checkpointer for project %s", project_id) - - # Get the project to find its directory - project = self.controller.get_project(project_id) - if not project: - log.error("Project %s not found in controller", project_id) - raise ValueError(f"Project {project_id} not found") - - # Create checkpoint file in the project directory - checkpointer_path = os.path.join(project.path, ".gns3-copilot", "copilot_checkpoints.db") - os.makedirs(os.path.dirname(checkpointer_path), exist_ok=True) - log.debug("Checkpoint path: %s", checkpointer_path) - - # Store the path for reference - self._project_checkpoint_path = checkpointer_path - - # Check if we already created a checkpointer for this project - if self._checkpointer and self._project_checkpoint_path == checkpointer_path: - log.debug("Reusing existing checkpointer") - return self._checkpointer - - # Create new checkpointer using AsyncSqliteSaver - log.debug("Creating new async checkpointer at %s", checkpointer_path) - - # Close existing connection if switching projects - if self._checkpointer_conn: - try: - await self._checkpointer_conn.close() - log.debug("Closed previous checkpointer connection") - except Exception as e: - log.warning("Error closing old checkpointer connection: %s", e) - - # Create new connection - conn = await aiosqlite.connect(checkpointer_path) - # Enable WAL mode for better concurrent performance - await conn.execute("PRAGMA journal_mode=WAL;") - self._checkpointer_conn = conn # Save connection reference to prevent GC - self._checkpointer = AsyncSqliteSaver(conn) - - # CRITICAL: Initialize database schema - await self._checkpointer.setup() - - log.info("Project async checkpointer created and initialized at %s", checkpointer_path) - - return self._checkpointer - - async def close(self): - """关闭 checkpointer 连接""" - if self._checkpointer_conn: - try: - await self._checkpointer_conn.close() - log.debug("Checkpointer connection closed") - except Exception as e: - log.warning("Error closing checkpointer connection: %s", e) - finally: - self._checkpointer_conn = None - self._checkpointer = None +**数据流**： ``` +1. API 层获取用户信息 + ├─ 从 FastAPI get_current_active_user 获取 user_id + ├─ 从 Authorization header 提取 jwt_token + └─ 从数据库查询 LLM 配置（已解密 API key） -### 关键设计要点 +2. 设置 ContextVars（内存临时存储） + ├─ set_current_jwt_token(jwt_token) + └─ set_current_llm_config(llm_config) -| 功能 | 说明 | -|------|------| -| **连接复用** | 同项目复用已有 checkpointer，避免重复创建 | -| **项目切换** | 切换项目时自动关闭旧连接，防止连接泄漏 | -| **GC 防护** | 保存 `self._checkpointer_conn` 引用，防止连接被垃圾回收 | -| **Schema 初始化** | 调用 `await self._checkpointer.setup()` 初始化数据库表结构 | -| **WAL 模式** | 启用 WAL 模式提升并发写入性能 | -| **资源清理** | `close()` 方法确保连接正确关闭 | -| **日志记录** | 完整的调试日志，便于问题排查 | - -### 注意事项 -- 需要安装 `aiosqlite` 包 -- 必须在异步环境中使用 -- 切换项目时自动关闭旧连接 -- 应用关闭时应调用 `close()` 清理资源 - -**依赖**: -``` -langgraph-checkpoint-sqlite>=3.0.1 -aiosqlite -``` - -## 文件结构 - -需要创建/修改以下文件： - -``` -gns3server/ -├── agent/gns3_copilot/ -│ ├── agent_service.py # [新建] AgentService 封装 -│ └── project_agent_manager.py # [新建] 项目级 Agent 管理器 -├── api/routes/controller/ -│ ├── chat.py # [新建] Chat API 路由 -│ └── __init__.py # [修改] 注册 chat router -├── schemas/controller/ -│ └── chat.py # [新建] Chat Request/Response 模型 -└── docs/ - └── ai-chat-api-design.md # [本文档] -``` - -## 用户认证信息传递机制 - -### 背景 - -GNS3 Copilot Agent 需要获取用户的 LLM 配置信息，这需要： -1. **user_id**: 用于从数据库获取用户专属的 LLM 配置 -2. **jwt_token**: 用于调用 GNS3 API 时进行身份验证 - -`model_factory.py` 已支持这些参数： - -```python -def create_base_model( - user_id: Optional[UUID] = None, - jwt_token: Optional[str] = None, - llm_config: Optional[dict[str, Any]] = None, -) -> Any: - # 优先级： - # 1. 提供的 llm_config 字典 - # 2. 从数据库获取 (需要 user_id 和 jwt_token) - # 3. 环境变量 (回退) -``` - -### 传递机制 - -使用 **LangGraph 的 config 参数** 传递用户信息： - -``` -┌─────────────────────────────────────────────────────────────┐ -│ LangGraph Config │ -│ { │ -│ "configurable": { │ -│ "thread_id": "session-xxx", # 会话 ID │ -│ "user_id": "user-uuid", # 用户 ID │ -│ "jwt_token": "eyJxxx..." # JWT Token │ -│ } │ -│ } │ -└─────────────────────────────────────────────────────────────┘ -``` - -### 实现步骤 - -#### 1. 修改 gns3_copilot.py 中的 llm_call 函数 - -```python -# gns3server/agent/gns3_copilot/agent/gns3_copilot.py - -def llm_call(state: dict, config: dict = None): - """LLM decides whether to call a tool or not""" - - # 从 config 中获取用户信息 - configurable = config.get("configurable", {}) if config else {} - user_id = configurable.get("user_id") - jwt_token = configurable.get("jwt_token") - - # ... 原有逻辑 ... - - # 传递 user_id 和 jwt_token 给 model factory - model_with_tools = create_base_model_with_tools( - tools, - user_id=user_id, - jwt_token=jwt_token - ) - - return { - "messages": [model_with_tools.invoke(full_messages)], - "llm_calls": state.get("llm_calls", 0) + 1, - "topology_info": topology_info, - } -``` - -#### 2. 修改 generate_title 函数（同样需要传递） - -```python -def generate_title(state: MessagesState, config: dict = None) -> dict: - """Generate conversation title""" - - configurable = config.get("configurable", {}) if config else {} - user_id = configurable.get("user_id") - jwt_token = configurable.get("jwt_token") - - # ... 使用 user_id 和 jwt_token ... -``` - -#### 3. 在 AgentService.stream_chat 中构建 config - -```python -# gns3server/agent/gns3_copilot/agent_service.py - -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" -) -> AsyncGenerator[Dict[str, Any], None]: - - # 构建包含用户信息的 config - config = { - "configurable": { - "thread_id": session_id, - "user_id": user_id, - "jwt_token": jwt_token, - } - } - - # 流式处理时传递 config - async for event in self.graph.astream_events(inputs, config=config, version="v2"): - # ... -``` - -#### 4. 在 API 路由中获取并传递用户信息 - -```python -# gns3server/api/routes/controller/chat.py - -from fastapi import Request - -@router.post("/stream") -async def stream_chat( - request: ChatRequest, - current_user = Depends(get_current_active_user) -): - # 获取 JWT token - auth_header = request.headers.get("Authorization") - jwt_token = auth_header.replace("Bearer ", "") if auth_header else None - - # 获取 user_id - user_id = str(current_user.user_id) - - # 传递用户信息给 AgentService - agent_service = agent_manager.get_agent(request.project_id, project_path) - - async def generate(): - async for chunk in agent_service.stream_chat( - message=request.message, - session_id=session_id, - project_id=request.project_id, - user_id=user_id, # 传递 user_id - jwt_token=jwt_token, # 传递 jwt_token - mode=request.mode - ): - # ... -``` - -### 完整数据流 - -``` -1. 前端发起请求 (带 Authorization: Bearer ) - -2. FastAPI get_current_active_user 验证并返回 User 对象 - -3. 从 Header 获取 JWT token - -4. 构建 LangGraph config: +3. 构建安全的 LangGraph config（仅包含非敏感标识符） { "configurable": { "thread_id": session_id, - "user_id": user_id, - "jwt_token": jwt_token + "project_id": project_id + }, + "metadata": { + "user_id": user_id } } -5. AgentService.stream_chat() 传递 config 给 astream_events() - -6. llm_call() / generate_title() 从 config 获取用户信息 - -7. create_base_model() 使用 user_id 从数据库获取 LLM 配置 +4. LLM 节点从 ContextVars 获取配置 + ├─ get_current_jwt_token() + └─ get_current_llm_config() ``` -## 消息格式定义 (参考 FlowNet-Lab) +**方案优势**： +- 敏感数据（JWT token、API key）仅存储在内存中 +- 请求结束后自动清理，不会持久化到数据库 +- 避免序列化/反序列化开销 +- 实现请求级别的数据隔离 -### 概述 +## 会话管理 -Chat API 使用 Server-Sent Events (SSE) 进行流式传输，消息格式分为： -- **请求格式** (ChatRequest) -- **响应格式** (ChatResponse) -- **消息类型** (Message Types) +### chat_sessions 表结构 -### 1. 请求格式 (ChatRequest) - -```python -class ChatRequest(BaseModel): - """Chat 请求模型""" - message: str # 用户消息内容 - session_id: Optional[str] = None # 会话 ID (可选，不提供则自动创建) - project_id: str # GNS3 项目 ID - stream: bool = True # 是否启用流式响应 - temperature: Optional[float] = None # LLM 温度参数 - mode: Literal["text"] = "text" # 交互模式 -``` - -### 2. 响应格式 (ChatResponse) - -```python -class OpenAIToolCall(BaseModel): - """工具调用信息 (OpenAI 兼容格式)""" - id: str - type: Literal["function"] = "function" - function: Dict[str, Any] # {"name": "...", "arguments": {...}} - - -class ChatResponse(BaseModel): - """流式响应模型""" - type: Literal[ - "content", # AI 文本内容 - "tool_call", # 工具调用请求 - "tool_start", # 工具开始执行 - "tool_end", # 工具执行完成 - "error", # 错误信息 - "done", # 流结束 - "heartbeat" # 心跳保活 - ] - content: Optional[str] = None # 文本内容 (type=content) - message_id: Optional[str] = None # 消息 ID - tool_call: Optional[OpenAIToolCall] = None # 工具调用 (type=tool_call) - tool_name: Optional[str] = None # 工具名称 (type=tool_start/end) - tool_output: Optional[str] = None # 工具输出 (type=tool_end) - error: Optional[str] = None # 错误信息 (type=error) - session_id: Optional[str] = None # 会话 ID (type=heartbeat/done) -``` - -### 3. SSE 消息示例 - -#### 3.1 文本内容 (content) - -```json -{"type": "content", "content": "Hello! How can I help you with your network today?"} -``` - -#### 3.2 工具调用 (tool_call) - -```json -{ - "type": "tool_call", - "tool_call": { - "id": "call_abc123", - "type": "function", - "function": { - "name": "GNS3TopologyTool", - "arguments": {"project_id": "550e8400-e29b-41d4-a716-446655440000"} - } - } -} -``` - -#### 3.3 工具开始执行 (tool_start) - -```json -{"type": "tool_start", "tool_name": "GNS3TopologyTool", "session_id": "xxx"} -``` - -#### 3.4 工具执行完成 (tool_end) - -```json -{ - "type": "tool_end", - "tool_name": "GNS3TopologyTool", - "tool_output": "{\"nodes\": [{\"name\": \"Router1\", ...}], \"links\": [...]}" -} -``` - -#### 3.5 错误 (error) - -```json -{"type": "error", "error": "Session not found", "session_id": "xxx"} -``` - -#### 3.6 完成 (done) - -```json -{"type": "done", "session_id": "xxx"} -``` - -#### 3.7 心跳 (heartbeat) - -```json -{"type": "heartbeat", "session_id": "xxx"} -``` - -**作用**: 保持 SSE 连接活跃，防止代理服务器/负载均衡器因超时断开连接。 - -**实现机制**: - -```python -# 使用 asyncio.wait 实现超时检测 -heartbeat_interval = 15.0 # 配置的心跳间隔（秒） - -done, pending = await asyncio.wait( - [next_event_task], - timeout=heartbeat_interval -) - -if done: - # 收到事件，正常处理 - event = next_event_task.result() - # ... -else: - # 超时 - 发送心跳，保持连接 - yield {"type": "heartbeat", "session_id": session_id} - # 继续等待下一个事件 -``` - -**配置项** (可选): - -```python -# 可通过配置控制 -heartbeat_interval = 15.0 # 心跳间隔（秒），0 表示禁用 -heartbeat_enabled = True # 是否启用 -``` - -**前端处理**: - -- 前端收到 `heartbeat` 类型消息时可以忽略 -- 主要用于维持连接，不需要渲染任何内容 - -### 4. 前端处理逻辑 - -前端 (`useChat.ts`) 根据 `type` 字段进行不同处理： - -| type | 处理逻辑 | -|------|----------| -| `content` | 追加到当前 AI 消息内容 | -| `tool_call` | 创建 tool_call 类型的消息，显示工具调用信息 | -| `tool_start` | 可选：显示工具开始执行的状态 | -| `tool_end` | 创建 tool_result 类型的消息，显示工具执行结果 | -| `error` | 显示错误信息 | -| `done` | 标记流结束 | -| `heartbeat` | 忽略（保活信号） | - -### 5. 会话历史格式 (ConversationHistory) - -```python -class OpenAIMessage(BaseModel): - """消息模型 (用于历史记录)""" - id: str - role: Literal["user", "assistant", "system", "tool"] - content: str - name: Optional[str] = None # 工具消息的名称 - tool_call_id: Optional[str] = None # 工具消息关联的 tool_call ID - tool_calls: Optional[List[OpenAIToolCall]] = None # 助手消息的工具调用 - metadata: Dict[str, Any] = {} - created_at: str - - -class ConversationHistory(BaseModel): - """会话历史模型""" - thread_id: str - title: str - messages: List[OpenAIMessage] - created_at: Optional[str] = None - updated_at: Optional[str] = None - llm_calls: int = 0 -``` - -## 核心实现 - -### 1. Chat Schemas - -**文件**: `gns3server/schemas/controller/chat.py` - -```python -from pydantic import BaseModel -from typing import Optional, List, Dict, Any, Literal - -class ChatRequest(BaseModel): - """Chat 请求模型""" - message: str - session_id: Optional[str] = None - project_id: str - stream: bool = True - temperature: Optional[float] = None - mode: Literal["text"] = "text" - - -class ChatResponse(BaseModel): - """Chat 流式响应模型""" - type: Literal["content", "tool_call", "tool_start", "tool_end", "error", "done", "heartbeat"] - content: Optional[str] = None - tool_call: Optional[Dict[str, Any]] = None - tool_name: Optional[str] = None - tool_output: Optional[str] = None - error: Optional[str] = None - session_id: Optional[str] = None - - -class ConversationHistory(BaseModel): - """会话历史模型""" - thread_id: str - title: str - messages: List[Dict[str, Any]] - created_at: Optional[str] = None - updated_at: Optional[str] = None - - -class ChatSession(BaseModel): - """会话模型""" - session_id: str - title: str - project_id: Optional[str] = None - created_at: Optional[str] = None - updated_at: Optional[str] = None -``` - -### 2. AgentService - -**文件**: `gns3server/agent/gns3_copilot/agent_service.py` - -```python -""" -GNS3 Copilot Agent Service -为每个项目提供独立的 Agent 实例，使用项目目录的 SQLite 作为 checkpoint（异步版本） -""" - -import os -import uuid -import logging -from typing import AsyncGenerator, Dict, List, Any, Optional -from langchain_core.messages import HumanMessage -from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver -import aiosqlite - -from gns3_copilot.agent.gns3_copilot import agent_builder - -log = logging.getLogger(__name__) - - -class AgentService: - """项目级 Agent Service（异步版本）""" - - def __init__(self, project_path: str, controller=None): - """ - Args: - project_path: GNS3 项目目录路径 - controller: GNS3 Controller 实例（可选，用于获取项目信息） - """ - self.project_path = project_path - self.controller = controller - - # 创建 checkpoint 目录 - self.checkpoint_dir = os.path.join(project_path, ".gns3-copilot") - os.makedirs(self.checkpoint_dir, exist_ok=True) - - self._checkpointer: Optional[AsyncSqliteSaver] = None - self._checkpointer_conn: Optional[aiosqlite.Connection] = None - self._project_checkpoint_path: Optional[str] = None - self._graph = None - self._model_with_tools = None - - async def _get_checkpointer(self) -> AsyncSqliteSaver: - """ - 获取或创建 SQLite checkpointer。 - - Returns: - AsyncSqliteSaver instance - """ - log.debug("Getting checkpointer for project at %s", self.project_path) - - # Create checkpoint file in the project directory - checkpointer_path = os.path.join(self.checkpoint_dir, "copilot_checkpoints.db") - log.debug("Checkpoint path: %s", checkpointer_path) - - # Store the path for reference - self._project_checkpoint_path = checkpointer_path - - # Check if we already created a checkpointer for this project - if self._checkpointer and self._project_checkpoint_path == checkpointer_path: - log.debug("Reusing existing checkpointer") - return self._checkpointer - - # Create new checkpointer using AsyncSqliteSaver - log.debug("Creating new async checkpointer at %s", checkpointer_path) - - # Close existing connection if switching projects - if self._checkpointer_conn: - try: - await self._checkpointer_conn.close() - log.debug("Closed previous checkpointer connection") - except Exception as e: - log.warning("Error closing old checkpointer connection: %s", e) - - # Create new connection - conn = await aiosqlite.connect(checkpointer_path) - # Enable WAL mode for better concurrent performance - await conn.execute("PRAGMA journal_mode=WAL;") - self._checkpointer_conn = conn # Save connection reference to prevent GC - self._checkpointer = AsyncSqliteSaver(conn) - - # CRITICAL: Initialize database schema - await self._checkpointer.setup() - - log.info("Project async checkpointer created and initialized at %s", checkpointer_path) - - return self._checkpointer - - def _get_model_with_tools(self): - """ - Get model with tools bound. - """ - if self._model_with_tools is None: - log.debug("Binding tools to model...") - from gns3server.agent.gns3_copilot.model_factory import create_base_model_with_tools - from gns3server.agent.gns3_copilot.tools import get_tools - - model = self._create_model() - tools = get_tools() - self._model_with_tools = model.bind_tools(tools) - log.info("Model bound with %d tools", len(tools)) - return self._model_with_tools - - def _create_model(self): - """创建基础模型""" - from gns3server.agent.gns3_copilot.model_factory import create_base_model - return create_base_model() - - @property - async def checkpointer(self) -> AsyncSqliteSaver: - """获取 SQLite checkpointer""" - if self._checkpointer is None: - return await self._get_checkpointer() - return self._checkpointer - - @property - async def graph(self): - """获取或编译 LangGraph""" - if self._graph is None: - checkpointer = await self.checkpointer - self._graph = agent_builder.compile(checkpointer=checkpointer) - return self._graph - - async def close(self): - """关闭连接""" - if self._checkpointer_conn: - try: - await self._checkpointer_conn.close() - log.debug("Checkpointer connection closed") - except Exception as e: - log.warning("Error closing checkpointer connection: %s", e) - finally: - self._checkpointer_conn = None - self._checkpointer = None - self._graph = None - self._model_with_tools = None - - 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" - ) -> AsyncGenerator[Dict[str, Any], None]: - """流式处理 chat 请求""" - - # 构建 config，包含用户认证信息 - config = { - "configurable": { - "thread_id": session_id, - "user_id": user_id, - "jwt_token": jwt_token, - } - } - - # 获取项目信息 - project_info = None - if project_id: - try: - from gns3server.controller import Controller - controller = Controller.instance() - project = controller.projects.get(project_id) - if project: - project_info = ( - project.name, - project.id, - len(project.nodes), - len(project.links), - project.status - ) - except Exception: - pass - - # 构建输入 - inputs = { - "messages": [HumanMessage(content=message)], - "llm_calls": 0, - "remaining_steps": 20, - "mode": mode, - } - - if project_info: - inputs["selected_project"] = project_info - - # 流式处理 - try: - async for event in self.graph.astream_events(inputs, config=config, version="v2"): - chunk = self._convert_event_to_chunk(event) - if chunk: - yield chunk - - yield {"type": "done", "session_id": session_id} - - except Exception as e: - yield {"type": "error", "error": str(e), "session_id": session_id} - - def _convert_event_to_chunk(self, event: Dict[str, Any]) -> Optional[Dict[str, Any]]: - """将 LangGraph 事件转换为 API 响应块""" - event_type = event.get("event", "") - - if event_type == "on_chat_model_stream": - content = event.get("data", {}).get("chunk", {}).get("content", "") - if content: - return {"type": "content", "content": content} - - elif event_type == "on_tool_start": - return { - "type": "tool_start", - "tool_name": event.get("name", ""), - "session_id": self.session_id - } - - elif event_type == "on_tool_end": - return { - "type": "tool_end", - "tool_name": event.get("name", ""), - "tool_output": event.get("data", {}).get("output", ""), - "session_id": self.session_id - } - - return None - - async def get_history(self, session_id: str, limit: int = 100) -> Dict[str, Any]: - """获取会话历史""" - config = {"configurable": {"thread_id": session_id}} - - try: - state = await self.graph.aget_state(config) - if state and "messages" in state.values: - messages = [] - for msg in state.values["messages"][-limit:]: - messages.append({ - "type": type(msg).__name__, - "content": msg.content if hasattr(msg, 'content') else str(msg) - }) - - title = state.values.get("conversation_title", "New Conversation") - - return { - "thread_id": session_id, - "title": title, - "messages": messages - } - except Exception: - pass - - return { - "thread_id": session_id, - "title": "New Conversation", - "messages": [] - } - - def close(self): - """关闭连接""" - if self._checkpointer: - self._checkpointer.conn.close() -``` - -### 3. ProjectAgentManager - -**文件**: `gns3server/agent/gns3_copilot/project_agent_manager.py` - -```python -""" -Project Agent Manager -管理每个项目的 Agent Service 实例（单例模式） -""" - -import os -from typing import Dict, Optional -from threading import Lock - -from gns3server.agent.gns3_copilot.agent_service import AgentService - - -class ProjectAgentManager: - """项目级 Agent 管理器""" - - _instance = None - _lock = Lock() - - def __new__(cls): - if cls._instance is None: - with cls._lock: - if cls._instance is None: - cls._instance = super().__new__(cls) - cls._instance._agents: Dict[str, AgentService] = {} - return cls._instance - - def get_agent(self, project_id: str, project_path: str) -> AgentService: - """ - 获取或创建项目的 Agent Service - """ - key = project_id - - with self._lock: - if key not in self._agents: - self._agents[key] = AgentService(project_path) - return self._agents[key] - - def remove_agent(self, project_id: str): - """移除项目的 Agent Service""" - key = project_id - - with self._lock: - if key in self._agents: - self._agents[key].close() - del self._agents[key] - - def close_all(self): - """关闭所有 Agent""" - with self._lock: - for agent in self._agents.values(): - agent.close() - self._agents.clear() - - -# 全局单例 -_project_agent_manager: Optional[ProjectAgentManager] = None - - -def get_project_agent_manager() -> ProjectAgentManager: - """获取项目 Agent 管理器""" - global _project_agent_manager - if _project_agent_manager is None: - _project_agent_manager = ProjectAgentManager() - return _project_agent_manager -``` - -### 4. Chat API Routes - -**文件**: `gns3server/api/routes/controller/chat.py` - -```python -""" -Chat API endpoints -""" - -import json -import uuid -from typing import Optional - -from fastapi import APIRouter, HTTPException, Depends, status, Request -from fastapi.responses import StreamingResponse - -from gns3server.schemas.controller.chat import ( - ChatRequest, ChatResponse, ConversationHistory, ChatSession -) -from gns3server.agent.gns3_copilot.project_agent_manager import get_project_agent_manager -from gns3server.controller import Controller -from gns3server.controller.controller_error import ControllerNotFoundError -from gns3server.api.routes.controller.dependencies.authentication import get_current_active_user - - -router = APIRouter() - - -@router.post("/stream") -async def stream_chat( - request: ChatRequest, - http_request: Request, - current_user = Depends(get_current_active_user) -): - """流式 Chat API""" - - # 验证项目 - project_path = None - if request.project_id: - try: - controller = Controller.instance() - project = controller.projects.get(request.project_id) - if not project: - raise ControllerNotFoundError(f"Project '{request.project_id}' not found") - project_path = project.path - except ControllerNotFoundError: - raise - except Exception as e: - raise HTTPException(status_code=400, detail=f"Invalid project: {e}") - - if not project_path: - raise HTTPException(status_code=400, detail="project_id is required") - - # 获取用户认证信息 - user_id = str(current_user.user_id) - - # 获取 JWT token (从 Authorization header) - auth_header = http_request.headers.get("Authorization", "") - jwt_token = auth_header.replace("Bearer ", "") if auth_header else None - - # 获取 Agent Service - agent_manager = get_project_agent_manager() - agent_service = agent_manager.get_agent(request.project_id, project_path) - - session_id = request.session_id or str(uuid.uuid4()) - - async def generate(): - try: - async for chunk in agent_service.stream_chat( - message=request.message, - session_id=session_id, - project_id=request.project_id, - user_id=user_id, - jwt_token=jwt_token, - mode=request.mode - ): - try: - validated = ChatResponse(**chunk) - yield f"data: {json.dumps(validated.model_dump(exclude_none=True), ensure_ascii=False)}\n\n" - except Exception: - pass - - yield f"data: {json.dumps({'type': 'done', 'session_id': session_id})}\n\n" - - except Exception as e: - yield f"data: {json.dumps({'type': 'error', 'error': str(e)})}\n\n" - - return StreamingResponse( - generate(), - media_type="text/event-stream", - headers={ - "Cache-Control": "no-cache", - "X-Accel-Buffering": "no" - } - ) - - -@router.get("/history/{session_id}") -async def get_history( - session_id: str, - project_id: str, - limit: int = 100, - current_user = Depends(get_current_active_user) -): - """获取会话历史""" - - controller = Controller.instance() - project = controller.projects.get(project_id) - if not project: - raise ControllerNotFoundError(f"Project '{project_id}' not found") - - agent_manager = get_project_agent_manager() - agent_service = agent_manager.get_agent(project_id, project.path) - - history = await agent_service.get_history(session_id, limit) - return history -``` - -### 5. 注册路由 - -**文件**: `gns3server/api/routes/controller/__init__.py` - -添加 chat router 注册： - -```python -from . import chat - -# ... 其他 router ... - -router.include_router( - chat.router, - prefix="/chat", - tags=["Chat"] -) -``` - -## API 端点 - -| 方法 | 端点 | 说明 | +| 字段 | 类型 | 说明 | |------|------|------| -| POST | `/v3/chat/stream` | 流式 Chat（主要接口） | -| GET | `/v3/chat/history/{session_id}?project_id=xxx` | 获取会话历史 | +| id | INTEGER | 主键（自增） | +| thread_id | TEXT | LangGraph thread_id（唯一） | +| user_id | TEXT | 用户 ID | +| project_id | TEXT | GNS3 项目 ID | +| title | TEXT | 会话标题 | +| message_count | INTEGER | 消息数量 | +| llm_calls_count | INTEGER | LLM 调用次数 | +| input_tokens | INTEGER | 输入 token 总数 | +| output_tokens | INTEGER | 输出 token 总数 | +| total_tokens | INTEGER | 总 token 数 | +| last_message_at | TIMESTAMP | 最后消息时间 | +| created_at | TIMESTAMP | 创建时间 | +| updated_at | TIMESTAMP | 更新时间 | +| metadata | TEXT | 预留元数据（JSON） | +| stats | TEXT | 额外统计信息（JSON） | + +### ChatSessionsRepository + +提供会话的 CRUD 操作： + +- **create_session**：创建新会话 +- **get_session_by_thread**：根据 thread_id 查询会话 +- **list_sessions**：列出用户的会话（支持过滤和分页） +- **update_session**：更新会话（支持增量更新计数器） +- **delete_session**：删除会话及其 checkpoints +- **delete_all_sessions**：删除项目的所有会话 + +### 统计信息自动收集 + +在 `stream_chat` 流程中自动追踪： +- 每次用户消息：message_count +1 +- 每次 AI 响应：message_count +1 +- 每次工具调用：llm_calls_count +1 +- LLM 返回的 token 使用量：实时累加 + +流结束后一次性更新到数据库。 + +### Title 自动同步 + +会话标题由 `generate_title` 节点自动生成，保存在 LangGraph checkpoint 的 `conversation_title` 字段中。 + +**同步机制**： +1. 流式 Chat 完成后，从 checkpoint 读取最终 state +2. 检查 `conversation_title` 是否有变化 +3. 如果有变化，更新到 `chat_sessions` 表 + +**优势**： +- 避免在节点中直接访问数据库（防止循环依赖） +- 所有数据库更新集中在流结束后 +- 逻辑清晰，易于维护 ## SSE 消息格式 +Chat API 使用 Server-Sent Events (SSE) 进行流式传输。 + +### 消息类型 + +| type | 说明 | 包含字段 | +|------|------|----------| +| content | AI 文本内容（流式） | content | +| tool_call | 工具调用请求 | tool_call (id, name, arguments) | +| tool_start | 工具开始执行 | tool_name | +| tool_end | 工具执行完成 | tool_name, tool_output | +| error | 错误信息 | error | +| done | 流结束 | session_id | +| heartbeat | 心跳保活 | session_id | + +### 消息示例 + ```json -// 内容块 -{"type": "content", "content": "Hello!"} +// AI 文本流式输出 +{"type": "content", "content": "Hello! How can I help"} + +// 工具调用 +{"type": "tool_call", "tool_call": {"id": "call_123", "function": {"name": "GNS3TopologyTool", "arguments": {"project_id": "xxx"}}}} // 工具开始 {"type": "tool_start", "tool_name": "GNS3TopologyTool", "session_id": "xxx"} -// 工具结束 -{"type": "tool_end", "tool_name": "GNS3TopologyTool", "tool_output": "...", "session_id": "xxx"} +// 工具完成 +{"type": "tool_end", "tool_name": "GNS3TopologyTool", "tool_output": "{...}", "session_id": "xxx"} // 完成 {"type": "done", "session_id": "xxx"} // 错误 -{"type": "error", "error": "Error message", "session_id": "xxx"} +{"type": "error", "error": "Project not found", "session_id": "xxx"} ``` +### 心跳机制 + +**作用**：防止代理服务器/负载均衡器因超时断开 SSE 连接。 + +**实现**：使用 `asyncio.wait` 设置超时，超时后发送 `heartbeat` 消息，然后继续等待下一个事件。 + +**前端处理**：收到 `heartbeat` 消息时直接忽略，不渲染任何内容。 + +## API 端点 + +所有端点都在 `/v3/projects/{project_id}/chat/` 路径下。 + +| 方法 | 端点 | 说明 | +|------|------|------| +| POST | `/stream` | 流式 Chat（主要接口） | +| GET | `/sessions` | 列出会话 | +| GET | `/sessions/{session_id}/history` | 获取会话历史 | +| PATCH | `/sessions/{session_id}` | 重命名会话 | +| DELETE | `/sessions/{session_id}` | 删除会话 | + +### POST /v3/projects/{project_id}/chat/stream + +**功能**：流式对话接口 + +**请求参数**： +- message: 用户消息内容 +- session_id: 会话 ID（可选，不提供则自动创建新会话） +- stream: 是否启用流式响应（默认 true） +- mode: 交互模式（当前仅支持 "text"） + +**响应**：SSE 流，包含多种类型的消息（见上文消息格式） + +**项目状态检查**：只允许项目状态为 "opened" 时进行对话 + +### GET /v3/projects/{project_id}/chat/sessions + +**功能**：列出项目的所有会话 + +**响应**：会话列表，包含统计信息（消息数、token 使用量等） + +### GET /v3/projects/{project_id}/chat/sessions/{session_id}/history + +**功能**：获取会话的完整历史记录 + +**参数**： +- session_id: 会话 ID +- limit: 最大消息数量（默认 100） + +**响应**： +- thread_id: 会话 ID +- title: 会话标题 +- messages: 消息列表（OpenAI 格式） +- llm_calls: LLM 调用次数 + +### PATCH /v3/projects/{project_id}/chat/sessions/{session_id} + +**功能**：重命名会话 + +**请求参数**： +- title: 新标题（1-255 字符） + +**响应**：更新后的会话信息 + +### DELETE /v3/projects/{project_id}/chat/sessions/{session_id} + +**功能**：删除会话及其所有 checkpoint 数据 + +**响应**：204 No Content + +## 数据模型 + +### ChatRequest + +- message: str - 用户消息内容 +- session_id: Optional[str] - 会话 ID（可选） +- stream: bool - 是否流式响应（默认 true） +- mode: Literal["text"] - 交互模式 + +### ChatSession + +- id: Optional[int] - 数据库 ID +- thread_id: str - Thread/Session ID +- user_id: str - 用户 ID +- project_id: str - 项目 ID +- title: str - 会话标题 +- message_count: int - 消息数量 +- llm_calls_count: int - LLM 调用次数 +- input_tokens: int - 输入 token 数 +- output_tokens: int - 输出 token 数 +- total_tokens: int - 总 token 数 +- last_message_at: Optional[str] - 最后消息时间 +- created_at: Optional[str] - 创建时间 +- updated_at: Optional[str] - 更新时间 +- metadata: Dict - 预留元数据 +- stats: Dict - 额外统计信息 + +### ConversationHistory + +- thread_id: str - 会话 ID +- title: str - 会话标题 +- messages: List[OpenAIMessage] - 消息列表 +- created_at: Optional[str] - 创建时间 +- updated_at: Optional[str] - 更新时间 +- llm_calls: int - LLM 调用次数 + +### OpenAIMessage + +- id: str - 消息 ID +- role: Literal["user", "assistant", "system", "tool"] - 角色 +- content: str - 消息内容 +- name: Optional[str] - 工具消息名称 +- tool_call_id: Optional[str] - 关联的工具调用 ID +- tool_calls: Optional[List] - 工具调用列表（assistant 消息） +- created_at: str - 创建时间 + +## 核心组件 + +### AgentService + +**职责**：项目级的 Agent 管理服务 + +**主要方法**： +- `stream_chat`：流式对话，自动管理会话和统计 +- `get_history`：获取会话历史 +- `list_sessions`：列出会话 +- `delete_session`：删除会话 +- `rename_session`：重命名会话 +- `close`：关闭数据库连接 + +**核心流程**（stream_chat）： +1. 获取或创建会话 +2. 设置 ContextVars（JWT token、LLM config） +3. 构建 LangGraph config +4. 流式执行 Agent，收集统计信息 +5. 流结束后更新会话统计 +6. 同步 auto-generated title + +**连接管理**： +- 使用 `AsyncSqliteSaver` 作为 checkpointer +- 支持 WAL 模式提升并发性能 +- 项目切换时自动关闭旧连接 +- 防止连接被垃圾回收（保存引用） + +### ProjectAgentManager + +**职责**：全局单例，管理所有项目的 AgentService 实例 + +**方法**： +- `get_agent(project_id, project_path)`：获取或创建项目的 AgentService +- `remove_agent(project_id)`：移除项目的 AgentService +- `close_all`：关闭所有 AgentService + +### Chat API Routes + +**文件**：`gns3server/api/routes/controller/chat.py` + +**路由注册**： +```python +router.include_router( + chat.router, + prefix="/{project_id}/chat", + tags=["Chat"] +) +``` + +**主要端点实现**： +- 所有端点都需要用户认证（`get_current_active_user`） +- 所有端点都检查项目状态是否为 "opened" +- stream 端点使用 `StreamingResponse` 返回 SSE 流 + ## 项目生命周期集成 -在项目打开/关闭时，需要管理 Agent Service 实例： +### 项目打开时 +创建或获取 AgentService 实例： ```python -# 项目打开时 -project_agent_manager.get_agent(project_id, project.path) - -# 项目关闭时 -project_agent_manager.remove_agent(project_id) +agent_manager = await get_project_agent_manager() +agent_service = await agent_manager.get_agent(project_id, project.path) ``` -可以监听项目事件或使用信号机制实现。 +### 项目关闭时 + +移除 AgentService 实例，释放资源： +```python +agent_manager.remove_agent(project_id) +``` + +### 项目删除时 + +1. 调用 `delete_all_sessions(project_id)` 删除所有会话和 checkpoint 数据 +2. 移除 AgentService 实例 +3. 项目目录被删除，数据库文件也被删除 + +## 前端集成 + +### useChat Hook + +根据 SSE 消息的 `type` 字段进行不同处理： + +| type | 处理逻辑 | +|------|----------| +| content | 追加到当前 AI 消息内容 | +| tool_call | 创建 tool_call 类型消息，显示工具调用信息 | +| tool_start | 可选：显示工具开始执行状态 | +| tool_end | 创建 tool_result 类型消息，显示工具执行结果 | +| error | 显示错误信息 | +| done | 标记流结束，停止加载状态 | +| heartbeat | 忽略（保活信号） | + +### 错误处理 + +- 网络错误：显示重试选项 +- LLM 错误：显示错误消息 +- 项目未打开：提示用户打开项目 +- LLM 未配置：引导用户配置 LLM + +## 安全考虑 + +### 用户隔离 + +- 每个用户只能访问自己的会话 +- user_id 存储在 config.metadata 中 +- 所有数据库查询都带 user_id 过滤 + +### 项目访问控制 + +- 只允许访问用户有权限的项目 +- 项目状态检查：只允许 "opened" 状态的项目使用 Chat + +### LLM 配置安全 + +- API key 加密存储在数据库 +- 使用 ContextVars 传递，不持久化到 checkpoint +- 请求结束后自动清理内存中的敏感信息 + +## 性能优化 + +### 数据库连接管理 + +- 使用 WAL 模式提升并发写入性能 +- 项目级连接复用 +- 项目切换时自动关闭旧连接 + +### Checkpoint 优化 + +- LangGraph 自动管理 checkpoints 表 +- 定期清理旧 checkpoint（可选） +- 使用索引加速查询（thread_id, user_id + project_id） + +### 统计信息批量更新 + +- 流结束后一次性更新统计信息 +- 避免频繁的数据库写入 ## 依赖项 -确保以下包已安装： - - `langchain` >= 0.3.0 - `langgraph` >= 0.2.0 - `langchain-core` -- `sqlalchemy` (LangGraph 依赖) +- `langgraph-checkpoint-sqlite` >= 3.0.1 +- `aiosqlite` + +## 扩展性 + +### 预留字段 + +- `metadata`（TEXT JSON）：存储会话级别的元数据 +- `stats`（TEXT JSON）：存储额外的统计信息 + +### 未来可能的扩展 + +- 多模态支持（图片、文件） +- 语音输入/输出 +- 多人协作会话 +- 会话分享和导出 +- 自定义工具注册 ## 参考资料 - [LangGraph Checkpoint Documentation](https://langchain-ai.github.io/langgraph/how-tos/checkpointers/) -- [FlowNet-Lab Chat API](file:///home/yueguobin/myCode/GNS3/FlowNet-Lab/backend/api/v1/chat.py) -- [FlowNet-Lab Agent Service](file:///home/yueguobin/myCode/GNS3/FlowNet-Lab/backend/core/agent.py) +- [Server-Sent Events (MDN)](https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events) +- [OpenAI Chat Format](https://platform.openai.com/docs/api-reference/chat) diff --git a/gns3server/agent/gns3_copilot/agent_service.py b/gns3server/agent/gns3_copilot/agent_service.py index 5275d405c..efb4afcdf 100644 --- a/gns3server/agent/gns3_copilot/agent_service.py +++ b/gns3server/agent/gns3_copilot/agent_service.py @@ -7,8 +7,10 @@ in the project directory. """ import asyncio +import json import logging import os +from datetime import datetime from typing import AsyncGenerator, Dict, Any, Optional from uuid import uuid4 @@ -17,6 +19,7 @@ from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver from gns3server.agent.gns3_copilot.agent.gns3_copilot import agent_builder +from gns3server.agent.gns3_copilot.chat_sessions_repository import ChatSessionsRepository log = logging.getLogger(__name__) @@ -84,12 +87,55 @@ class AgentService: # CRITICAL: Initialize database schema await self._checkpointer.setup() + # Create chat_sessions table in the same database + await self._create_chat_sessions_table(conn) + self._checkpointer_path = checkpointer_path self._initialized = True log.info("Project checkpointer created at: %s", checkpointer_path) return self._checkpointer + async def _create_chat_sessions_table(self, conn: aiosqlite.Connection): + """ + Create the chat_sessions table in the checkpoint database. + + Args: + conn: aiosqlite connection + """ + await conn.execute(""" + CREATE TABLE IF NOT EXISTS chat_sessions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + thread_id TEXT UNIQUE NOT NULL, + user_id TEXT NOT NULL, + project_id TEXT NOT NULL, + title TEXT DEFAULT 'New Conversation', + + -- Statistics + message_count INTEGER DEFAULT 0, + llm_calls_count INTEGER DEFAULT 0, + input_tokens INTEGER DEFAULT 0, + output_tokens INTEGER DEFAULT 0, + total_tokens INTEGER DEFAULT 0, + + -- Timestamps + last_message_at TIMESTAMP, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + + -- Reserved fields (JSON strings) + metadata TEXT DEFAULT '{}', + stats TEXT DEFAULT '{}' + ) + """) + + # Create indexes + await conn.execute("CREATE INDEX IF NOT EXISTS idx_thread_id ON chat_sessions(thread_id)") + await conn.execute("CREATE INDEX IF NOT EXISTS idx_user_project ON chat_sessions(user_id, project_id)") + + await conn.commit() + log.debug("chat_sessions table created in checkpoint database") + async def _get_graph(self): """Get or compile the LangGraph agent.""" if self._graph is None: @@ -131,6 +177,21 @@ class AgentService: mode, ) + # 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( + thread_id=session_id, + user_id=user_id or "", + project_id=project_id or "", + title="New Conversation" + ) + log.debug("Created new chat session: thread_id=%s", session_id) + # Set request-scoped context variables (memory-only, not persisted) if jwt_token: from gns3server.agent.gns3_copilot.gns3_client import set_current_jwt_token @@ -165,14 +226,59 @@ class AgentService: graph = await self._get_graph() log.debug("LangGraph graph obtained, starting stream") + # Track statistics for session update + message_count = 1 # User message + llm_calls_count = 0 + tool_calls_count = 0 + input_tokens = 0 + output_tokens = 0 + last_message_at = datetime.utcnow().isoformat() + # Stream events try: async for event in graph.astream_events(inputs, config=config, version="v2"): chunk = self._convert_event_to_chunk(event, session_id) if chunk: + # Track statistics + if chunk.get("type") == "content": + message_count += 1 # AI response + elif chunk.get("type") == "tool_start": + tool_calls_count += 1 + elif chunk.get("type") == "tool_end": + message_count += 1 # Tool message + + # Track tokens if available + if chunk.get("type") == "content" and "input_tokens" in chunk: + input_tokens += chunk.get("input_tokens", 0) + if chunk.get("type") == "content" and "output_tokens" in chunk: + output_tokens += chunk.get("output_tokens", 0) + log.debug("Yielding chunk: type=%s", chunk.get("type")) yield chunk + # Update session statistics after successful stream + await repo.update_session( + thread_id=session_id, + message_count=message_count, + llm_calls_count=llm_calls_count, + input_tokens=input_tokens, + output_tokens=output_tokens, + total_tokens=input_tokens + output_tokens, + last_message_at=last_message_at + ) + log.debug("Updated session statistics: thread_id=%s, messages=%d, tokens=%d", + session_id, message_count, input_tokens + output_tokens) + + # Sync auto-generated title from checkpoint state + final_state = await graph.aget_state(config) + if final_state and "conversation_title" in final_state.values: + generated_title = final_state.values["conversation_title"] + current_session = await repo.get_session_by_thread(session_id) + if current_session and current_session.title != generated_title: + await repo.update_session(thread_id=session_id, title=generated_title) + log.info("Auto-generated title synced: thread_id=%s, title=%s", + session_id, generated_title) + except Exception as e: log.error("Error in stream_chat: %s", e, exc_info=True) yield {"type": "error", "error": str(e), "session_id": session_id} @@ -288,6 +394,58 @@ class AgentService: return result + async def list_sessions(self, user_id: Optional[str] = None, limit: int = 100) -> List[Dict[str, Any]]: + """ + List chat sessions for this project. + + Args: + user_id: Filter by user ID (optional) + limit: Maximum number of sessions to return + + Returns: + List of session dictionaries + """ + if not self._checkpointer_conn: + await self._get_checkpointer() + + repo = ChatSessionsRepository(self._checkpointer_conn) + sessions = await repo.list_sessions(user_id=user_id, limit=limit) + return [s.to_dict() for s in sessions] + + async def delete_session(self, session_id: str) -> bool: + """ + Delete a chat session and its checkpoints. + + Args: + session_id: Thread ID to delete + + Returns: + True if deleted, False if not found + """ + if not self._checkpointer_conn: + await self._get_checkpointer() + + repo = ChatSessionsRepository(self._checkpointer_conn) + return await repo.delete_session(session_id) + + async def rename_session(self, session_id: str, new_title: str) -> Optional[Dict[str, Any]]: + """ + Rename a chat session. + + Args: + session_id: Thread ID + new_title: New title + + Returns: + Updated session dictionary or None + """ + if not self._checkpointer_conn: + await self._get_checkpointer() + + repo = ChatSessionsRepository(self._checkpointer_conn) + session = await repo.update_session(thread_id=session_id, title=new_title) + return session.to_dict() if session else None + async def close(self): """ Close the checkpointer connection and cleanup resources. diff --git a/gns3server/agent/gns3_copilot/chat_sessions_repository.py b/gns3server/agent/gns3_copilot/chat_sessions_repository.py new file mode 100644 index 000000000..9ccb4502d --- /dev/null +++ b/gns3server/agent/gns3_copilot/chat_sessions_repository.py @@ -0,0 +1,365 @@ +""" +Chat Sessions Repository for managing chat session data. + +Provides CRUD operations for the chat_sessions table in the project's +checkpoint database. +""" + +import json +import logging +from datetime import datetime +from typing import Any, Dict, List, Optional +from uuid import UUID + +import aiosqlite + +log = logging.getLogger(__name__) + + +class ChatSession: + """Chat session model.""" + + def __init__( + self, + id: Optional[int] = None, + thread_id: str = "", + user_id: str = "", + project_id: str = "", + title: str = "New Conversation", + message_count: int = 0, + llm_calls_count: int = 0, + input_tokens: int = 0, + output_tokens: int = 0, + total_tokens: int = 0, + last_message_at: Optional[str] = None, + created_at: Optional[str] = None, + updated_at: Optional[str] = None, + metadata: str = "{}", + stats: str = "{}" + ): + self.id = id + self.thread_id = thread_id + self.user_id = user_id + self.project_id = project_id + self.title = title + self.message_count = message_count + self.llm_calls_count = llm_calls_count + self.input_tokens = input_tokens + self.output_tokens = output_tokens + self.total_tokens = total_tokens + self.last_message_at = last_message_at + self.created_at = created_at + self.updated_at = updated_at + self.metadata = metadata + self.stats = stats + + def to_dict(self) -> Dict[str, Any]: + """Convert to dictionary.""" + return { + "id": self.id, + "thread_id": self.thread_id, + "user_id": self.user_id, + "project_id": self.project_id, + "title": self.title, + "message_count": self.message_count, + "llm_calls_count": self.llm_calls_count, + "input_tokens": self.input_tokens, + "output_tokens": self.output_tokens, + "total_tokens": self.total_tokens, + "last_message_at": self.last_message_at, + "created_at": self.created_at, + "updated_at": self.updated_at, + "metadata": json.loads(self.metadata) if self.metadata else {}, + "stats": json.loads(self.stats) if self.stats else {}, + } + + +class ChatSessionsRepository: + """ + Repository for managing chat sessions in the checkpoint database. + """ + + def __init__(self, conn: aiosqlite.Connection): + """ + Initialize repository with a database connection. + + Args: + conn: aiosqlite connection to the checkpoint database + """ + self.conn = conn + + async def create_session( + self, + thread_id: str, + user_id: str, + project_id: str, + title: str = "New Conversation" + ) -> ChatSession: + """ + Create a new chat session. + + Args: + thread_id: Unique thread identifier + user_id: User ID + project_id: Project ID + title: Session title + + Returns: + Created ChatSession + """ + now = datetime.utcnow().isoformat() + cursor = await self.conn.execute( + """ + INSERT INTO chat_sessions ( + thread_id, user_id, project_id, title, + created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?) + """, + (thread_id, user_id, project_id, title, now, now) + ) + await self.conn.commit() + + session_id = cursor.lastrowid + log.info("Created chat session: id=%s, thread_id=%s", session_id, thread_id) + + return await self.get_session_by_id(session_id) + + async def get_session_by_id(self, session_id: int) -> Optional[ChatSession]: + """ + Get a session by its database ID. + + Args: + session_id: Database row ID + + Returns: + ChatSession or None + """ + cursor = await self.conn.execute( + "SELECT * FROM chat_sessions WHERE id = ?", + (session_id,) + ) + row = await cursor.fetchone() + + if row: + return self._row_to_session(row) + return None + + async def get_session_by_thread(self, thread_id: str) -> Optional[ChatSession]: + """ + Get a session by thread_id. + + Args: + thread_id: Thread identifier + + Returns: + ChatSession or None + """ + cursor = await self.conn.execute( + "SELECT * FROM chat_sessions WHERE thread_id = ?", + (thread_id,) + ) + row = await cursor.fetchone() + + if row: + return self._row_to_session(row) + return None + + async def list_sessions( + self, + user_id: Optional[str] = None, + project_id: Optional[str] = None, + limit: int = 100 + ) -> List[ChatSession]: + """ + List sessions with optional filters. + + Args: + user_id: Filter by user ID + project_id: Filter by project ID + limit: Maximum number of sessions to return + + Returns: + List of ChatSession + """ + query = "SELECT * FROM chat_sessions" + params = [] + + conditions = [] + if user_id: + conditions.append("user_id = ?") + params.append(user_id) + if project_id: + conditions.append("project_id = ?") + params.append(project_id) + + if conditions: + query += " WHERE " + " AND ".join(conditions) + + query += " ORDER BY updated_at DESC LIMIT ?" + params.append(limit) + + cursor = await self.conn.execute(query, params) + rows = await cursor.fetchall() + + return [self._row_to_session(row) for row in rows] + + async def update_session( + self, + thread_id: str, + title: Optional[str] = None, + message_count: Optional[int] = None, + llm_calls_count: Optional[int] = None, + input_tokens: Optional[int] = None, + output_tokens: Optional[int] = None, + total_tokens: Optional[int] = None, + last_message_at: Optional[str] = None + ) -> Optional[ChatSession]: + """ + Update a session. + + Args: + thread_id: Thread identifier + title: New title + message_count: Increment message count + llm_calls_count: Increment LLM call count + input_tokens: Add to input tokens + output_tokens: Add to output tokens + total_tokens: Add to total tokens + last_message_at: Last message timestamp + + Returns: + Updated ChatSession or None + """ + updates = [] + params = [] + + now = datetime.utcnow().isoformat() + + if title is not None: + updates.append("title = ?") + params.append(title) + + if message_count is not None: + updates.append("message_count = message_count + ?") + params.append(message_count) + + if llm_calls_count is not None: + updates.append("llm_calls_count = llm_calls_count + ?") + params.append(llm_calls_count) + + if input_tokens is not None: + updates.append("input_tokens = input_tokens + ?") + params.append(input_tokens) + + if output_tokens is not None: + updates.append("output_tokens = output_tokens + ?") + params.append(output_tokens) + + if total_tokens is not None: + updates.append("total_tokens = total_tokens + ?") + params.append(total_tokens) + + if last_message_at is not None: + updates.append("last_message_at = ?") + params.append(last_message_at) + + if not updates: + return await self.get_session_by_thread(thread_id) + + updates.append("updated_at = ?") + params.append(now) + params.append(thread_id) + + query = f"UPDATE chat_sessions SET {', '.join(updates)} WHERE thread_id = ?" + + await self.conn.execute(query, params) + await self.conn.commit() + + log.debug("Updated chat session: thread_id=%s", thread_id) + return await self.get_session_by_thread(thread_id) + + async def delete_session(self, thread_id: str) -> bool: + """ + Delete a session by thread_id. + + Args: + thread_id: Thread identifier + + Returns: + True if deleted, False if not found + """ + # First, delete the checkpoint data + await self.conn.execute( + "DELETE FROM checkpoints WHERE thread_id = ?", + (thread_id,) + ) + + # Then delete the session + cursor = await self.conn.execute( + "DELETE FROM chat_sessions WHERE thread_id = ?", + (thread_id,) + ) + await self.conn.commit() + + deleted = cursor.rowcount > 0 + if deleted: + log.info("Deleted chat session and checkpoints: thread_id=%s", thread_id) + + return deleted + + async def delete_all_sessions(self, project_id: str) -> int: + """ + Delete all sessions for a project. + + Args: + project_id: Project ID + + Returns: + Number of sessions deleted + """ + # Get all thread_ids for this project + cursor = await self.conn.execute( + "SELECT thread_id FROM chat_sessions WHERE project_id = ?", + (project_id,) + ) + rows = await cursor.fetchall() + thread_ids = [row[0] for row in rows] + + # Delete checkpoints and sessions + for thread_id in thread_ids: + await self.conn.execute( + "DELETE FROM checkpoints WHERE thread_id = ?", + (thread_id,) + ) + + cursor = await self.conn.execute( + "DELETE FROM chat_sessions WHERE project_id = ?", + (project_id,) + ) + await self.conn.commit() + + deleted_count = cursor.rowcount + if deleted_count > 0: + log.info("Deleted %d sessions for project: %s", deleted_count, project_id) + + return deleted_count + + def _row_to_session(self, row) -> ChatSession: + """Convert database row to ChatSession object.""" + return ChatSession( + id=row[0], + thread_id=row[1], + user_id=row[2], + project_id=row[3], + title=row[4], + message_count=row[5], + llm_calls_count=row[6], + input_tokens=row[7], + output_tokens=row[8], + total_tokens=row[9], + last_message_at=row[10], + created_at=row[11], + updated_at=row[12], + metadata=row[13], + stats=row[14], + ) diff --git a/gns3server/api/routes/controller/chat.py b/gns3server/api/routes/controller/chat.py index 596af59b6..efb45d138 100644 --- a/gns3server/api/routes/controller/chat.py +++ b/gns3server/api/routes/controller/chat.py @@ -189,7 +189,7 @@ async def stream_chat( "/sessions", response_model=List[schemas.ChatSession], summary="List chat sessions", - description="List all chat sessions for a project (not yet implemented)." + description="List all chat sessions for a project." ) async def list_sessions( project: Project = Depends(dep_project), @@ -197,9 +197,6 @@ async def list_sessions( ) -> list[schemas.ChatSession]: """ List chat sessions for a project. - - Note: This endpoint is a placeholder. Full session listing functionality - requires checkpoint metadata inspection which is not yet implemented. """ # Check if project is opened @@ -209,8 +206,15 @@ async def list_sessions( detail=f"Project must be opened to access chat sessions. Current status: {project.status}" ) - # TODO: Implement session listing from checkpoint metadata - return [] + # Get AgentService for this project + agent_manager = await get_project_agent_manager() + agent_service = await agent_manager.get_agent(str(project.id), project.path) + + # List sessions + sessions = await agent_service.list_sessions(user_id=str(current_user.user_id)) + + # Convert to schemas + return [schemas.ChatSession(**s) for s in sessions] @router.get( @@ -250,7 +254,7 @@ async def get_history( "/sessions/{session_id}", status_code=status.HTTP_204_NO_CONTENT, summary="Delete a chat session", - description="Delete a specific chat session (not yet implemented)." + description="Delete a specific chat session and its checkpoints." ) async def delete_session( session_id: str, @@ -259,9 +263,6 @@ async def delete_session( ): """ Delete a chat session. - - Note: This endpoint is a placeholder. Full session deletion functionality - requires checkpoint manipulation which is not yet implemented. """ # Check if project is opened @@ -271,8 +272,54 @@ async def delete_session( detail=f"Project must be opened to delete chat sessions. Current status: {project.status}" ) - # TODO: Implement session deletion from checkpoint - raise HTTPException( - status_code=status.HTTP_501_NOT_IMPLEMENTED, - detail="Session deletion not yet implemented" - ) + # Get AgentService for this project + agent_manager = await get_project_agent_manager() + agent_service = await agent_manager.get_agent(str(project.id), project.path) + + # Delete session + deleted = await agent_service.delete_session(session_id) + + if not deleted: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Session '{session_id}' not found" + ) + + +@router.patch( + "/sessions/{session_id}", + response_model=schemas.ChatSession, + summary="Rename a chat session", + description="Rename a specific chat session." +) +async def rename_session( + session_id: str, + request: schemas.RenameSession, + project: Project = Depends(dep_project), + current_user: schemas.User = Depends(get_current_active_user), +) -> schemas.ChatSession: + """ + Rename a chat session. + """ + + # Check if project is opened + if project.status != "opened": + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=f"Project must be opened to rename chat sessions. Current status: {project.status}" + ) + + # Get AgentService for this project + agent_manager = await get_project_agent_manager() + agent_service = await agent_manager.get_agent(str(project.id), project.path) + + # Rename session + session = await agent_service.rename_session(session_id, request.title) + + if not session: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Session '{session_id}' not found" + ) + + return schemas.ChatSession(**session) diff --git a/gns3server/schemas/__init__.py b/gns3server/schemas/__init__.py index 843349891..2ea48dd6c 100644 --- a/gns3server/schemas/__init__.py +++ b/gns3server/schemas/__init__.py @@ -45,7 +45,8 @@ from .controller.chat import ( ChatResponse, OpenAIMessage, ConversationHistory, - ChatSession + ChatSession, + RenameSession ) from .controller.rbac import RoleCreate, RoleUpdate, Role, Privilege, ACECreate, ACEUpdate, ACE from .controller.pools import Resource, ResourceCreate, ResourcePoolCreate, ResourcePoolUpdate, ResourcePool diff --git a/gns3server/schemas/controller/chat.py b/gns3server/schemas/controller/chat.py index e19a18de8..0662d65f2 100644 --- a/gns3server/schemas/controller/chat.py +++ b/gns3server/schemas/controller/chat.py @@ -88,8 +88,24 @@ class ConversationHistory(BaseModel): class ChatSession(BaseModel): """Chat session model.""" - session_id: str = Field(..., description="Session ID") + id: Optional[int] = Field(None, description="Database ID") + thread_id: str = Field(..., description="Thread/session ID") + user_id: str = Field(..., description="User ID") + project_id: str = Field(..., description="Associated GNS3 project ID") title: str = Field(..., description="Session title") - project_id: Optional[str] = Field(None, description="Associated GNS3 project ID") + message_count: int = Field(default=0, description="Number of messages") + llm_calls_count: int = Field(default=0, description="Number of LLM calls") + input_tokens: int = Field(default=0, description="Input tokens used") + output_tokens: int = Field(default=0, description="Output tokens generated") + total_tokens: int = Field(default=0, description="Total tokens used") + last_message_at: Optional[str] = Field(None, description="Last message timestamp (ISO 8601)") created_at: Optional[str] = Field(None, description="Creation timestamp (ISO 8601)") updated_at: Optional[str] = Field(None, description="Last update timestamp (ISO 8601)") + metadata: Dict[str, Any] = Field(default_factory=dict, description="Session metadata") + stats: Dict[str, Any] = Field(default_factory=dict, description="Session statistics") + + +class RenameSession(BaseModel): + """Rename session request model.""" + + title: str = Field(..., description="New session title", min_length=1, max_length=255)