feat: stabilize context compression and deep-compact flow

This commit is contained in:
JOJO 2026-04-06 15:12:25 +08:00
parent cb92517cf5
commit 7e03e3cd5f
21 changed files with 1224 additions and 119 deletions

130
AGENTS.md
View File

@ -1,39 +1,103 @@
# Repository Guidelines
# Repository Guidelines (Code-Verified)
## Project Structure & Module Organization
- `main.py` runs the CLI agent; `web_server.py` serves the Flask/SocketIO UI from `static/`.
- `core/` contains terminal orchestration (`main_terminal.py`, `web_terminal.py`) and should stay I/O-light.
- `modules/` collects reusable capabilities (file/search/memory/terminal); extend here before touching entrypoints.
- `utils/` provides API clients, context helpers, and logging; import from here instead of adding globals.
- `prompts/`, `data/`, and `project/` bundle prompt templates, fixtures, and demos used primarily for reference.
- `test/` stores integration helpers (`api_interceptor_server.py`, `test_deepseek_output.py`); add new suites alongside them.
- 前端:`static/src/components/chat/monitor` 是虚拟显示器动画层;涉及新场景请在 `MonitorDirector`/`monitor store` 内对齐数据与动画的状态同步。
> Last verified against current codebase: 2026-04-06
> Scope: `/Users/jojo/Desktop/agents/正在修复中/agents`
## Build, Test, and Development Commands
- `pip install -r requirements.txt` installs Python dependencies.
- `python main.py` launches the CLI agent loop.
- `python web_server.py` serves the web UI at `http://localhost:8091`.
- `python test/api_interceptor_server.py` starts the mock proxy that logs traffic to `api_logs/`.
- `python test_deepseek_output.py` exercises streaming tool-calls; populate `config.py` with valid API credentials first.
- `npm install` then `npm run build` builds the Vue frontend to `static/dist`; `npm run dev` watches; `npm run lint` for TS/Vue linting.
这份文档基于当前仓库实际代码重写。若与未来代码冲突,以代码为准并及时更新本文档。
## Coding Style & Naming Conventions
- Target Python 3.11+, 4-space indentation, `snake_case` functions, and `PascalCase` classes.
- TypeScript 5 / Vue 3 单文件组件,保持 `<script setup lang="ts">` 风格;新增资源放入 `static/src` 对应子目录。
- Add type hints和简短双语 docstring日志统一用 `utils.logger.setup_logger`,仅在需要时使用 `print`
- 模块聚焦:新增能力优先放 `modules/`(或 `static/src/components/chat/monitor` 对应区域)而非入口脚本。
## 1) 当前项目结构(按代码现状)
## Testing Guidelines
- Prefer `pytest` for automation and name files `test/test_<feature>.py`, reusing assets in `data/` when relevant.
- For network flows, run `python test/api_interceptor_server.py` and point the app to the proxy to avoid external calls.
- Capture manual verification steps in PRs and add replay scripts under `test/` for repeated workflows.
- **后端入口**
- `main.py`: 统一启动入口(当前默认走 Web 模式 + thinking_mode=True
- `server/app.py`: 推荐的 Web 服务入口(封装并转发到 `server/app_legacy.py`
- `web_server.py`: 兼容入口,已标记 deprecated但仍可启动
- **后端核心目录**
- `server/`: Flask + Socket.IO 业务主线chat flow、conversation、tasks、status、api_v1 等)
- `core/`: 终端与工具编排(`main_terminal.py`、`web_terminal.py`、`main_terminal_parts/*`
- `modules/`: 可复用能力模块terminal/file/memory/sub_agent/upload_security/user 等)
- `config/`: 配置拆分(`api.py`, `limits.py`, `terminal.py`, `paths.py` ...),由 `config/__init__.py` 聚合并加载 `.env`
- `utils/`: API client、日志、上下文与对话工具等公共函数
- **前端目录**
- `static/src/`: Vue 3 + TS 前端
- 监控动画相关核心文件:
- `static/src/components/chat/monitor/MonitorDirector.ts`
- `static/src/stores/monitor.ts`
- `static/src/components/chat/monitor/*`
- **其他子项目/资源**
- `android-webview-app/`: Android WebView 客户端工程
- `easyagent/`: 独立前端子目录(与根目录前端并存)
## Commit & Pull Request Guidelines
- With no history yet, adopt Conventional Commits (`feat:`, `fix:`, `refactor:`) and keep summaries under 72 characters.
- Squash iterative commits; PRs should outline changes, touched modules, and validation.
- Link issues and attach UI captures or log snippets when behavior shifts.
## 2) 当前可用启动/构建命令(已按代码核对)
## Security & Configuration Tips
- Keep real API keys out of git; override defaults via environment variables.
- Clean sensitive data from `api_logs/`, `logs/`, and `experiment_outputs/` before sharing.
- Extend `.gitignore` when new tooling produces temporary artifacts.
### Python / 服务端
- 安装依赖:`pip install -r requirements.txt`
- 推荐启动 Web`python -m server.app`
- 可选参数:`python -m server.app --path ./project --port 8091 --debug --thinking-mode`
- 兼容启动方式:`python web_server.py`
- 统一入口:`python main.py`(当前实现会默认进入 Web 启动流程)
### Frontend根目录
- 安装依赖:`npm install`
- 构建:`npm run build`
- 开发监听(当前脚本是 build watch`npm run dev`
- Lint`npm run lint`
## 3) 测试现状(不要再写过时命令)
- 当前仓库内可见自动化冒烟:`test/test_server_refactor_smoke.py``unittest`
- 运行方式:`python -m unittest test.test_server_refactor_smoke`
- `test_system_message.py` 依赖外部 `MOONSHOT_API_KEY` 与网络,不属于离线稳定 CI 用例。
- 当前仓库未发现 `pytest.ini`/`pyproject.toml`/`tox.ini`;不要默认要求 `pytest` 作为唯一入口。
## 4) 代码修改约定(实用版)
- **最小改动原则**:只改与需求直接相关文件,避免顺手重构。
- **后端改动优先级**:先改 `modules/`、`server/` 内对应模块,最后才动入口。
- **前端改动优先级**:按 `static/src` 现有分层改(`app/`、`stores/`、`components/`、`composables/`)。
- 涉及 monitor 动画/事件联动时,至少同步检查:
- `MonitorDirector.ts`(动画与场景执行)
- `stores/monitor.ts`(事件队列与状态机)
## 5) 风格与质量
- Python4 空格、`snake_case` 函数、`PascalCase` 类,新增公共函数尽量补 type hints。
- Vue/TS保持现有代码风格不做无关风格清洗。
- 日志:优先复用现有 logger/日志路径,不引入大量临时 `print`
- 提交前至少做与改动相关的最小验证(命令输出或手工步骤要可复现)。
## 6) 提交与 PR
- 使用 Conventional Commits`feat:` `fix:` `refactor:` `chore:` ...
- PR 描述至少包含:
- 改了哪些文件
- 为什么改
- 如何验证
- 风险/回滚点
## 7) 安全与仓库卫生
- 严禁提交真实密钥(`.env`、token、cookie、用户隐私
- `logs/`, `data/`, `users/`, `project/` 为运行态目录(见 `.gitignore`),分享前需脱敏。
- 不要把本地构建产物(如 `static/dist/`、`node_modules/`)纳入提交。
## 8) Android App 发布联动要求(重要)
当修改前端并需要发布 Android WebView App 时,必须同步执行以下步骤(依据 `doc/android_app_release_and_update.md`
1. 同时更新版本信息
- `android-webview-app/app/build.gradle.kts`:递增 `versionCode`,更新 `versionName`
2. 同时更新更新说明
- `android-webview-app/APP_CHANGELOG.md`:在顶部新增当前版本说明
3. 完成修改后提醒用户运行上传脚本
- 在发布流程中运行:`bash /Users/jojo/Desktop/agents/正在修复中/agents/upload_android_apk.sh`
> 禁止只改前端代码而不更新版本号/更新说明,否则会导致客户端更新提示与分发信息不一致。
## 9) 给 Agent 的硬性要求
- 先读当前代码再执行,不依赖历史文档记忆。
- 输出结果必须带:
1. 修改文件列表
2. 核心变更点
3. 验证结果(已执行命令/未执行原因)
- 如果发现本文档过时,直接更新 `AGENTS.md` 并在结果中说明依据。

View File

@ -211,3 +211,17 @@ agents/
- 容器镜像定制:编辑 `docker/terminal.Dockerfile` 并重新构建
- 前端开发:`npm install && npm run dev` (在 `static/` 目录)
- 查看详细执行流程:监控 `logs/debug_stream.log`
## Android Release Notes (Required for frontend update)
当改动前端并准备发 Android WebView 壳版本时,必须同步完成:
1. 更新版本号(`android-webview-app/app/build.gradle.kts`
- `versionCode` 递增
- `versionName` 更新
2. 更新版本说明(`android-webview-app/APP_CHANGELOG.md`
- 新版本条目写在顶部
3. 修改完成后提醒用户运行上传脚本:
- `bash /Users/jojo/Desktop/agents/正在修复中/agents/upload_android_apk.sh`
参考文档:`doc/android_app_release_and_update.md`

View File

@ -72,7 +72,7 @@ except ImportError:
from modules.container_monitor import collect_stats, inspect_state
from core.tool_config import TOOL_CATEGORIES
from utils.api_client import DeepSeekClient
from utils.context_manager import ContextManager
from utils.context_manager import ContextManager, AUTO_SHALLOW_PLACEHOLDER
from utils.tool_result_formatter import format_tool_result_for_context
from utils.logger import setup_logger
from config.model_profiles import (
@ -147,6 +147,7 @@ class MainTerminalContextMixin:
]
personalization_config = getattr(self.context_manager, "custom_personalization_config", None) or load_personalization_config(self.data_dir)
shallow_replace_enabled = bool(personalization_config.get("auto_shallow_compress_enabled", False)) if isinstance(personalization_config, dict) else False
skills_catalog = get_skills_catalog()
enabled_skills = merge_enabled_skills(
personalization_config.get("enabled_skills") if isinstance(personalization_config, dict) else None,
@ -208,6 +209,7 @@ class MainTerminalContextMixin:
# 添加对话历史保留完整结构包括tool_calls和tool消息
conversation = context["conversation"]
replaced_tool_count = 0
for idx, conv in enumerate(conversation):
metadata = conv.get("metadata") or {}
if conv["role"] == "assistant":
@ -226,6 +228,15 @@ class MainTerminalContextMixin:
messages.append(message)
elif conv["role"] == "tool":
if shallow_replace_enabled and metadata.get("auto_shallow_compacted"):
messages.append({
"role": "tool",
"content": AUTO_SHALLOW_PLACEHOLDER,
"tool_call_id": conv.get("tool_call_id", ""),
"name": conv.get("name", "")
})
replaced_tool_count += 1
continue
# Tool消息需要保留完整结构
images = conv.get("images") or metadata.get("images") or []
videos = conv.get("videos") or metadata.get("videos") or []
@ -281,6 +292,8 @@ class MainTerminalContextMixin:
"role": "system",
"content": disabled_notice
})
if shallow_replace_enabled:
print(f"[ContextCompression] build_messages 替换tool占位符: {replaced_tool_count}")
return messages
def load_prompt(self, name: str) -> str:

View File

@ -42,6 +42,8 @@ DEFAULT_PERSONALIZATION_CONFIG: Dict[str, Any] = {
"skill_hints_enabled": False, # Skill 提示系统开关(默认关闭)
"default_model": "kimi-k2.5",
"image_compression": "original", # original / 1080p / 720p / 540p
"auto_shallow_compress_enabled": False,
"auto_deep_compress_enabled": False,
"silent_tool_disable": False, # 禁用工具时不向模型插入提示
"enhanced_tool_display": True, # 增强工具显示
}
@ -186,6 +188,16 @@ def sanitize_personalization_payload(
elif base.get("image_compression") not in allowed_image_modes:
base["image_compression"] = "original"
if "auto_shallow_compress_enabled" in data:
base["auto_shallow_compress_enabled"] = bool(data.get("auto_shallow_compress_enabled"))
else:
base["auto_shallow_compress_enabled"] = bool(base.get("auto_shallow_compress_enabled", False))
if "auto_deep_compress_enabled" in data:
base["auto_deep_compress_enabled"] = bool(data.get("auto_deep_compress_enabled"))
else:
base["auto_deep_compress_enabled"] = bool(base.get("auto_deep_compress_enabled", False))
# 静默禁用工具提示
if "silent_tool_disable" in data:
base["silent_tool_disable"] = bool(data.get("silent_tool_disable"))

View File

@ -129,6 +129,7 @@ from .chat_flow_runtime import (
from .chat_flow_task_support import process_sub_agent_updates
from .chat_flow_tool_loop import execute_tool_calls
from .chat_flow_stream_loop import run_streaming_attempts
from .deep_compression import run_deep_compression
async def poll_sub_agent_completion(*, web_terminal, workspace, conversation_id, client_sid, username):
"""后台轮询子智能体完成状态,完成后触发新一轮对话"""
@ -547,7 +548,10 @@ async def handle_task_with_sender(terminal: WebTerminal, workspace: UserWorkspac
except Exception:
personal_config = {}
auto_title_enabled = personal_config.get("auto_generate_title", True)
if auto_title_enabled:
skip_auto_title_generation = bool(
(web_terminal.context_manager.conversation_metadata or {}).get("skip_auto_title_generation", False)
)
if auto_title_enabled and not skip_auto_title_generation:
conv_id = getattr(web_terminal.context_manager, "current_conversation_id", None)
socketio.start_background_task(
generate_conversation_title_background,
@ -556,6 +560,55 @@ async def handle_task_with_sender(terminal: WebTerminal, workspace: UserWorkspac
message,
username
)
# 自动深层压缩(用户输入后触发)
try:
personal_config = load_personalization_config(workspace.data_dir)
except Exception:
personal_config = {}
auto_deep_enabled = bool(personal_config.get("auto_deep_compress_enabled", False))
current_tokens_for_deep = web_terminal.context_manager.get_current_context_tokens(conversation_id)
if (
auto_deep_enabled
and current_tokens_for_deep > 150_000
and not web_terminal.context_manager.is_compression_in_progress()
):
web_terminal.context_manager._set_meta_flag("is_ultra_long_conversation", True)
sender('compression_state', {
"conversation_id": conversation_id,
"in_progress": True,
"mode": "auto",
"stage": "queued"
})
deep_result = await run_deep_compression(
web_terminal=web_terminal,
workspace=workspace,
conversation_id=conversation_id,
mode="auto",
sender=sender,
)
if not deep_result.get("success"):
sender('error', {
"message": deep_result.get("error") or "自动深层压缩失败",
"conversation_id": conversation_id,
})
else:
guide_message = (deep_result.get("guide_message") or "").strip()
if guide_message:
finalize_user_work_timer()
await handle_task_with_sender(
web_terminal,
workspace,
guide_message,
[],
sender,
client_sid,
username,
[]
)
return
finalize_user_work_timer()
return
# === 移除不在这里计算输入token改为在每次API调用前计算 ===
@ -911,11 +964,30 @@ async def handle_task_with_sender(terminal: WebTerminal, workspace: UserWorkspac
mark_force_thinking=mark_force_thinking,
get_stop_flag=get_stop_flag,
clear_stop_flag=clear_stop_flag,
workspace=workspace,
)
last_tool_call_time = tool_loop_result.get("last_tool_call_time", last_tool_call_time)
if tool_loop_result.get("stopped"):
finalize_user_work_timer()
return
if tool_loop_result.get("deep_compressed"):
deep_result = tool_loop_result.get("deep_result") or {}
guide_message = (deep_result.get("guide_message") or "").strip()
if deep_result.get("success") and guide_message:
finalize_user_work_timer()
await handle_task_with_sender(
web_terminal,
workspace,
guide_message,
[],
sender,
client_sid,
username,
[]
)
return
finalize_user_work_timer()
return
# 标记不再是第一次迭代
is_first_iteration = False

View File

@ -13,9 +13,11 @@ from .chat_flow_helpers import detect_tool_failure
from .chat_flow_runner_helpers import resolve_monitor_path, resolve_monitor_memory, capture_monitor_snapshot
from utils.tool_result_formatter import format_tool_result_for_context
from config import TOOL_CALL_COOLDOWN
from modules.personalization_manager import load_personalization_config
from .deep_compression import run_deep_compression
async def execute_tool_calls(*, web_terminal, tool_calls, sender, messages, client_sid: str, username: str, iteration: int, conversation_id: Optional[str], last_tool_call_time: float, process_sub_agent_updates, maybe_mark_failure_from_message, mark_force_thinking, get_stop_flag, clear_stop_flag):
async def execute_tool_calls(*, web_terminal, tool_calls, sender, messages, client_sid: str, username: str, iteration: int, conversation_id: Optional[str], last_tool_call_time: float, process_sub_agent_updates, maybe_mark_failure_from_message, mark_force_thinking, get_stop_flag, clear_stop_flag, workspace=None):
previous_tool_loop_active = getattr(web_terminal, "_tool_loop_active", False)
web_terminal._tool_loop_active = True
# 执行每个工具
@ -415,17 +417,11 @@ async def execute_tool_calls(*, web_terminal, tool_calls, sender, messages, clie
):
img_path = inj.get("path") if isinstance(inj, dict) else None
if img_path:
text_part = tool_result_content if isinstance(tool_result_content, str) else ""
tool_message_content = web_terminal.context_manager._build_content_with_images(
text_part,
[img_path]
)
tool_images = [img_path]
if metadata_payload is None:
metadata_payload = {}
metadata_payload["tool_image_path"] = img_path
sender('system_message', {
'content': f'系统已按模型请求将图片附加到工具结果: {img_path}'
'content': f'系统已记录图片路径(不再附带二进制数据): {img_path}'
})
# view_video: 将视频直接附加到 tool 结果中(不再插入 user 消息)
@ -439,19 +435,11 @@ async def execute_tool_calls(*, web_terminal, tool_calls, sender, messages, clie
):
video_path = inj.get("path") if isinstance(inj, dict) else None
if video_path:
text_part = tool_result_content if isinstance(tool_result_content, str) else ""
video_payload = [video_path]
tool_message_content = web_terminal.context_manager._build_content_with_images(
text_part,
[],
video_payload
)
tool_videos = [video_path]
if metadata_payload is None:
metadata_payload = {}
metadata_payload["tool_video_path"] = video_path
sender('system_message', {
'content': f'系统已按模型请求将视频附加到工具结果: {video_path}'
'content': f'系统已记录视频路径(不再附带二进制数据): {video_path}'
})
# 立即保存工具结果
@ -464,6 +452,26 @@ async def execute_tool_calls(*, web_terminal, tool_calls, sender, messages, clie
images=tool_images,
videos=tool_videos
)
try:
workspace_data_dir = getattr(workspace, "data_dir", None) if workspace else None
personal_config = load_personalization_config(workspace_data_dir) if workspace_data_dir else {}
except Exception:
personal_config = {}
auto_shallow_enabled = bool(personal_config.get("auto_shallow_compress_enabled", False))
auto_deep_enabled = bool(personal_config.get("auto_deep_compress_enabled", False))
compressed_count = 0
try:
compressed_count = int(
web_terminal.context_manager.on_tool_call_finished(function_name, enable_shallow=auto_shallow_enabled) or 0
)
except Exception as exc:
debug_log(f"[ContextCompression] 工具后浅压缩检测失败: {exc}")
if compressed_count > 0:
sender('shallow_compression', {
"conversation_id": conversation_id,
"compressed_count": compressed_count,
"keep_recent_tools": 15,
})
debug_log(f"💾 增量保存:工具结果 {function_name}")
if function_name != "wait_sub_agent":
system_message = result_data.get("system_message") if isinstance(result_data, dict) else None
@ -471,6 +479,40 @@ async def execute_tool_calls(*, web_terminal, tool_calls, sender, messages, clie
web_terminal._record_sub_agent_message(system_message, result_data.get("task_id"), inline=False)
maybe_mark_failure_from_message(web_terminal, system_message)
# 自动深层压缩(工具调用后触发)
current_context_tokens = web_terminal.context_manager.get_current_context_tokens(conversation_id)
if (
auto_deep_enabled
and current_context_tokens > 150_000
and not web_terminal.context_manager.is_compression_in_progress()
):
web_terminal.context_manager._set_meta_flag("is_ultra_long_conversation", True)
sender('compression_state', {
"conversation_id": conversation_id,
"in_progress": True,
"mode": "auto",
"stage": "queued"
})
deep_result = await run_deep_compression(
web_terminal=web_terminal,
workspace=workspace,
conversation_id=conversation_id,
mode="auto",
sender=sender,
)
if not deep_result.get("success"):
sender('error', {
"message": deep_result.get("error") or "自动深层压缩失败",
"conversation_id": conversation_id,
})
web_terminal._tool_loop_active = previous_tool_loop_active
return {
"stopped": False,
"deep_compressed": True,
"deep_result": deep_result,
"last_tool_call_time": last_tool_call_time
}
# 添加到消息历史用于API继续对话
messages.append({
"role": "tool",

View File

@ -82,6 +82,7 @@ from .conversation_stats import (
collect_upload_events,
summarize_upload_events,
)
from .deep_compression import run_deep_compression
conversation_bp = Blueprint('conversation', __name__)
@ -452,16 +453,24 @@ def get_conversation_messages(conversation_id, terminal: WebTerminal, workspace:
@api_login_required
@with_terminal
def compress_conversation(conversation_id, terminal: WebTerminal, workspace: UserWorkspace, username: str):
"""压缩指定对话的大体积消息,生成压缩版新对话"""
"""深层压缩指定对话,生成 compact 文件并切换到新对话。"""
try:
policy = resolve_admin_policy(get_current_user_record())
if policy.get("ui_blocks", {}).get("block_compress_conversation"):
return jsonify({"success": False, "error": "压缩对话已被管理员禁用"}), 403
normalized_id = conversation_id if conversation_id.startswith('conv_') else f"conv_{conversation_id}"
result = terminal.context_manager.compress_conversation(normalized_id)
result = asyncio.run(
run_deep_compression(
web_terminal=terminal,
workspace=workspace,
conversation_id=normalized_id,
mode="manual",
sender=None,
)
)
if not result.get("success"):
status_code = 404 if "不存在" in result.get("error", "") else 400
status_code = 404 if "不存在" in result.get("error", "") else (409 if result.get("in_progress") else 400)
return jsonify(result), status_code
new_conversation_id = result["compressed_conversation_id"]
@ -486,11 +495,36 @@ def compress_conversation(conversation_id, terminal: WebTerminal, workspace: Use
response_payload = {
"success": True,
"compressed_conversation_id": new_conversation_id,
"compressed_types": result.get("compressed_types", []),
"system_message": result.get("system_message"),
"compact_file": result.get("compact_file"),
"summary_failed": result.get("summary_failed", False),
"guide_message": result.get("guide_message"),
"load_result": load_result
}
guide_message = (result.get("guide_message") or "").strip()
if guide_message:
try:
from .tasks import task_manager
workspace_id = session.get("workspace_id") or "default"
task_rec = task_manager.create_chat_task(
username,
workspace_id,
guide_message,
images=[],
conversation_id=new_conversation_id,
videos=[],
)
response_payload["auto_task_started"] = True
response_payload["auto_task_id"] = task_rec.task_id
response_payload["auto_task_status"] = task_rec.status
except Exception as exc:
# 任务启动失败时前端会回退到手动发送 guide_message
debug_log(f"[Compression] 自动发送引导消息失败: {exc}")
response_payload["auto_task_started"] = False
response_payload["auto_task_error"] = str(exc)
else:
response_payload["auto_task_started"] = False
return jsonify(response_payload)
except Exception as e:
@ -502,6 +536,53 @@ def compress_conversation(conversation_id, terminal: WebTerminal, workspace: Use
}), 500
@conversation_bp.route('/api/conversations/<conversation_id>/compression_status', methods=['GET'])
@api_login_required
@with_terminal
def get_conversation_compression_status(conversation_id, terminal: WebTerminal, workspace: UserWorkspace, username: str):
try:
normalized_id = conversation_id if conversation_id.startswith('conv_') else f"conv_{conversation_id}"
data = terminal.context_manager.conversation_manager.load_conversation(normalized_id) or {}
meta = data.get("metadata", {}) or {}
return jsonify({
"success": True,
"data": {
"conversation_id": normalized_id,
"compression_in_progress": bool(meta.get("compression_in_progress", False)),
"compression_mode": meta.get("compression_mode"),
"compression_stage": meta.get("compression_stage"),
"compression_error": meta.get("compression_error"),
"compression_count": int(meta.get("compression_count", 0) or 0),
"compression_job_id": meta.get("compression_job_id"),
}
})
except Exception as e:
return jsonify({"success": False, "error": str(e)}), 500
@conversation_bp.route('/api/conversations/<conversation_id>/compression_cancel', methods=['POST'])
@api_login_required
@with_terminal
def cancel_conversation_compression(conversation_id, terminal: WebTerminal, workspace: UserWorkspace, username: str):
try:
normalized_id = conversation_id if conversation_id.startswith('conv_') else f"conv_{conversation_id}"
ok = terminal.context_manager.conversation_manager.update_conversation_metadata(
normalized_id,
{
"compression_in_progress": False,
"compression_mode": None,
"compression_stage": None,
"compression_resume_payload": None,
"compression_error": "用户切换对话导致压缩取消",
}
)
if not ok:
return jsonify({"success": False, "error": "对话不存在或取消失败"}), 404
return jsonify({"success": True, "conversation_id": normalized_id})
except Exception as e:
return jsonify({"success": False, "error": str(e)}), 500
@conversation_bp.route('/api/sub_agents', methods=['GET'])
@api_login_required
@with_terminal

295
server/deep_compression.py Normal file
View File

@ -0,0 +1,295 @@
from __future__ import annotations
import asyncio
import json
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
SUMMARY_PROMPT = (
"由于当前对话过长,系统正在自动压缩。请你基于已有上下文输出一份可继续执行的工作总结,要求:\n"
"1) 任务目标与用户真实诉求\n"
"2) 已完成工作(按时间顺序)\n"
"3) 关键决策与原因\n"
"4) 修改过的文件与核心变更点\n"
"5) 工具调用中的重要结果/错误与修复\n"
"6) 当前未完成事项与下一步计划(可直接执行)\n"
"7) 风险与注意事项\n"
"请使用中文,结构清晰,尽量具体,不要省略关键上下文。"
)
GUIDE_USER_MESSAGE_TEMPLATE = "当前对话已经被自动压缩,请阅读{path}并继续工作"
def _emit(sender, event_type: str, payload: Dict[str, Any]):
if not callable(sender):
return
try:
sender(event_type, payload)
except Exception:
pass
def _extract_text_only(content: Any) -> str:
if content is None:
return ""
if isinstance(content, str):
return content
if isinstance(content, list):
parts: List[str] = []
for item in content:
if not isinstance(item, dict):
continue
if item.get("type") == "text":
txt = item.get("text")
if isinstance(txt, str) and txt.strip():
parts.append(txt)
return "\n".join(parts).strip()
return str(content)
def _extract_tool_arg_map(messages: List[Dict[str, Any]]) -> Dict[str, Dict[str, Any]]:
tool_map: Dict[str, Dict[str, Any]] = {}
for msg in messages:
if msg.get("role") != "assistant":
continue
for tc in msg.get("tool_calls") or []:
tc_id = tc.get("id") or tc.get("tool_call_id")
if not tc_id:
continue
func = tc.get("function") or {}
name = func.get("name") or tc.get("name") or "unknown_tool"
args_raw = func.get("arguments")
args_obj: Any = args_raw
if isinstance(args_raw, str):
try:
args_obj = json.loads(args_raw)
except Exception:
args_obj = args_raw
tool_map[tc_id] = {"name": name, "arguments": args_obj}
return tool_map
def _collect_last_tool_entries(messages: List[Dict[str, Any]], limit: int = 5, max_content_chars: int = 3000) -> List[Dict[str, Any]]:
tool_arg_map = _extract_tool_arg_map(messages)
entries: List[Dict[str, Any]] = []
for msg in messages:
if msg.get("role") != "tool":
continue
tc_id = msg.get("tool_call_id") or msg.get("id")
mapping = tool_arg_map.get(tc_id, {})
content_text = _extract_text_only(msg.get("content"))
if len(content_text) > max_content_chars:
content_text = content_text[:max_content_chars] + "\n...(已截断)"
entries.append({
"tool_call_id": tc_id,
"tool_name": msg.get("name") or mapping.get("name") or "unknown_tool",
"arguments": mapping.get("arguments"),
"content": content_text,
})
return entries[-max(1, limit):]
def _collect_user_texts(messages: List[Dict[str, Any]]) -> List[str]:
result: List[str] = []
for msg in messages:
if msg.get("role") != "user":
continue
text = _extract_text_only(msg.get("content"))
if text.strip():
result.append(text.strip())
return result
async def _generate_summary(web_terminal, prompt: str, retries: int = 5) -> Tuple[str, Optional[str]]:
last_reason: Optional[str] = None
context = web_terminal.build_context()
messages = web_terminal.build_messages(context, prompt)
# build_messages 不会自动附加 user_input压缩总结必须显式注入
if prompt and isinstance(prompt, str):
messages = list(messages) + [{"role": "user", "content": prompt}]
for _ in range(max(1, retries)):
try:
response_text = ""
async for chunk in web_terminal.api_client.chat(messages, tools=None, stream=False):
if not isinstance(chunk, dict):
continue
choices = chunk.get("choices") or []
if choices:
msg = choices[0].get("message") or {}
content = msg.get("content")
if isinstance(content, str):
response_text = content
if response_text.strip():
return response_text.strip(), None
last_reason = "模型返回空内容"
except Exception as exc:
last_reason = str(exc)
await asyncio.sleep(0.2)
return f"生成总结失败({last_reason or '未知原因'}", last_reason
def _write_compact_file(project_path: Path, *, compression_index: int, summary_text: str, user_inputs: List[str], last_tools: List[Dict[str, Any]]) -> str:
compact_dir = project_path / "compact_result"
compact_dir.mkdir(parents=True, exist_ok=True)
filename = f"compact_{datetime.now().strftime('%Y%m%d_%H%M%S')}_{compression_index:03d}.md"
file_path = compact_dir / filename
lines: List[str] = [
f"# 对话已被第{compression_index}次压缩",
"",
"## 工作总结",
summary_text or "生成总结失败",
"",
"## 用户的所有输入(仅文字)",
]
if user_inputs:
for idx, text in enumerate(user_inputs, start=1):
lines.append(f"{idx}. {text}")
else:
lines.append("- (无)")
lines.extend(["", "## 最近5条工具调用参数 + 结果)"])
if last_tools:
for idx, item in enumerate(last_tools, start=1):
lines.extend([
f"### {idx}. {item.get('tool_name')}",
f"- tool_call_id: {item.get('tool_call_id')}",
"- 参数:",
"```json",
json.dumps(item.get("arguments"), ensure_ascii=False, indent=2),
"```",
"- 结果:",
"```text",
str(item.get("content") or ""),
"```",
"",
])
else:
lines.append("- (无)")
file_path.write_text("\n".join(lines).strip() + "\n", encoding="utf-8")
return str(file_path.relative_to(project_path))
async def run_deep_compression(
*,
web_terminal,
workspace,
conversation_id: str,
mode: str,
sender=None,
) -> Dict[str, Any]:
cm = web_terminal.context_manager
conv_data = cm.conversation_manager.load_conversation(conversation_id)
if not conv_data:
return {"success": False, "error": f"对话不存在: {conversation_id}"}
metadata = conv_data.get("metadata", {}) or {}
if metadata.get("compression_in_progress"):
return {"success": False, "error": "对话正在压缩中", "in_progress": True}
old_title = conv_data.get("title") or "未命名"
compression_count = int(metadata.get("compression_count", 0) or 0)
target_count = compression_count + 1
job_id = f"cmp_{datetime.now().strftime('%Y%m%d_%H%M%S_%f')}"
cm.set_compression_state(
in_progress=True,
mode=mode,
stage="generating_summary",
job_id=job_id,
resume_payload={"conversation_id": conversation_id, "mode": mode},
)
_emit(sender, "compression_state", {
"conversation_id": conversation_id,
"in_progress": True,
"mode": mode,
"stage": "generating_summary",
"job_id": job_id,
})
summary_text, summary_fail_reason = await _generate_summary(web_terminal, SUMMARY_PROMPT, retries=5)
if summary_fail_reason:
_emit(sender, "system_message", {"content": f"自动压缩总结失败,将使用失败占位文本:{summary_fail_reason}"})
cm.set_compression_state(
in_progress=True,
mode=mode,
stage="writing_compact",
job_id=job_id,
)
_emit(sender, "compression_state", {
"conversation_id": conversation_id,
"in_progress": True,
"mode": mode,
"stage": "writing_compact",
"job_id": job_id,
})
messages = conv_data.get("messages") or []
user_inputs = _collect_user_texts(messages)
last_tools = _collect_last_tool_entries(messages, limit=5, max_content_chars=3000)
relative_compact_path = _write_compact_file(
Path(workspace.project_path),
compression_index=target_count,
summary_text=summary_text,
user_inputs=user_inputs,
last_tools=last_tools,
)
cm.set_compression_state(
in_progress=True,
mode=mode,
stage="creating_new_conversation",
job_id=job_id,
)
new_conv_id = cm.conversation_manager.create_conversation(
project_path=str(workspace.project_path),
thinking_mode=bool(metadata.get("thinking_mode", False)),
run_mode=metadata.get("run_mode") or ("thinking" if metadata.get("thinking_mode") else "fast"),
initial_messages=[],
model_key=metadata.get("model_key"),
has_images=False,
has_videos=False,
metadata_overrides={
"compression_count": target_count,
"title_locked": True,
"skip_auto_title_generation": True,
}
)
cm.conversation_manager.update_conversation_title(new_conv_id, f"{old_title}对话 压缩后")
web_terminal.load_conversation(new_conv_id)
guide_message = GUIDE_USER_MESSAGE_TEMPLATE.format(path=relative_compact_path)
# 清理旧对话压缩状态
cm.conversation_manager.update_conversation_metadata(
conversation_id,
{
"compression_in_progress": False,
"compression_mode": None,
"compression_stage": None,
"compression_job_id": None,
"compression_error": summary_fail_reason,
"compression_resume_payload": None,
"is_ultra_long_conversation": True,
},
)
_emit(sender, "conversation_resolved", {
"conversation_id": new_conv_id,
"title": f"{old_title}对话 压缩后",
"created": True,
})
_emit(sender, "compression_finished", {
"source_conversation_id": conversation_id,
"conversation_id": new_conv_id,
"in_progress": False,
"compact_file": relative_compact_path,
"job_id": job_id,
})
return {
"success": True,
"compressed_conversation_id": new_conv_id,
"compact_file": relative_compact_path,
"summary_failed": bool(summary_fail_reason),
"guide_message": guide_message,
}

View File

@ -71,6 +71,15 @@ def get_status(terminal, workspace, username):
status['conversation']['title'] = current_conv_data.get('title', '未知对话')
status['conversation']['created_at'] = current_conv_data.get('created_at')
status['conversation']['updated_at'] = current_conv_data.get('updated_at')
meta = current_conv_data.get("metadata", {}) or {}
status['conversation']['compression'] = {
"in_progress": bool(meta.get("compression_in_progress", False)),
"mode": meta.get("compression_mode"),
"stage": meta.get("compression_stage"),
"error": meta.get("compression_error"),
"count": int(meta.get("compression_count", 0) or 0),
"job_id": meta.get("compression_job_id"),
}
except Exception as exc:
print(f"[Status] 获取当前对话信息失败: {exc}")
status['project_path'] = str(workspace.project_path)

View File

@ -262,17 +262,34 @@ class TaskManager:
except Exception:
pass
# 轮询模式需要把 context_manager 的回调切到当前任务 sender
# 否则 token_update 等事件只走 websocket前端任务轮询拿不到实时更新。
previous_ctx_callback = None
try:
if terminal and getattr(terminal, "context_manager", None):
previous_ctx_callback = getattr(terminal.context_manager, "_web_terminal_callback", None)
terminal.context_manager.set_web_terminal_callback(sender)
except Exception as exc:
debug_log(f"[Task] 设置上下文回调失败: {exc}")
# 将 task_id 作为 client_sid供 stop_flags 检测
run_chat_task_sync(
terminal=terminal,
message=rec.message,
images=images,
sender=sender,
client_sid=rec.task_id,
workspace=workspace,
username=username,
videos=videos,
)
try:
run_chat_task_sync(
terminal=terminal,
message=rec.message,
images=images,
sender=sender,
client_sid=rec.task_id,
workspace=workspace,
username=username,
videos=videos,
)
finally:
try:
if terminal and getattr(terminal, "context_manager", None):
terminal.context_manager.set_web_terminal_callback(previous_ctx_callback)
except Exception as exc:
debug_log(f"[Task] 恢复上下文回调失败: {exc}")
# 结束状态
canceled_flag = rec.stop_requested or stop_hint or bool(stop_flags.get(rec.task_id, {}).get("stop"))

View File

@ -155,14 +155,14 @@ export const computed = {
return !!this.policyUiBlocks.block_virtual_monitor;
},
displayLockEngaged() {
return false;
return !!this.compressionInProgress || !!this.compressing;
},
streamingUi() {
return this.streamingMessage || this.hasPendingToolActions();
},
composerBusy() {
const monitorLock = this.monitorIsLocked && this.chatDisplayMode === 'monitor';
return this.streamingUi || this.taskInProgress || monitorLock || this.stopRequested;
return this.streamingUi || this.taskInProgress || monitorLock || this.stopRequested || this.compressionInProgress || this.compressing;
},
composerHeroActive() {
return (this.blankHeroActive || this.blankHeroExiting) && !this.composerInteractionActive;

View File

@ -212,6 +212,31 @@ export const conversationMethods = {
return;
}
if ((this.compressionInProgress || this.compressing) && !force) {
const confirmed = await this.confirmAction({
title: '压缩进行中',
message: '对话正在压缩中,切换对话会导致压缩失败,确认要继续吗?',
confirmText: '确认',
cancelText: '取消'
});
if (!confirmed) {
this.suppressTitleTyping = false;
this.titleReady = true;
return;
}
try {
if (this.currentConversationId) {
await fetch(`/api/conversations/${this.currentConversationId}/compression_cancel`, { method: 'POST' });
}
} catch (e) {
console.warn('取消压缩失败:', e);
}
this.compressionInProgress = false;
this.compressing = false;
this.compressionMode = '';
this.compressionStage = '';
}
// 应用个性化设置中的默认模型和思考模式
try {
const { usePersonalizationStore } = await import('../../stores/personalization');
@ -354,6 +379,28 @@ export const conversationMethods = {
async createNewConversation() {
console.log('[DEBUG_AWAITING] ===== createNewConversation =====');
if (this.compressionInProgress || this.compressing) {
const confirmed = await this.confirmAction({
title: '压缩进行中',
message: '对话正在压缩中,切换对话会导致压缩失败,确认要继续吗?',
confirmText: '确认',
cancelText: '取消'
});
if (!confirmed) {
return;
}
try {
if (this.currentConversationId) {
await fetch(`/api/conversations/${this.currentConversationId}/compression_cancel`, { method: 'POST' });
}
} catch (e) {
console.warn('取消压缩失败:', e);
}
this.compressionInProgress = false;
this.compressing = false;
this.compressionMode = '';
this.compressionStage = '';
}
debugLog('创建新对话...');
traceLog('createNewConversation:start', {
@ -585,7 +632,7 @@ export const conversationMethods = {
if (response.ok && result.success) {
const newId = result.duplicate_conversation_id;
if (newId) {
this.currentConversationId = newId;
await this.loadConversation(newId, { force: true });
}
this.conversationsOffset = 0;

View File

@ -102,6 +102,14 @@ export const messageMethods = {
},
handleSendOrStop() {
if (this.compressionInProgress) {
this.uiPushToast({
title: '对话自动压缩中',
message: '当前不可发送/停止,请等待压缩完成',
type: 'warning'
});
return;
}
if (this.composerBusy) {
this.stopTask();
} else {
@ -112,6 +120,14 @@ export const messageMethods = {
async sendMessage() {
console.log('[DEBUG_AWAITING] ===== sendMessage =====');
if (this.compressionInProgress) {
this.uiPushToast({
title: '对话自动压缩中',
message: '压缩完成后才能继续发送消息',
type: 'warning'
});
return;
}
if (this.streamingUi) {
return;
}
@ -301,6 +317,14 @@ export const messageMethods = {
// 新增:停止任务方法
async stopTask() {
console.log('[DEBUG_AWAITING] ===== stopTask =====');
if (this.compressionInProgress) {
this.uiPushToast({
title: '对话自动压缩中',
message: '压缩进行中,当前不可停止任务',
type: 'warning'
});
return;
}
const canStop = this.composerBusy && !this.stopRequested;
if (!canStop) {
@ -428,6 +452,16 @@ export const messageMethods = {
}
this.compressing = true;
this.compressionInProgress = true;
this.compressionMode = 'manual';
this.compressionStage = 'requesting';
this.compressionError = '';
this.uiPushToast({
title: '压缩中',
message: '对话正在压缩,请稍候…',
type: 'info',
duration: 2500
});
try {
const response = await fetch(`/api/conversations/${this.currentConversationId}/compress`, {
@ -437,17 +471,46 @@ export const messageMethods = {
const result = await response.json();
if (response.ok && result.success) {
this.compressionStage = 'switching';
const newId = result.compressed_conversation_id;
if (newId) {
await this.loadConversation(newId, { force: true });
}
const guideMessage = (result.guide_message || '').trim();
const autoTaskStarted = !!result.auto_task_started;
const autoTaskId = result.auto_task_id;
if (autoTaskStarted && autoTaskId) {
const { useTaskStore } = await import('../../stores/task');
const taskStore = useTaskStore();
if (typeof this.clearProcessedEvents === 'function') {
this.clearProcessedEvents();
}
taskStore.resumeTask(autoTaskId, {
status: result.auto_task_status || 'running',
resetOffset: true,
eventHandler: (event) => this.handleTaskEvent(event)
});
this.taskInProgress = true;
if (typeof this.monitorShowPendingReply === 'function') {
this.monitorShowPendingReply();
}
} else if (guideMessage) {
await this.sendAutoUserMessage(guideMessage);
}
this.conversationsOffset = 0;
await this.loadConversationsList();
debugLog('对话压缩完成:', result);
this.uiPushToast({
title: '压缩完成',
message: '已生成压缩后的新对话',
type: 'success',
duration: 2200
});
} else {
const message = result.message || result.error || '压缩失败';
this.compressionError = message;
this.uiPushToast({
title: '压缩失败',
message,
@ -456,6 +519,7 @@ export const messageMethods = {
}
} catch (error) {
console.error('压缩对话异常:', error);
this.compressionError = error.message || '请稍后重试';
this.uiPushToast({
title: '压缩对话异常',
message: error.message || '请稍后重试',
@ -463,6 +527,9 @@ export const messageMethods = {
});
} finally {
this.compressing = false;
this.compressionInProgress = false;
this.compressionMode = '';
this.compressionStage = '';
}
},

View File

@ -145,6 +145,18 @@ export const resourceMethods = {
if (status && typeof status.has_videos !== 'undefined') {
this.conversationHasVideos = !!status.has_videos;
}
const compression = status?.conversation?.compression;
if (compression && typeof compression === 'object') {
this.compressionInProgress = !!compression.in_progress;
this.compressionMode = compression.mode || '';
this.compressionStage = compression.stage || '';
this.compressionError = compression.error || '';
} else {
this.compressionInProgress = false;
this.compressionMode = '';
this.compressionStage = '';
this.compressionError = '';
}
},
updateContainerStatus(status) {

View File

@ -51,7 +51,7 @@ export const taskPollingMethods = {
// 检查事件的 conversation_id 是否匹配当前对话
// 如果不匹配,忽略该事件(避免切换对话后旧任务的事件显示到新对话中)
if (eventData.conversation_id && this.currentConversationId) {
if (eventType !== 'shallow_compression' && eventData.conversation_id && this.currentConversationId) {
if (eventData.conversation_id !== this.currentConversationId) {
console.log(`[DEBUG_AWAITING] 忽略不匹配的事件`, {
eventType,
@ -151,6 +151,15 @@ export const taskPollingMethods = {
case 'conversation_resolved':
this.handleConversationResolved(eventData, eventIdx);
break;
case 'compression_state':
this.handleCompressionState(eventData, eventIdx);
break;
case 'compression_finished':
this.handleCompressionFinished(eventData, eventIdx);
break;
case 'shallow_compression':
this.handleShallowCompression(eventData, eventIdx);
break;
case 'user_message':
this.handleUserMessage(eventData, eventIdx);
@ -922,6 +931,43 @@ export const taskPollingMethods = {
}
},
handleCompressionState(data: any) {
if (!data || typeof data !== 'object') {
return;
}
this.compressionInProgress = !!data.in_progress;
this.compressionMode = data.mode || '';
this.compressionStage = data.stage || '';
if (!this.compressionInProgress) {
this.compressionError = data.error || '';
}
},
async handleCompressionFinished(data: any) {
this.compressionInProgress = false;
this.compressionMode = '';
this.compressionStage = '';
this.compressionError = '';
const newId = data?.conversation_id;
if (newId && newId !== this.currentConversationId) {
await this.loadConversation(newId, { force: true });
}
},
handleShallowCompression(data: any, eventIdx?: number) {
const count = Number(data?.compressed_count || 0);
if (count <= 0) {
return;
}
debugLog('[TaskPolling] 自动浅层压缩触发, idx:', eventIdx, data);
this.uiPushToast({
title: '自动浅层压缩',
message: `已自动压缩 ${count} 条较早工具结果`,
type: 'info',
duration: 2500
});
},
/**
*
*/

View File

@ -534,6 +534,14 @@ export const uiMethods = {
if (this.compressing || this.streamingMessage || !this.isConnected) {
return;
}
if (this.compressionInProgress) {
this.uiPushToast({
title: '对话自动压缩中',
message: '当前对话正在压缩,请稍后再试',
type: 'warning'
});
return;
}
if (this.isPolicyBlocked('block_compress_conversation', '压缩对话已被管理员禁用')) {
return;
}

View File

@ -40,6 +40,10 @@ export function dataState() {
// 对话压缩状态
compressing: false,
compressionInProgress: false,
compressionMode: '',
compressionStage: '',
compressionError: '',
skipConversationLoadedEvent: false,
skipConversationHistoryReload: false,
_scrollListenerReady: false,

View File

@ -483,6 +483,46 @@
<span>显示格式化的工具结果更清晰易读</span>
</label>
</div>
<div class="behavior-field">
<div class="behavior-field-header">
<span class="field-title">上下文压缩策略</span>
<p class="field-desc">可分别控制自动浅层压缩与自动深层压缩</p>
</div>
<label class="toggle-row">
<input
type="checkbox"
:checked="form.auto_shallow_compress_enabled"
@change="personalization.updateField({ key: 'auto_shallow_compress_enabled', value: $event.target.checked })"
/>
<span class="fancy-check" aria-hidden="true">
<svg viewBox="0 0 64 64">
<path
d="M 0 16 V 56 A 8 8 90 0 0 8 64 H 56 A 8 8 90 0 0 64 56 V 8 A 8 8 90 0 0 56 0 H 8 A 8 8 90 0 0 0 8 V 16 L 32 48 L 64 16 V 8 A 8 8 90 0 0 56 0 H 8 A 8 8 90 0 0 0 8 V 56 A 8 8 90 0 0 8 64 H 56 A 8 8 90 0 0 64 56 V 16"
pathLength="575.0541381835938"
class="fancy-path"
></path>
</svg>
</span>
<span>自动浅层压缩</span>
</label>
<label class="toggle-row">
<input
type="checkbox"
:checked="form.auto_deep_compress_enabled"
@change="personalization.updateField({ key: 'auto_deep_compress_enabled', value: $event.target.checked })"
/>
<span class="fancy-check" aria-hidden="true">
<svg viewBox="0 0 64 64">
<path
d="M 0 16 V 56 A 8 8 90 0 0 8 64 H 56 A 8 8 90 0 0 64 56 V 8 A 8 8 90 0 0 56 0 H 8 A 8 8 90 0 0 0 8 V 16 L 32 48 L 64 16 V 8 A 8 8 90 0 0 56 0 H 8 A 8 8 90 0 0 0 8 V 56 A 8 8 90 0 0 8 64 H 56 A 8 8 90 0 0 64 56 V 16"
pathLength="575.0541381835938"
class="fancy-path"
></path>
</svg>
</span>
<span>自动深层压缩</span>
</label>
</div>
<div class="behavior-field">
<div class="behavior-field-header">
<span class="field-title">堆叠块显示模式</span>

View File

@ -22,6 +22,8 @@ interface PersonalForm {
default_run_mode: RunMode | null;
default_model: string | null;
image_compression: string;
auto_shallow_compress_enabled: boolean;
auto_deep_compress_enabled: boolean;
}
interface LiquidGlassPosition {
@ -79,7 +81,9 @@ const defaultForm = (): PersonalForm => ({
disabled_tool_categories: [],
default_run_mode: null,
default_model: 'kimi-k2.5',
image_compression: 'original'
image_compression: 'original',
auto_shallow_compress_enabled: false,
auto_deep_compress_enabled: false
});
const defaultExperimentState = (): ExperimentState => ({
@ -228,7 +232,9 @@ export const usePersonalizationStore = defineStore('personalization', {
? data.default_run_mode as RunMode
: null,
default_model: typeof data.default_model === 'string' ? data.default_model : fallbackModel,
image_compression: typeof data.image_compression === 'string' ? data.image_compression : 'original'
image_compression: typeof data.image_compression === 'string' ? data.image_compression : 'original',
auto_shallow_compress_enabled: !!data.auto_shallow_compress_enabled,
auto_deep_compress_enabled: !!data.auto_deep_compress_enabled
};
this.clearFeedback();
},

View File

@ -45,6 +45,21 @@ except ImportError:
from config.model_profiles import get_model_prompt_replacements
from utils.conversation_manager import ConversationManager
AUTO_SHALLOW_PLACEHOLDER = "过早的工具结果已经被自动压缩"
AUTO_SHALLOW_TOOL_WHITELIST = {
"write_file",
"read_file",
"edit_file",
"terminal_input",
"terminal_snapshot",
"web_search",
"extract_webpage",
"run_python",
"run_command",
"view_image",
"view_video",
}
class ContextManager:
def __init__(self, project_path: str, data_dir: Optional[str] = None):
self.project_path = Path(project_path).resolve()
@ -66,6 +81,7 @@ class ContextManager:
self.conversation_metadata: Dict[str, Any] = {}
self.project_snapshot: Optional[Dict[str, Any]] = None
self._host_runtime_cache: Optional[Dict[str, str]] = None
self._shallow_compact_round: int = 0
# 新增:对话持久化管理器
self.conversation_manager = ConversationManager(base_dir=self.data_dir)
@ -429,8 +445,10 @@ class ContextManager:
prompt_tokens = int(usage.get("prompt_tokens") or 0)
completion_tokens = int(usage.get("completion_tokens") or 0)
total_tokens = int(usage.get("total_tokens") or (prompt_tokens + completion_tokens))
# 当前上下文长度优先取专用字段;缺失时回退到 prompt_tokens
current_context_tokens = int(usage.get("current_context_tokens") or prompt_tokens)
except (TypeError, ValueError):
prompt_tokens = completion_tokens = total_tokens = 0
prompt_tokens = completion_tokens = total_tokens = current_context_tokens = 0
try:
self._increment_workspace_token_totals(prompt_tokens, completion_tokens, total_tokens)
@ -446,7 +464,8 @@ class ContextManager:
self.current_conversation_id,
prompt_tokens,
completion_tokens,
total_tokens
total_tokens,
current_context_tokens=current_context_tokens,
)
if success:
@ -481,6 +500,144 @@ class ContextManager:
if not stats:
return 0
return stats.get("current_context_tokens", 0)
def _get_meta_flag(self, key: str, default: Any = None) -> Any:
return (self.conversation_metadata or {}).get(key, default)
def _set_meta_flag(self, key: str, value: Any, save: bool = True):
self.conversation_metadata[key] = value
if save and self.current_conversation_id:
try:
self.conversation_manager.update_conversation_metadata(
self.current_conversation_id,
{key: value}
)
except Exception as exc:
print(f"[ContextCompression] 保存 metadata 失败 {key}: {exc}")
def is_compression_in_progress(self) -> bool:
return bool(self._get_meta_flag("compression_in_progress", False))
def set_compression_state(
self,
*,
in_progress: bool,
mode: Optional[str] = None,
stage: Optional[str] = None,
error: Optional[str] = None,
resume_payload: Optional[Dict[str, Any]] = None,
job_id: Optional[str] = None,
):
updates: Dict[str, Any] = {
"compression_in_progress": bool(in_progress),
"compression_mode": mode if in_progress else None,
"compression_stage": stage if in_progress else None,
"compression_error": error if in_progress else None,
"compression_resume_payload": resume_payload if in_progress else None,
"compression_job_id": job_id if in_progress else None,
}
for k, v in updates.items():
self.conversation_metadata[k] = v
if self.current_conversation_id:
try:
self.conversation_manager.update_conversation_metadata(self.current_conversation_id, updates)
except Exception as exc:
print(f"[ContextCompression] 更新压缩状态失败: {exc}")
def on_tool_call_finished(self, tool_name: Optional[str] = None, *, enable_shallow: bool = True) -> int:
"""每次工具调用完成后触发:更新计数并执行自动浅压缩。返回本轮浅压缩条数。"""
if not self.current_conversation_id:
return 0
# 兼容历史对话tool_call_count 可能从 0 开始,但历史里已存在大量 tool 结果。
# 这里优先对齐到“当前历史中的 tool 消息数量”,避免长期无法触发浅压缩。
meta_tool_count = int(self._get_meta_flag("tool_call_count", 0) or 0)
history_tool_count = 0
try:
history_tool_count = sum(
1 for msg in (self.conversation_history or [])
if isinstance(msg, dict) and msg.get("role") == "tool"
)
except Exception:
history_tool_count = 0
tool_count = max(meta_tool_count + 1, history_tool_count)
self._set_meta_flag("tool_call_count", tool_count, save=False)
current_tokens = self.get_current_context_tokens(self.current_conversation_id)
if current_tokens > 80_000 and not self._get_meta_flag("is_long_conversation", False):
self._set_meta_flag("is_long_conversation", True, save=False)
if current_tokens > 150_000 and not self._get_meta_flag("is_ultra_long_conversation", False):
self._set_meta_flag("is_ultra_long_conversation", True, save=False)
# 每 10 次工具调用触发一次浅压缩(当 long 已标记)
last_checkpoint = int(self._get_meta_flag("last_shallow_compress_tool_count", 0) or 0)
should_try_shallow = (
bool(enable_shallow)
and bool(self._get_meta_flag("is_long_conversation", False))
and (tool_count - last_checkpoint >= 10)
)
changed = False
compressed_count = 0
if should_try_shallow:
compressed = self._run_auto_shallow_compression(batch_size=10, keep_recent_tools=15)
self._set_meta_flag("last_shallow_compress_tool_count", tool_count, save=False)
compressed_count = max(0, int(compressed or 0))
changed = compressed > 0
if self.current_conversation_id:
updates = {
"tool_call_count": tool_count,
"is_long_conversation": bool(self._get_meta_flag("is_long_conversation", False)),
"is_ultra_long_conversation": bool(self._get_meta_flag("is_ultra_long_conversation", False)),
"last_shallow_compress_tool_count": int(self._get_meta_flag("last_shallow_compress_tool_count", 0) or 0),
}
try:
self.conversation_manager.update_conversation_metadata(self.current_conversation_id, updates)
except Exception as exc:
print(f"[ContextCompression] 写入计数失败: {exc}")
if changed:
self.auto_save_conversation(force=True)
return compressed_count
def _run_auto_shallow_compression(self, batch_size: int = 10, keep_recent_tools: int = 15) -> int:
"""浅压缩:仅打标记,不修改原文。返回本轮标记条数。"""
history = self.conversation_history or []
if not history:
return 0
tool_indices = [idx for idx, msg in enumerate(history) if msg.get("role") == "tool"]
protected_indices = set(tool_indices[-max(0, keep_recent_tools):]) if keep_recent_tools > 0 else set()
candidates: List[int] = []
for idx, msg in enumerate(history):
if msg.get("role") != "tool":
continue
if idx in protected_indices:
continue
tool_name = (msg.get("name") or "").strip()
if tool_name not in AUTO_SHALLOW_TOOL_WHITELIST:
continue
metadata = msg.get("metadata") or {}
if metadata.get("auto_shallow_compacted"):
continue
candidates.append(idx)
if len(candidates) >= max(1, batch_size):
break
if not candidates:
return 0
self._shallow_compact_round += 1
now = datetime.now().isoformat()
for idx in candidates:
msg = history[idx]
metadata = msg.get("metadata") or {}
metadata["auto_shallow_compacted"] = True
metadata["auto_shallow_compacted_at"] = now
metadata["auto_shallow_compact_round"] = self._shallow_compact_round
msg["metadata"] = metadata
history[idx] = msg
self.conversation_history = history
return len(candidates)
# ===========================================
# 新增:对话持久化相关方法
@ -1765,6 +1922,18 @@ class ContextManager:
if workspace_system:
messages.append({"role": "system", "content": workspace_system})
# 浅压缩替换开关:仅当个人空间开启自动浅层压缩时,才将已打标 tool 结果替换为占位符
shallow_replace_enabled = False
try:
personalization_config = (
getattr(self, "custom_personalization_config", None)
or load_personalization_config(self.data_dir)
)
shallow_replace_enabled = bool(personalization_config.get("auto_shallow_compress_enabled", False))
except Exception:
shallow_replace_enabled = False
replaced_tool_count = 0
# 添加对话历史
for conv in context["conversation"]:
if conv["role"] == "assistant":
@ -1779,6 +1948,17 @@ class ContextManager:
message["tool_calls"] = conv["tool_calls"]
messages.append(message)
elif conv["role"] == "tool":
conv_meta = conv.get("metadata") or {}
if shallow_replace_enabled and conv_meta.get("auto_shallow_compacted"):
message = {
"role": "tool",
"content": AUTO_SHALLOW_PLACEHOLDER,
"tool_call_id": conv.get("tool_call_id", ""),
"name": conv.get("name", "")
}
messages.append(message)
replaced_tool_count += 1
continue
images = conv.get("images") or (conv.get("metadata") or {}).get("images") or []
videos = conv.get("videos") or (conv.get("metadata") or {}).get("videos") or []
content_value = conv.get("content")
@ -1803,6 +1983,8 @@ class ContextManager:
"role": conv["role"],
"content": content_payload
})
if shallow_replace_enabled:
print(f"[ContextCompression] build_messages 替换tool占位符: {replaced_tool_count}")
# 添加终端内容(如果有的话)
# 这里需要从参数传入或获取

View File

@ -3,6 +3,8 @@
import json
import os
import time
import tempfile
import threading
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional, Any
@ -42,6 +44,7 @@ class ConversationManager:
self.base_dir = Path(base_dir).expanduser().resolve() if base_dir else Path(DATA_DIR).resolve()
self.conversations_dir = self.base_dir / "conversations"
self.index_file = self.conversations_dir / "index.json"
self._io_lock = threading.RLock()
self.current_conversation_id: Optional[str] = None
self.workspace_root = Path(__file__).resolve().parents[1]
self._ensure_directories()
@ -60,11 +63,47 @@ class ConversationManager:
def _iter_conversation_files(self, sort_by_mtime: bool = True):
"""遍历对话文件(排除索引文件),可按修改时间降序排序。"""
files = [p for p in self.conversations_dir.glob("*.json") if p != self.index_file]
files = []
for p in self.conversations_dir.glob("*.json"):
if p == self.index_file:
continue
stem = p.stem or ""
# 跳过索引备份/损坏文件,避免被误当作对话文件参与重建
if stem == "index" or stem.startswith("index_corrupt_"):
continue
# 仅纳入标准对话文件,避免其他 json 干扰索引
if not stem.startswith("conv_"):
continue
files.append(p)
if sort_by_mtime:
files.sort(key=lambda p: p.stat().st_mtime, reverse=True)
return files
def _atomic_write_json(self, target_file: Path, payload: Dict[str, Any]):
"""原子写入 JSON使用唯一临时文件避免并发写同名 .tmp 产生竞态。"""
target_file.parent.mkdir(parents=True, exist_ok=True)
temp_path: Optional[Path] = None
try:
with tempfile.NamedTemporaryFile(
mode="w",
encoding="utf-8",
dir=str(target_file.parent),
prefix=f".{target_file.name}.",
suffix=".tmp",
delete=False,
) as fh:
temp_path = Path(fh.name)
json.dump(payload, fh, ensure_ascii=False, indent=2)
fh.flush()
os.fsync(fh.fileno())
os.replace(str(temp_path), str(target_file))
finally:
if temp_path and temp_path.exists():
try:
temp_path.unlink()
except Exception:
pass
def _rebuild_index_from_files(self, max_count: Optional[int] = None) -> Dict:
"""
从现有对话文件重建索引
@ -156,7 +195,7 @@ class ConversationManager:
except (json.JSONDecodeError, Exception) as e:
print(f"⚠️ 加载对话索引失败,将尝试重建: {e}")
backup_path = self.index_file.with_name(
f"{self.index_file.stem}_corrupt_{int(time.time())}{self.index_file.suffix}"
f"{self.index_file.stem}_corrupt_{int(time.time() * 1000)}{self.index_file.suffix}"
)
try:
if self.index_file.exists():
@ -164,7 +203,8 @@ class ConversationManager:
print(f"🗄️ 已备份损坏的索引文件到: {backup_path.name}")
except Exception as backup_exc:
print(f"⚠️ 备份损坏索引文件失败: {backup_exc}")
rebuilt = self._rebuild_index_from_files(max_count=max_rebuild)
# 索引损坏场景优先全量重建,避免仅重建部分导致“对话丢失”错觉
rebuilt = self._rebuild_index_from_files(max_count=None)
if rebuilt:
self._save_index(rebuilt)
if ensure_integrity:
@ -174,17 +214,10 @@ class ConversationManager:
def _save_index(self, index: Dict):
"""保存对话索引"""
temp_file = self.index_file.with_suffix(self.index_file.suffix + ".tmp")
try:
with open(temp_file, 'w', encoding='utf-8') as f:
json.dump(index, f, ensure_ascii=False, indent=2)
os.replace(temp_file, self.index_file)
with self._io_lock:
self._atomic_write_json(self.index_file, index)
except Exception as e:
try:
if temp_file.exists():
temp_file.unlink()
except Exception:
pass
print(f"⌘ 保存对话索引失败: {e}")
def _ensure_index_covering(self, limit: int, offset: int) -> Dict:
@ -237,14 +270,12 @@ class ConversationManager:
return "新对话"
def _count_tools_in_messages(self, messages: List[Dict]) -> int:
"""统计消息中的工具调用数量"""
"""统计消息中的工具调用数量(仅统计 assistant.tool_calls"""
tool_count = 0
for msg in messages:
if msg.get("role") == "assistant" and "tool_calls" in msg:
tool_calls = msg.get("tool_calls", [])
tool_count += len(tool_calls) if isinstance(tool_calls, list) else 0
elif msg.get("role") == "tool":
tool_count += 1
return tool_count
def _prepare_project_path_metadata(self, project_path: Optional[str]) -> Dict[str, Optional[str]]:
@ -317,7 +348,8 @@ class ConversationManager:
initial_messages: List[Dict] = None,
model_key: Optional[str] = None,
has_images: bool = False,
has_videos: bool = False
has_videos: bool = False,
metadata_overrides: Optional[Dict[str, Any]] = None
) -> str:
"""
创建新对话
@ -336,6 +368,38 @@ class ConversationManager:
# 创建对话数据
path_metadata = self._prepare_project_path_metadata(project_path)
normalized_mode = run_mode if run_mode in {"fast", "thinking", "deep"} else ("thinking" if thinking_mode else "fast")
metadata = {
"project_path": path_metadata["project_path"],
"project_relative_path": path_metadata["project_relative_path"],
"thinking_mode": thinking_mode,
"run_mode": normalized_mode,
"model_key": model_key,
"has_images": has_images,
"has_videos": has_videos,
# 首次对话尚未生成文件树快照,待首次用户消息时填充
"project_file_tree": None,
"project_statistics": None,
"project_snapshot_at": None,
"total_messages": len(messages),
"total_tools": self._count_tools_in_messages(messages),
"status": "active",
# 压缩相关字段
"compression_count": 0,
"is_long_conversation": False,
"is_ultra_long_conversation": False,
"tool_call_count": 0,
"last_shallow_compress_tool_count": 0,
"compression_in_progress": False,
"compression_mode": None,
"compression_stage": None,
"compression_job_id": None,
"compression_error": None,
"compression_resume_payload": None,
"skip_auto_title_generation": False,
}
if isinstance(metadata_overrides, dict):
metadata.update(metadata_overrides)
conversation_data = {
"id": conversation_id,
"title": self._extract_title_from_messages(messages),
@ -343,23 +407,7 @@ class ConversationManager:
"updated_at": datetime.now().isoformat(),
"messages": messages,
"todo_list": None,
"metadata": {
"project_path": path_metadata["project_path"],
"project_relative_path": path_metadata["project_relative_path"],
"thinking_mode": thinking_mode,
"run_mode": normalized_mode,
"model_key": model_key,
"has_images": has_images,
"has_videos": has_videos,
"has_videos": has_videos,
# 首次对话尚未生成文件树快照,待首次用户消息时填充
"project_file_tree": None,
"project_statistics": None,
"project_snapshot_at": None,
"total_messages": len(messages),
"total_tools": self._count_tools_in_messages(messages),
"status": "active"
},
"metadata": metadata,
"token_statistics": self._initialize_token_statistics() # 新增
}
@ -374,6 +422,25 @@ class ConversationManager:
return conversation_id
def update_conversation_metadata(self, conversation_id: str, updates: Dict[str, Any]) -> bool:
"""合并更新对话 metadata。"""
if not conversation_id or not isinstance(updates, dict):
return False
try:
data = self.load_conversation(conversation_id)
if not data:
return False
metadata = data.get("metadata", {}) or {}
metadata.update(updates)
data["metadata"] = metadata
data["updated_at"] = datetime.now().isoformat()
self._save_conversation_file(conversation_id, data)
self._update_index(conversation_id, data)
return True
except Exception as exc:
print(f"⚠️ 更新对话 metadata 失败 {conversation_id}: {exc}")
return False
def update_conversation_title(self, conversation_id: str, title: str) -> bool:
"""更新对话标题并刷新索引。"""
if not conversation_id or not title:
@ -399,19 +466,12 @@ class ConversationManager:
def _save_conversation_file(self, conversation_id: str, data: Dict):
"""保存对话文件"""
file_path = self._get_conversation_file_path(conversation_id)
temp_file = file_path.with_suffix(file_path.suffix + ".tmp")
try:
# 确保Token统计数据有效
data = self._validate_token_statistics(data)
with open(temp_file, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
os.replace(temp_file, file_path)
with self._io_lock:
self._atomic_write_json(file_path, data)
except Exception as e:
try:
if temp_file.exists():
temp_file.unlink()
except Exception:
pass
print(f"⌘ 保存对话文件失败 {conversation_id}: {e}")
def _update_index(self, conversation_id: str, conversation_data: Dict):
@ -662,6 +722,15 @@ class ConversationManager:
data["metadata"] = metadata
self._save_conversation_file(conversation_id, data)
print(f"🔧 为对话 {conversation_id} 补齐项目快照字段")
# 兼容历史口径:旧版本 total_tools 可能把 role=tool 也计入,导致翻倍。
expected_total_tools = self._count_tools_in_messages(data.get("messages") or [])
if int(metadata.get("total_tools", 0) or 0) != expected_total_tools:
metadata["total_tools"] = expected_total_tools
data["metadata"] = metadata
self._save_conversation_file(conversation_id, data)
self._update_index(conversation_id, data)
print(f"🔧 修正对话 {conversation_id} 的 total_tools 统计为 {expected_total_tools}")
# 回填缺失的模型字段:从最近的助手消息元数据推断
if metadata.get("model_key") is None:
@ -688,7 +757,8 @@ class ConversationManager:
conversation_id: str,
input_tokens: int,
output_tokens: int,
total_tokens: int
total_tokens: int,
current_context_tokens: Optional[int] = None,
) -> bool:
"""
更新对话的Token统计
@ -698,6 +768,7 @@ class ConversationManager:
input_tokens: 输入Token数量
output_tokens: 输出Token数量
total_tokens: 本次请求的总Token数量prompt+completion
current_context_tokens: 当前上下文长度用于压缩阈值判断
Returns:
bool: 更新是否成功
@ -717,7 +788,10 @@ class ConversationManager:
token_stats["total_input_tokens"] = token_stats.get("total_input_tokens", 0) + input_tokens
token_stats["total_output_tokens"] = token_stats.get("total_output_tokens", 0) + output_tokens
token_stats["total_tokens"] = token_stats.get("total_tokens", 0) + total_tokens
token_stats["current_context_tokens"] = total_tokens
if current_context_tokens is None:
# 兼容旧调用:未显式传入时,默认以输入 token 作为当前上下文长度
current_context_tokens = input_tokens
token_stats["current_context_tokens"] = max(0, int(current_context_tokens or 0))
token_stats["updated_at"] = datetime.now().isoformat()
# 保存更新