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]], latest_user_input: str, ) -> 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("- （无）") lines.extend(["", "## 用户最新的一次输入"]) if (latest_user_input or "").strip(): lines.append(latest_user_input.strip()) 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) latest_user_input = user_inputs[-1] if user_inputs else "" 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, latest_user_input=latest_user_input, ) 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, }