JOJO
811974d6e7
放弃完全隔离策略，改为在现有 MainTerminal/SubAgentManager/SubAgentTask 主链路 按对话级开关 metadata.multi_agent_mode=true 增加多智能体分支。 新增模块： - modules/multi_agent/__init__.py: 模块入口 - modules/multi_agent/role_store.py: 角色 Markdown Frontmatter 解析与归档 - modules/multi_agent/state.py: 多智能体会话状态机与消息格式化 - modules/multi_agent/prompts.py: 主智能体(Team Leader) + 子智能体提示词 - modules/multi_agent/tools.py: 9 个主智能体工具 + 4 个子智能体工具定义 - server/multi_agent.py: /multiagent/new 页面 + /api/multiagent/* 蓝图 现有代码改动： - modules/sub_agent/task.py: 扩展 multi_agent_mode/multi_agent_state/display_name 字段， 增加 ask_master/ask_other_agent/answer_other_agent/list_active_sub_agents 工具处理逻辑， 子智能体自然结束 assistant 输出即本轮结束（不调用 finish_task），上下文保留。 - modules/sub_agent/manager.py: create_sub_agent 增加 multi_agent_mode/role_id/display_name 参数， 增加 get_or_create_multi_agent_state/get_multi_agent_state/inject_message_to_sub_agent/_on_multi_agent_task_done 方法。 - core/main_terminal_parts/tools_definition/agent_tools.py: 多智能体模式下用 modules.multi_agent.tools 替换旧版工具集。 - core/main_terminal_parts/context/messages.py: 多智能体模式下追加 Team Leader 系统提示词。 - core/main_terminal_parts/tools_execution.py: create_sub_agent handler 增加多智能体分支，新增 send_message_to_sub_agent/ask_sub_agent/answer_sub_agent_question/create_custom_agent/list_agents/list_active_sub_agents handler。 - core/web_terminal.py: load_conversation 时检测 metadata.multi_agent_mode 设置 self.multi_agent_mode。 - server/app_legacy.py: 注册 multi_agent_bp 蓝图。 前端改动： - static/src/auth/LoginApp.vue: 登录页增加'多智能体模式(beta)'按钮 - static/src/app/methods/ui/route.ts: 识别 /multiagent/new 和 /multiagent/conv_xxx 路径，进入多智能体模式并创建带 metadata.multi_agent_mode=true 的对话 - static/src/app/state.ts: 增加 multiAgentMode 状态字段 数据： - ~/.astrion/astrion/host/mutiagents/agents/: 4 个预置角色 ui-operator / full-stack-engineer / code-reviewer / researcher - ~/.astrion/astrion/host/mutiagents/conversations/: 会话数据 验证：所有 Python 文件语法检查通过；冒烟测试 test.test_server_refactor_smoke 6 项全通过；前端构建通过（6.04s）；模块导入与功能断言测试全部通过。
208 lines
6.3 KiB
Python
208 lines
6.3 KiB
Python
"""多智能体角色存储与解析。
|
||
|
||
角色定义文件格式（Markdown + Frontmatter）:
|
||
|
||
---
|
||
id: ui-operator
|
||
name: UI Operator
|
||
description: 界面设计与前端实现
|
||
model: ""
|
||
thinking_mode: fast
|
||
skills:
|
||
- frontend-design
|
||
---
|
||
|
||
自定义 prompt body...
|
||
|
||
角色目录：
|
||
- ~/.astrion/astrion/host/mutiagents/agents/<role_id>.md
|
||
|
||
存储约定见 .astrion/memory/multi_agent_mode_design.md。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
from pathlib import Path
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
try:
|
||
from config.paths import IS_HOST_MODE, RUNTIME_ROOT
|
||
except ImportError:
|
||
# 防止直接 import 时 config 尚未加载，提供回退值
|
||
IS_HOST_MODE = True
|
||
RUNTIME_ROOT = str(Path.home() / ".astrion" / "astrion")
|
||
|
||
# 复用项目的 MUTIAGENTS 数据目录（保留原拼写 "mutiagents"，见项目记忆）
|
||
DEFAULT_MUTIAGENTS_DIR = Path(RUNTIME_ROOT) / ("host" if IS_HOST_MODE else "web") / "mutiagents"
|
||
AGENTS_DIR_NAME = "agents"
|
||
|
||
FRONTMATTER_RE = re.compile(r"^---\s*\n(?P<body>.*?)\n---\s*\n(?P<rest>.*)$", re.S)
|
||
|
||
|
||
def _agents_dir() -> Path:
|
||
"""返回角色根目录（按运行模式自动选择 host/web）。"""
|
||
base = Path(DEFAULT_MUTIAGENTS_DIR)
|
||
out = base / AGENTS_DIR_NAME
|
||
out.mkdir(parents=True, exist_ok=True)
|
||
return out
|
||
|
||
|
||
def _parse_frontmatter(text: str) -> tuple[Dict[str, Any], str]:
|
||
"""解析 Markdown frontmatter，返回 (元数据 dict, body)。"""
|
||
m = FRONTMATTER_RE.match(text)
|
||
if not m:
|
||
return {}, text
|
||
raw_meta = m.group("body") or ""
|
||
body = m.group("rest") or ""
|
||
meta: Dict[str, Any] = {}
|
||
# 简易解析（k: v 与 k: 列表项，不引入 yaml 依赖）
|
||
current_key: Optional[str] = None
|
||
for line in raw_meta.splitlines():
|
||
if not line.strip():
|
||
continue
|
||
if line.startswith(" - ") or line.startswith("- "):
|
||
value = line.lstrip(" ").lstrip("- ").strip()
|
||
if current_key and isinstance(meta.get(current_key), list):
|
||
meta[current_key].append(value)
|
||
continue
|
||
if ":" in line:
|
||
k, v = line.split(":", 1)
|
||
k = k.strip()
|
||
v = v.strip().strip('"').strip("'")
|
||
if v:
|
||
meta[k] = v
|
||
current_key = None
|
||
else:
|
||
# 可能是列表块
|
||
meta[k] = []
|
||
current_key = k
|
||
return meta, body.strip()
|
||
|
||
|
||
def load_role_from_file(file_path: Path) -> Optional["RoleConfig"]:
|
||
"""从 Markdown 文件加载角色定义。"""
|
||
if not file_path.exists() or not file_path.is_file():
|
||
return None
|
||
text = file_path.read_text(encoding="utf-8")
|
||
meta, body = _parse_frontmatter(text)
|
||
if not meta.get("id") or not meta.get("name"):
|
||
return None
|
||
return RoleConfig(
|
||
role_id=str(meta["id"]),
|
||
name=str(meta["name"]),
|
||
description=str(meta.get("description") or ""),
|
||
body_prompt=body,
|
||
model_key=(str(meta["model"]) if meta.get("model") else None) or None,
|
||
thinking_mode=str(meta.get("thinking_mode") or "fast"),
|
||
skills=list(meta.get("skills") or []),
|
||
source_file=str(file_path),
|
||
)
|
||
|
||
|
||
def load_preset_role(role_id: str) -> Optional["RoleConfig"]:
|
||
"""从预设/自定义角色目录加载一个角色。"""
|
||
f = _agents_dir() / f"{role_id}.md"
|
||
return load_role_from_file(f)
|
||
|
||
|
||
def list_roles() -> List["RoleConfig"]:
|
||
"""列出全部角色（预设 + 用户自定义）。"""
|
||
agents_dir = _agents_dir()
|
||
if not agents_dir.exists():
|
||
return []
|
||
roles: List[RoleConfig] = []
|
||
for p in sorted(agents_dir.glob("*.md")):
|
||
r = load_role_from_file(p)
|
||
if r and r.role_id:
|
||
roles.append(r)
|
||
return roles
|
||
|
||
|
||
def save_custom_role(role: "RoleConfig") -> Path:
|
||
"""把自定义角色保存到 agents 目录。"""
|
||
f = _agents_dir() / f"{role.role_id}.md"
|
||
f.parent.mkdir(parents=True, exist_ok=True)
|
||
f.write_text(_serialize_role(role), encoding="utf-8")
|
||
return f
|
||
|
||
|
||
def _serialize_role(role: "RoleConfig") -> str:
|
||
"""把 RoleConfig 序列化为 Markdown frontmatter 字符串。"""
|
||
lines = ["---"]
|
||
lines.append(f"id: {role.role_id}")
|
||
lines.append(f"name: {role.name}")
|
||
if role.description:
|
||
# 单行描述转义
|
||
desc = role.description.replace('"', '\\"')
|
||
lines.append(f'description: "{desc}"')
|
||
if role.model_key:
|
||
lines.append(f"model: {role.model_key}")
|
||
lines.append(f"thinking_mode: {role.thinking_mode or 'fast'}")
|
||
if role.skills:
|
||
lines.append("skills:")
|
||
for s in role.skills:
|
||
lines.append(f" - {s}")
|
||
lines.append("---")
|
||
lines.append("")
|
||
lines.append(role.body_prompt or "")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def build_role_system_prompt(role: "RoleConfig") -> str:
|
||
"""生成子智能体的「专属 prompt」部分（frontmatter 之后的 body）。"""
|
||
return role.body_prompt or ""
|
||
|
||
|
||
class RoleConfig:
|
||
"""一个多智能体角色配置。"""
|
||
|
||
__slots__ = (
|
||
"role_id",
|
||
"name",
|
||
"description",
|
||
"body_prompt",
|
||
"model_key",
|
||
"thinking_mode",
|
||
"skills",
|
||
"source_file",
|
||
)
|
||
|
||
def __init__(
|
||
self,
|
||
*,
|
||
role_id: str,
|
||
name: str,
|
||
description: str = "",
|
||
body_prompt: str = "",
|
||
model_key: Optional[str] = None,
|
||
thinking_mode: str = "fast",
|
||
skills: Optional[List[str]] = None,
|
||
source_file: str = "",
|
||
):
|
||
self.role_id = role_id
|
||
self.name = name
|
||
self.description = description
|
||
self.body_prompt = body_prompt
|
||
self.model_key = model_key
|
||
self.thinking_mode = thinking_mode or "fast"
|
||
self.skills = list(skills or [])
|
||
self.source_file = source_file
|
||
|
||
def display_name(self, agent_id: int) -> str:
|
||
"""根据实例 id 生成显示名。"""
|
||
return f"{self.name}_{agent_id}"
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
return {
|
||
"role_id": self.role_id,
|
||
"name": self.name,
|
||
"description": self.description,
|
||
"model_key": self.model_key,
|
||
"thinking_mode": self.thinking_mode,
|
||
"skills": self.skills,
|
||
"source_file": self.source_file,
|
||
}
|
||
|
||
def __repr__(self) -> str:
|
||
return f"<RoleConfig id={self.role_id} name={self.name}>" |