311 lines
11 KiB
Python
311 lines
11 KiB
Python
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,
|
||
}
|