新增子 Agent 与技能系统管理:
All checks were successful
Deploy Noval to K3s / deploy (push) Successful in 54s

- **子 Agent 支持**:
  - 新增 `dispatch_subagent` 工具,支持分配特定任务给子 Agent,由其独立操作。
  - 前端新增 `ChatSubAgent` 组件,支持子 Agent 任务状态展示(运行中/已完成)、工具调用进度及输出预览等。

- **技能管理扩展**:
  - 新增 Skill 编辑器页面(`SkillEditorView.vue`),支持创建、编辑、删除技能。
  - 完善技能工具权限分组选择功能,按类型灵活分配工具。

- **工具增强**:
  - 新增工具接口:`get_outline_detail`、`get_character_detail`、`get_chapter_detail`,用于获取详情内容。
  - 优化部分工具返回数据,仅输出基础信息以减少上下文体积。

- 后端添加 `SPECIAL_TOOLS` 支持,增强 `dispatch_subagent` 的特殊处理能力。

同时优化部分前后端交互,提升代码维护性与可读性。
This commit is contained in:
2026-03-23 02:23:22 +08:00
parent b39377e487
commit 0525c8217e
12 changed files with 2028 additions and 209 deletions

View File

@@ -1,3 +1,4 @@
import asyncio
import json
from datetime import datetime, timezone
@@ -12,7 +13,7 @@ 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
from ..services.tools import get_openai_tools, get_anthropic_tools, execute_tool, get_tools_description, SPECIAL_TOOLS
router = APIRouter(prefix="/api/chat", tags=["chat"])
@@ -24,6 +25,7 @@ TOOL_LABELS: dict[str, str] = {
"get_novel_info": "查看小说信息",
"update_novel_info": "更新小说信息",
"list_outlines": "查询大纲",
"get_outline_detail": "查看大纲详情",
"create_outline": "创建大纲",
"update_outline": "更新大纲",
"delete_outline": "删除大纲",
@@ -31,17 +33,23 @@ TOOL_LABELS: dict[str, str] = {
"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,
@@ -78,6 +86,21 @@ def _build_system_prompt(
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## 可用子AgentSkill"]
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)
@@ -108,12 +131,342 @@ def _save_assistant_message(novel_id: int | None, full_response: str, tool_call_
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']}"
# 列表类工具 → 仅显示数量
if tool_name == "list_outlines":
return f"{data.get('total', 0)}条大纲"
if tool_name == "list_characters":
return f"{data.get('total', 0)}个角色"
if tool_name == "list_chapters":
return f"{data.get('total', 0)}个章节"
if tool_name == "list_files":
return 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 + 用户审批"""
"""SSE 流式对话端点,支持 tool calling + 用户审批
核心工作在独立 Task 中运行,客户端断开不会中断。"""
config = session.exec(select(AIConfig)).first()
if not config or not config.api_key:
return StreamingResponse(
@@ -157,120 +510,23 @@ async def chat_stream(
)
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():
nonlocal messages
full_response = data.assistant_text or ""
usage_info: dict = {}
# 记录本次对话中的工具调用摘要(用于保存到 DB
tool_call_logs: list[str] = []
"""SSE 生成器:从队列读取事件。客户端断开时后台任务继续运行。"""
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 消息(含工具调用)
assistant_msg = build_assistant_message(
config.provider, data.assistant_text or "", tool_calls
)
messages.append(assistant_msg)
# 执行每个工具,发送结果
tool_results = []
for tc in tool_calls:
label = TOOL_LABELS.get(tc.name, tc.name)
result = execute_tool(tc.name, tc.arguments, data.novel_id)
tool_results.append({"id": tc.id, "result": result})
tool_call_logs.append(f"[工具调用: {label}] 参数: {json.dumps(tc.arguments, ensure_ascii=False)} → 结果: {result}")
yield _sse({"tool_result": {
"name": tc.name,
"label": label,
"result": result,
}})
# 追加工具结果消息
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 = []
async for event in stream_chat(config, messages, system_prompt, tools):
if event.text:
round_text += event.text
yield _sse({"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
while True:
event = await queue.get()
if event is None:
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:
# 自动执行模式:直接执行工具,不暂停
yield _sse({"tool_calls_auto": calls_data})
assistant_msg = build_assistant_message(
config.provider, full_response, round_tool_calls
)
messages.append(assistant_msg)
tool_results = []
for tc in round_tool_calls:
label = TOOL_LABELS.get(tc.name, tc.name)
result = execute_tool(tc.name, tc.arguments, data.novel_id)
tool_results.append({"id": tc.id, "result": result})
tool_call_logs.append(f"[工具调用: {label}] 参数: {json.dumps(tc.arguments, ensure_ascii=False)} → 结果: {result}")
yield _sse({"tool_result": {
"name": tc.name,
"label": label,
"result": result,
}})
result_msgs = build_tool_results_messages(config.provider, tool_results)
messages.extend(result_msgs)
# 继续循环,让 LLM 基于工具结果生成回复
continue
else:
# 审批模式:暂停,发给前端审批
yield _sse({
"tool_calls_pending": calls_data,
"assistant_text": full_response,
})
yield _sse({"done": True, "pending": True, "usage": usage_info or None})
return # 暂停,不保存到 DB
except Exception as e:
yield _sse({"error": str(e)})
finally:
# 无论连接是否断开,都保存已有的回复和工具调用到 DB
# 这确保刷新页面后 loadHistory 能恢复正确状态
_save_assistant_message(data.novel_id, full_response, tool_call_logs)
yield _sse({"done": True, "usage": usage_info or None})
yield _sse(event)
except (asyncio.CancelledError, GeneratorExit):
# 客户端断开连接 — 后台任务继续运行,最终会保存到 DB
pass
return StreamingResponse(
generate(),
@@ -354,18 +610,16 @@ async def compress_context(
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 old_msgs:
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:
@@ -373,10 +627,11 @@ async def compress_context(
# 调用 LLM 生成摘要
summary_prompt = (
"你是一个对话摘要助手。请将以下对话历史压缩为简洁的摘要,保留关键信息:\n"
"你是一个对话摘要助手。请将以下完整对话历史压缩为简洁的摘要,保留关键信息:\n"
"1. 用户提出的核心需求和决策\n"
"2. AI 执行的重要操作及结果(如创建/修改了哪些大纲、角色、章节等)\n"
"3. 达成的共识和待办事项\n\n"
"3. 达成的共识和待办事项\n"
"4. 当前工作进展和下一步计划\n\n"
"只输出摘要内容,不要加前缀或解释。用简洁的条目列表形式。"
)
summary_messages = [
@@ -392,12 +647,14 @@ async def compress_context(
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=recent_msgs[0].created_at,
created_at=earliest_kept - timedelta(seconds=1),
)
session.add(summary_msg)
session.commit()