680 lines
25 KiB
Python
680 lines
25 KiB
Python
import asyncio
|
||
import json
|
||
from datetime import datetime, timezone
|
||
|
||
from fastapi import APIRouter, Depends, Query
|
||
from fastapi.responses import StreamingResponse
|
||
from sqlmodel import Session, select, func
|
||
|
||
from ..database import get_session, engine
|
||
from ..models import AIConfig, ChatMessage, Novel, Skill
|
||
from ..schemas import ChatRequest, ChatMessageRead, ChatMessageList
|
||
from ..services.ai_provider import (
|
||
stream_chat, simple_completion,
|
||
build_assistant_message, build_tool_results_messages, ToolCall,
|
||
)
|
||
from ..services.tools import get_openai_tools, get_anthropic_tools, execute_tool, get_tools_description, SPECIAL_TOOLS
|
||
|
||
router = APIRouter(prefix="/api/chat", tags=["chat"])
|
||
|
||
# 最大工具调用轮次
|
||
MAX_TOOL_ROUNDS = 8
|
||
|
||
# 工具名称到中文描述
|
||
TOOL_LABELS: dict[str, str] = {
|
||
"get_novel_info": "查看小说信息",
|
||
"update_novel_info": "更新小说信息",
|
||
"list_outlines": "查询大纲",
|
||
"get_outline_detail": "查看大纲详情",
|
||
"create_outline": "创建大纲",
|
||
"update_outline": "更新大纲",
|
||
"delete_outline": "删除大纲",
|
||
"reorder_outlines": "调整大纲排序",
|
||
"get_world_setting": "查询世界观",
|
||
"save_world_setting": "保存世界观",
|
||
"list_characters": "查询角色",
|
||
"get_character_detail": "查看角色详情",
|
||
"create_character": "创建角色",
|
||
"update_character": "更新角色",
|
||
"delete_character": "删除角色",
|
||
"list_chapters": "查询章节",
|
||
"get_chapter_detail": "查看章节详情",
|
||
"create_chapter": "创建章节",
|
||
"update_chapter": "更新章节",
|
||
"delete_chapter": "删除章节",
|
||
"list_files": "查询文件",
|
||
"read_file": "读取文件",
|
||
"dispatch_subagent": "派遣子Agent",
|
||
}
|
||
|
||
# 子 Agent 最大轮次
|
||
MAX_SUBAGENT_ROUNDS = 5
|
||
|
||
|
||
def _build_system_prompt(
|
||
novel_id: int | None,
|
||
session: Session,
|
||
page_context: str | None = None,
|
||
tools_enabled: bool = False,
|
||
skill: Skill | None = None,
|
||
allowed_tools: list[str] | None = None,
|
||
) -> str:
|
||
"""根据小说上下文、Skill 和工具配置构建 system prompt"""
|
||
parts = [
|
||
"你是 Writing Red Dot 写作助手，专注于小说创作领域。",
|
||
"你擅长：构思情节、塑造角色、打磨文笔、设计世界观、规划大纲结构。",
|
||
]
|
||
|
||
# 注入 Skill 提示词
|
||
if skill and skill.system_prompt:
|
||
parts.append(f"\n## 当前技能：{skill.name}\n{skill.system_prompt}")
|
||
|
||
if novel_id:
|
||
novel = session.get(Novel, novel_id)
|
||
if novel:
|
||
parts.append(f"\n当前小说：《{novel.title}》（ID: {novel.id}）")
|
||
parts.append(f"小说简介：{novel.description or '暂无'}")
|
||
|
||
if page_context:
|
||
parts.append(f"\n用户当前正在查看：{page_context}。请结合当前页面场景提供针对性建议。")
|
||
|
||
if tools_enabled:
|
||
# 动态生成工具能力描述（根据 allowed_tools 过滤）
|
||
tools_desc = get_tools_description(allowed_tools)
|
||
if tools_desc:
|
||
parts.append(f"\n## 你的工具能力\n{tools_desc}")
|
||
if not novel_id:
|
||
parts.append("\n注意：工具操作需要关联到具体小说。如果用户需要使用工具，请提示他们先进入某本小说。")
|
||
|
||
# 注入可用 Skill 列表（用于 dispatch_subagent）
|
||
from sqlmodel import or_
|
||
skills = session.exec(
|
||
select(Skill).where(
|
||
or_(Skill.novel_id == novel_id, Skill.novel_id.is_(None)) # type: ignore
|
||
)
|
||
).all()
|
||
if skills:
|
||
skill_lines = ["\n## 可用子Agent（Skill）"]
|
||
skill_lines.append("你可以使用 dispatch_subagent 工具派遣以下子Agent处理子任务：")
|
||
for sk in skills:
|
||
skill_lines.append(f"- skill_id={sk.id} {sk.name}: {sk.description}")
|
||
skill_lines.append("当任务复杂时，考虑将子任务分派给专门的子Agent处理。")
|
||
parts.append("\n".join(skill_lines))
|
||
|
||
return "\n".join(parts)
|
||
|
||
|
||
def _get_tools(provider: str, allowed: list[str] | None = None) -> list[dict]:
|
||
if provider == "anthropic":
|
||
return get_anthropic_tools(allowed)
|
||
return get_openai_tools(allowed)
|
||
|
||
|
||
def _sse(data: dict) -> str:
|
||
return f"data: {json.dumps(data, ensure_ascii=False)}\n\n"
|
||
|
||
|
||
def _save_assistant_message(novel_id: int | None, full_response: str, tool_call_logs: list[str]):
|
||
"""保存 AI 回复到 DB。工具调用日志和回复文本分别存为独立记录。"""
|
||
with Session(engine) as s:
|
||
# 工具调用日志作为单独一条记录
|
||
if tool_call_logs:
|
||
tool_summary = "\n".join(tool_call_logs)
|
||
s.add(ChatMessage(
|
||
novel_id=novel_id,
|
||
role="assistant",
|
||
content=f"[以下操作已执行]\n{tool_summary}",
|
||
))
|
||
# AI 回复文本作为单独一条记录
|
||
if full_response.strip():
|
||
s.add(ChatMessage(
|
||
novel_id=novel_id,
|
||
role="assistant",
|
||
content=full_response,
|
||
))
|
||
s.commit()
|
||
|
||
|
||
async def _run_subagent(
|
||
queue: asyncio.Queue,
|
||
config: AIConfig,
|
||
skill: Skill,
|
||
task: str,
|
||
novel_id: int,
|
||
subagent_id: str = "",
|
||
) -> str:
|
||
"""运行子 Agent：独立 LLM 循环，通过 queue 发送事件。返回最终结果文本。
|
||
subagent_id 用于前端区分并行运行的多个子 Agent。"""
|
||
# 解析 Skill 的工具白名单
|
||
sub_allowed: list[str] | None = None
|
||
if skill.allowed_tools:
|
||
try:
|
||
parsed = json.loads(skill.allowed_tools)
|
||
if isinstance(parsed, list) and len(parsed) > 0:
|
||
sub_allowed = [t for t in parsed if t != "dispatch_subagent"]
|
||
except (json.JSONDecodeError, TypeError):
|
||
pass
|
||
|
||
# 构建子 Agent 系统提示词
|
||
sub_system_parts = [
|
||
f"你是子Agent「{skill.name}」。",
|
||
f"你的职责：{skill.description}" if skill.description else "",
|
||
]
|
||
if skill.system_prompt:
|
||
sub_system_parts.append(skill.system_prompt)
|
||
|
||
with Session(engine) as s:
|
||
novel = s.get(Novel, novel_id)
|
||
if novel:
|
||
sub_system_parts.append(f"\n当前小说：《{novel.title}》（ID: {novel.id}）")
|
||
sub_tools_desc = get_tools_description(sub_allowed)
|
||
if sub_tools_desc:
|
||
sub_system_parts.append(f"\n## 你的工具能力\n{sub_tools_desc}")
|
||
|
||
sub_system = "\n".join(p for p in sub_system_parts if p)
|
||
sub_tools = _get_tools(config.provider, sub_allowed)
|
||
sub_messages: list[dict] = [{"role": "user", "content": task}]
|
||
|
||
full_response = ""
|
||
sid = subagent_id # 简写
|
||
|
||
for _round in range(MAX_SUBAGENT_ROUNDS):
|
||
round_text = ""
|
||
round_tool_calls: list[ToolCall] = []
|
||
|
||
async for event in stream_chat(config, sub_messages, sub_system, sub_tools):
|
||
if event.text:
|
||
round_text += event.text
|
||
await queue.put({"subagent_chunk": event.text, "subagent_id": sid})
|
||
if event.tool_calls:
|
||
round_tool_calls = event.tool_calls
|
||
|
||
if not round_tool_calls:
|
||
full_response += round_text
|
||
break
|
||
|
||
full_response += round_text
|
||
|
||
assistant_msg = build_assistant_message(
|
||
config.provider, round_text, round_tool_calls
|
||
)
|
||
sub_messages.append(assistant_msg)
|
||
|
||
tool_results = []
|
||
for tc in round_tool_calls:
|
||
label = TOOL_LABELS.get(tc.name, tc.name)
|
||
await queue.put({"subagent_tool_call": {
|
||
"name": tc.name, "label": label, "arguments": tc.arguments,
|
||
}, "subagent_id": sid})
|
||
result = execute_tool(tc.name, tc.arguments, novel_id)
|
||
tool_results.append({"id": tc.id, "result": result})
|
||
await queue.put({"subagent_tool_result": {
|
||
"name": tc.name, "label": label, "result": result,
|
||
}, "subagent_id": sid})
|
||
|
||
result_msgs = build_tool_results_messages(config.provider, tool_results)
|
||
sub_messages.extend(result_msgs)
|
||
|
||
await queue.put({"subagent_done": full_response, "subagent_id": sid})
|
||
return full_response
|
||
|
||
|
||
def _summarize_result(tool_name: str, result_json: str) -> str:
|
||
"""将工具结果精简为简短摘要，减少上下文 token 占用。"""
|
||
try:
|
||
data = json.loads(result_json)
|
||
except (json.JSONDecodeError, TypeError):
|
||
return result_json[:100]
|
||
|
||
if "error" in data:
|
||
return f"错误: {data['error']}"
|
||
|
||
# 列表类工具 → 显示数量 + 各条目 ID 和标题
|
||
if tool_name == "list_outlines":
|
||
items = data.get("outlines", [])
|
||
brief = ", ".join(f"#{o['id']}{o.get('summary', '')[:15]}" for o in items[:10])
|
||
return f"{data.get('total', 0)}条大纲: {brief}" if brief else f"{data.get('total', 0)}条大纲"
|
||
if tool_name == "list_characters":
|
||
items = data.get("characters", [])
|
||
brief = ", ".join(f"#{c['id']}{c.get('name', '')}" for c in items[:15])
|
||
return f"{data.get('total', 0)}个角色: {brief}" if brief else f"{data.get('total', 0)}个角色"
|
||
if tool_name == "list_chapters":
|
||
items = data.get("chapters", [])
|
||
brief = ", ".join(f"#{c['id']}{c.get('title', '')[:15]}" for c in items[:10])
|
||
return f"{data.get('total', 0)}个章节: {brief}" if brief else f"{data.get('total', 0)}个章节"
|
||
if tool_name == "list_files":
|
||
items = data.get("files", [])
|
||
brief = ", ".join(f"#{f['id']}{f.get('filename', '')}" for f in items[:5])
|
||
return f"{data.get('total', 0)}个文件: {brief}" if brief else f"{data.get('total', 0)}个文件"
|
||
|
||
# 创建类 → ID + 标题
|
||
if tool_name == "create_outline":
|
||
return f"已创建 #{data.get('id')} {data.get('summary', '')[:30]}"
|
||
if tool_name == "create_character":
|
||
return f"已创建 #{data.get('id')} {data.get('name', '')}"
|
||
if tool_name == "create_chapter":
|
||
return f"已创建 #{data.get('id')} {data.get('title', '')[:30]}"
|
||
|
||
# 更新类
|
||
if tool_name == "update_outline":
|
||
return f"已更新 #{data.get('id')}"
|
||
if tool_name == "update_character":
|
||
return f"已更新 #{data.get('id')} {data.get('name', '')}"
|
||
if tool_name == "update_chapter":
|
||
return f"已更新 #{data.get('id')}"
|
||
if tool_name == "update_novel_info":
|
||
return f"已更新小说信息"
|
||
|
||
# 删除类
|
||
if tool_name in ("delete_outline", "delete_character", "delete_chapter"):
|
||
return f"已删除 #{data.get('id')}"
|
||
|
||
# 排序
|
||
if tool_name == "reorder_outlines":
|
||
return f"已排序 {data.get('count', 0)}个节点"
|
||
|
||
# 世界观
|
||
if tool_name == "save_world_setting":
|
||
return f"已保存 ({data.get('content_length', 0)}字)"
|
||
if tool_name == "get_world_setting":
|
||
return f"{'有内容' if data.get('exists') else '暂无内容'}"
|
||
|
||
# 详情查看类
|
||
if tool_name == "get_outline_detail":
|
||
return f"大纲 #{data.get('id')} {data.get('summary', '')[:30]}"
|
||
if tool_name == "get_character_detail":
|
||
return f"角色 #{data.get('id')} {data.get('name', '')}"
|
||
if tool_name == "get_chapter_detail":
|
||
return f"章节 #{data.get('id')} {data.get('title', '')[:30]}"
|
||
if tool_name == "get_novel_info":
|
||
return f"小说: {data.get('title', '')}"
|
||
|
||
# 文件读取
|
||
if tool_name == "read_file":
|
||
return f"读取 {data.get('filename', '')} ({data.get('offset', 0)}-{data.get('offset', 0) + len(data.get('text', ''))}字/{data.get('total_length', 0)}字)"
|
||
|
||
# 其他 → 截取前 100 字符
|
||
return result_json[:100]
|
||
|
||
|
||
async def _execute_tool_with_queue(
|
||
queue: asyncio.Queue,
|
||
tc: ToolCall,
|
||
config: AIConfig,
|
||
novel_id: int,
|
||
) -> tuple[str, str]:
|
||
"""执行单个工具（含子Agent），通过 queue 发事件。返回 (result, log_line)。
|
||
子 Agent 使用 tc.id 作为 subagent_id，用于前端区分并行的子 Agent。"""
|
||
label = TOOL_LABELS.get(tc.name, tc.name)
|
||
|
||
if tc.name == "dispatch_subagent":
|
||
skill_id = tc.arguments.get("skill_id")
|
||
task_desc = tc.arguments.get("task", "")
|
||
sub_skill = None
|
||
with Session(engine) as sub_s:
|
||
sub_skill = sub_s.get(Skill, skill_id)
|
||
if not sub_skill:
|
||
result = json.dumps({"error": f"Skill {skill_id} 不存在"}, ensure_ascii=False)
|
||
else:
|
||
await queue.put({"subagent_start": {
|
||
"skill_id": skill_id, "skill_name": sub_skill.name,
|
||
"task": task_desc, "subagent_id": tc.id,
|
||
}})
|
||
result = await _run_subagent(
|
||
queue, config, sub_skill, task_desc, novel_id, subagent_id=tc.id,
|
||
)
|
||
skill_name = sub_skill.name if sub_skill else "未知"
|
||
# 子 Agent 结果也精简（取前 200 字）
|
||
brief = result[:200] + "…" if len(result) > 200 else result
|
||
log_line = f"[子Agent: {skill_name}] 任务: {task_desc} → {brief}"
|
||
else:
|
||
result = execute_tool(tc.name, tc.arguments, novel_id)
|
||
await queue.put({"tool_result": {"name": tc.name, "label": label, "result": result}})
|
||
# 精简日志：仅记录摘要，避免完整 JSON 结果撑爆上下文
|
||
log_line = f"[工具调用: {label}] 参数: {json.dumps(tc.arguments, ensure_ascii=False)} → {_summarize_result(tc.name, result)}"
|
||
|
||
return result, log_line
|
||
|
||
|
||
async def _execute_tools_parallel(
|
||
queue: asyncio.Queue,
|
||
tool_calls: list[ToolCall],
|
||
config: AIConfig,
|
||
novel_id: int,
|
||
) -> tuple[list[dict], list[str]]:
|
||
"""执行一组工具调用。子 Agent 并行执行，普通工具顺序执行。
|
||
返回 (tool_results, tool_call_logs)。"""
|
||
# 分离子 Agent 和普通工具
|
||
subagent_calls = [tc for tc in tool_calls if tc.name == "dispatch_subagent"]
|
||
normal_calls = [tc for tc in tool_calls if tc.name != "dispatch_subagent"]
|
||
|
||
results_map: dict[str, tuple[str, str]] = {} # tc.id → (result, log_line)
|
||
|
||
# 普通工具顺序执行
|
||
for tc in normal_calls:
|
||
result, log_line = await _execute_tool_with_queue(queue, tc, config, novel_id)
|
||
results_map[tc.id] = (result, log_line)
|
||
|
||
# 子 Agent 并行执行
|
||
if subagent_calls:
|
||
async def _run_one(tc: ToolCall) -> tuple[str, tuple[str, str]]:
|
||
r, l = await _execute_tool_with_queue(queue, tc, config, novel_id)
|
||
return tc.id, (r, l)
|
||
|
||
tasks = [_run_one(tc) for tc in subagent_calls]
|
||
for coro in asyncio.as_completed(tasks):
|
||
tc_id, rl = await coro
|
||
results_map[tc_id] = rl
|
||
|
||
# 按原始顺序组装结果
|
||
tool_results = []
|
||
tool_call_logs = []
|
||
for tc in tool_calls:
|
||
result, log_line = results_map[tc.id]
|
||
tool_results.append({"id": tc.id, "result": result})
|
||
tool_call_logs.append(log_line)
|
||
|
||
return tool_results, tool_call_logs
|
||
|
||
|
||
async def _chat_worker(
|
||
queue: asyncio.Queue,
|
||
config: AIConfig,
|
||
messages: list[dict],
|
||
system_prompt: str,
|
||
tools: list[dict] | None,
|
||
data: ChatRequest,
|
||
):
|
||
"""核心聊天逻辑 —— 运行在独立 Task 中，客户端断开也不会停止。"""
|
||
full_response = data.assistant_text or ""
|
||
usage_info: dict = {}
|
||
tool_call_logs: list[str] = []
|
||
pending_paused = False
|
||
|
||
try:
|
||
# ── 阶段一：执行待审批的工具调用 ──
|
||
if data.pending_tool_calls:
|
||
tool_calls = [
|
||
ToolCall(id=tc.id, name=tc.name, arguments=tc.arguments)
|
||
for tc in data.pending_tool_calls
|
||
]
|
||
assistant_msg = build_assistant_message(
|
||
config.provider, data.assistant_text or "", tool_calls
|
||
)
|
||
messages.append(assistant_msg)
|
||
|
||
tool_results, logs = await _execute_tools_parallel(
|
||
queue, tool_calls, config, data.novel_id,
|
||
)
|
||
tool_call_logs.extend(logs)
|
||
|
||
result_msgs = build_tool_results_messages(config.provider, tool_results)
|
||
messages.extend(result_msgs)
|
||
|
||
# ── 阶段二：LLM 循环 ──
|
||
for _round in range(MAX_TOOL_ROUNDS):
|
||
round_text = ""
|
||
round_tool_calls: list[ToolCall] = []
|
||
|
||
async for event in stream_chat(config, messages, system_prompt, tools):
|
||
if event.text:
|
||
round_text += event.text
|
||
await queue.put({"content": event.text})
|
||
if event.usage:
|
||
for k, v in event.usage.items():
|
||
if v:
|
||
usage_info[k] = usage_info.get(k, 0) + v
|
||
if event.tool_calls:
|
||
round_tool_calls = event.tool_calls
|
||
|
||
if not round_tool_calls:
|
||
full_response += round_text
|
||
break
|
||
|
||
full_response += round_text
|
||
calls_data = [
|
||
{"id": tc.id, "name": tc.name,
|
||
"label": TOOL_LABELS.get(tc.name, tc.name),
|
||
"arguments": tc.arguments}
|
||
for tc in round_tool_calls
|
||
]
|
||
|
||
if data.auto_approve_tools:
|
||
await queue.put({"tool_calls_auto": calls_data})
|
||
|
||
assistant_msg = build_assistant_message(
|
||
config.provider, full_response, round_tool_calls
|
||
)
|
||
messages.append(assistant_msg)
|
||
|
||
tool_results, logs = await _execute_tools_parallel(
|
||
queue, round_tool_calls, config, data.novel_id,
|
||
)
|
||
tool_call_logs.extend(logs)
|
||
|
||
result_msgs = build_tool_results_messages(config.provider, tool_results)
|
||
messages.extend(result_msgs)
|
||
continue
|
||
else:
|
||
# 审批模式：暂停
|
||
await queue.put({"tool_calls_pending": calls_data, "assistant_text": full_response})
|
||
await queue.put({"done": True, "pending": True, "usage": usage_info or None})
|
||
pending_paused = True
|
||
return # 不保存到 DB
|
||
|
||
except Exception as e:
|
||
await queue.put({"error": str(e)})
|
||
finally:
|
||
if not pending_paused:
|
||
_save_assistant_message(data.novel_id, full_response, tool_call_logs)
|
||
await queue.put({"done": True, "usage": usage_info or None})
|
||
# 哨兵值：通知 SSE 生成器结束
|
||
await queue.put(None)
|
||
|
||
|
||
@router.post("/stream")
|
||
async def chat_stream(
|
||
data: ChatRequest,
|
||
session: Session = Depends(get_session),
|
||
):
|
||
"""SSE 流式对话端点，支持 tool calling + 用户审批。
|
||
核心工作在独立 Task 中运行，客户端断开不会中断。"""
|
||
config = session.exec(select(AIConfig)).first()
|
||
if not config or not config.api_key:
|
||
return StreamingResponse(
|
||
_error_stream("请先配置AI服务"),
|
||
media_type="text/event-stream",
|
||
)
|
||
|
||
# 保存用户消息（仅首次请求，续传审批时不保存）
|
||
if not data.pending_tool_calls:
|
||
user_msg = data.messages[-1] if data.messages else None
|
||
if user_msg and user_msg.role == "user":
|
||
db_msg = ChatMessage(
|
||
novel_id=data.novel_id,
|
||
role="user",
|
||
content=user_msg.content,
|
||
)
|
||
session.add(db_msg)
|
||
session.commit()
|
||
|
||
# 构建消息列表
|
||
messages = [{"role": m.role, "content": m.content} for m in data.messages]
|
||
|
||
# 读取 Skill 配置
|
||
skill: Skill | None = None
|
||
allowed_tools: list[str] | None = None
|
||
if data.skill_id:
|
||
skill = session.get(Skill, data.skill_id)
|
||
if skill and skill.allowed_tools:
|
||
try:
|
||
parsed = json.loads(skill.allowed_tools)
|
||
if isinstance(parsed, list) and len(parsed) > 0:
|
||
allowed_tools = parsed
|
||
except (json.JSONDecodeError, TypeError):
|
||
pass
|
||
|
||
# 构建上下文
|
||
use_tools = bool(data.tools_enabled)
|
||
system_prompt = _build_system_prompt(
|
||
data.novel_id, session, data.page_context, use_tools,
|
||
skill=skill, allowed_tools=allowed_tools,
|
||
)
|
||
tools = _get_tools(config.provider, allowed_tools) if use_tools else None
|
||
|
||
# 创建事件队列和后台工作任务
|
||
queue: asyncio.Queue = asyncio.Queue()
|
||
asyncio.create_task(
|
||
_chat_worker(queue, config, messages, system_prompt, tools, data)
|
||
)
|
||
|
||
async def generate():
|
||
"""SSE 生成器：从队列读取事件。客户端断开时后台任务继续运行。"""
|
||
try:
|
||
while True:
|
||
event = await queue.get()
|
||
if event is None:
|
||
break
|
||
yield _sse(event)
|
||
except (asyncio.CancelledError, GeneratorExit):
|
||
# 客户端断开连接 — 后台任务继续运行，最终会保存到 DB
|
||
pass
|
||
|
||
return StreamingResponse(
|
||
generate(),
|
||
media_type="text/event-stream",
|
||
headers={
|
||
"Cache-Control": "no-cache",
|
||
"X-Accel-Buffering": "no",
|
||
},
|
||
)
|
||
|
||
|
||
async def _error_stream(message: str):
|
||
yield _sse({"error": message})
|
||
yield _sse({"done": True})
|
||
|
||
|
||
@router.get("/messages", response_model=ChatMessageList)
|
||
def list_messages(
|
||
novel_id: int | None = Query(default=None),
|
||
limit: int = Query(default=50, ge=1, le=200),
|
||
session: Session = Depends(get_session),
|
||
):
|
||
query = select(ChatMessage)
|
||
count_query = select(func.count(ChatMessage.id))
|
||
if novel_id is not None:
|
||
query = query.where(ChatMessage.novel_id == novel_id)
|
||
count_query = count_query.where(ChatMessage.novel_id == novel_id)
|
||
else:
|
||
query = query.where(ChatMessage.novel_id.is_(None)) # type: ignore
|
||
count_query = count_query.where(ChatMessage.novel_id.is_(None)) # type: ignore
|
||
|
||
total = session.exec(count_query).one()
|
||
items = session.exec(
|
||
query.order_by(ChatMessage.created_at.desc()).limit(limit)
|
||
).all()
|
||
items = list(reversed(items))
|
||
return ChatMessageList(
|
||
items=[ChatMessageRead.model_validate(m, from_attributes=True) for m in items],
|
||
total=total,
|
||
)
|
||
|
||
|
||
@router.delete("/messages", status_code=204)
|
||
def clear_messages(
|
||
novel_id: int | None = Query(default=None),
|
||
session: Session = Depends(get_session),
|
||
):
|
||
query = select(ChatMessage)
|
||
if novel_id is not None:
|
||
query = query.where(ChatMessage.novel_id == novel_id)
|
||
else:
|
||
query = query.where(ChatMessage.novel_id.is_(None)) # type: ignore
|
||
messages = session.exec(query).all()
|
||
for msg in messages:
|
||
session.delete(msg)
|
||
session.commit()
|
||
|
||
|
||
# 保留最近对话轮次数（每轮 = 1 user + 1 assistant）
|
||
KEEP_RECENT_ROUNDS = 2
|
||
|
||
|
||
@router.post("/compress")
|
||
async def compress_context(
|
||
novel_id: int | None = Query(default=None),
|
||
session: Session = Depends(get_session),
|
||
):
|
||
"""压缩对话上下文：用 LLM 总结旧消息，仅保留最近几轮对话。"""
|
||
config = session.exec(select(AIConfig)).first()
|
||
if not config or not config.api_key:
|
||
return {"error": "请先配置AI服务"}
|
||
|
||
# 查询所有消息（按时间正序）
|
||
query = select(ChatMessage)
|
||
if novel_id is not None:
|
||
query = query.where(ChatMessage.novel_id == novel_id)
|
||
else:
|
||
query = query.where(ChatMessage.novel_id.is_(None)) # type: ignore
|
||
all_msgs = list(session.exec(query.order_by(ChatMessage.created_at)).all())
|
||
|
||
if len(all_msgs) <= KEEP_RECENT_ROUNDS * 2:
|
||
return {"compressed": False, "reason": "消息太少，无需压缩"}
|
||
|
||
# 分割：旧消息（将被删除）+ 最近消息（保留原文）
|
||
keep_count = KEEP_RECENT_ROUNDS * 2
|
||
old_msgs = all_msgs[:-keep_count]
|
||
recent_msgs = all_msgs[-keep_count:]
|
||
|
||
# 将 **所有消息**（包括最近保留的）都给 LLM 看，确保摘要完整准确
|
||
conversation_text = ""
|
||
for msg in all_msgs:
|
||
role_label = "用户" if msg.role == "user" else "AI助手"
|
||
content = msg.content
|
||
if content.startswith("[对话摘要]"):
|
||
conversation_text += f"[之前的摘要]{content[5:]}\n\n"
|
||
else:
|
||
conversation_text += f"{role_label}: {content}\n\n"
|
||
|
||
# 调用 LLM 生成摘要
|
||
summary_prompt = (
|
||
"你是一个对话摘要助手。请将以下完整对话历史压缩为简洁的摘要，保留关键信息：\n"
|
||
"1. 用户提出的核心需求和决策\n"
|
||
"2. AI 执行的重要操作及结果（如创建/修改了哪些大纲、角色、章节等）\n"
|
||
"3. 达成的共识和待办事项\n"
|
||
"4. 当前工作进展和下一步计划\n\n"
|
||
"只输出摘要内容，不要加前缀或解释。用简洁的条目列表形式。"
|
||
)
|
||
summary_messages = [
|
||
{"role": "user", "content": f"请压缩以下对话历史：\n\n{conversation_text}"}
|
||
]
|
||
|
||
try:
|
||
summary = await simple_completion(config, summary_messages, summary_prompt)
|
||
except Exception as e:
|
||
return {"error": f"摘要生成失败: {str(e)}"}
|
||
|
||
# 删除旧消息
|
||
for msg in old_msgs:
|
||
session.delete(msg)
|
||
|
||
# 插入摘要消息，时间设为保留消息之前（确保排在最前面）
|
||
from datetime import timedelta
|
||
earliest_kept = recent_msgs[0].created_at
|
||
summary_msg = ChatMessage(
|
||
novel_id=novel_id,
|
||
role="assistant",
|
||
content=f"[对话摘要]\n{summary}",
|
||
created_at=earliest_kept - timedelta(seconds=1),
|
||
)
|
||
session.add(summary_msg)
|
||
session.commit()
|
||
|
||
return {
|
||
"compressed": True,
|
||
"removed_count": len(old_msgs),
|
||
"kept_count": len(recent_msgs),
|
||
"summary_length": len(summary),
|
||
}
|