feat: 技能六级可插拔隔离(global/org/pipeline/role/project/user)+slash命令注册表
This commit is contained in:
parent
c3fcc35e08
commit
82c2724789
@ -36,6 +36,14 @@ from .ability import (
|
||||
get_ability_tools,
|
||||
get_ability_handler,
|
||||
)
|
||||
from .slash import (
|
||||
SlashCommand,
|
||||
register_slash_command,
|
||||
get_slash_commands,
|
||||
resolve_slash_command,
|
||||
parse_slash_args,
|
||||
list_commands,
|
||||
)
|
||||
from .tool_registry import (
|
||||
ToolRegistry,
|
||||
get_tool_registry,
|
||||
|
||||
@ -49,12 +49,15 @@ class MemoryConfig:
|
||||
|
||||
@dataclass
|
||||
class SkillConfig:
|
||||
"""技能配置 — 三级隔离：全局 + 组织 + 用户"""
|
||||
"""技能配置 — 六级隔离：global→org→pipeline→role→project→user"""
|
||||
enabled: bool = True
|
||||
base_dir: str = "skills" # 技能根目录，其下为 global/orgs/users/
|
||||
base_dir: str = "skills" # 技能根目录，其下为 global/pipelines/projects/orgs/users/
|
||||
max_skills_per_turn: int = 5
|
||||
enable_global: bool = True # 启用全局技能
|
||||
enable_org: bool = True # 启用组织技能
|
||||
enable_pipeline: bool = True # 启用产线技能
|
||||
enable_role: bool = True # 启用角色技能
|
||||
enable_project: bool = True # 启用项目/会话技能
|
||||
enable_user: bool = True # 启用用户技能
|
||||
|
||||
|
||||
@ -117,6 +120,9 @@ class AgentConfig:
|
||||
"base_dir": self.skills.base_dir,
|
||||
"enable_global": self.skills.enable_global,
|
||||
"enable_org": self.skills.enable_org,
|
||||
"enable_pipeline": self.skills.enable_pipeline,
|
||||
"enable_role": self.skills.enable_role,
|
||||
"enable_project": self.skills.enable_project,
|
||||
"enable_user": self.skills.enable_user,
|
||||
},
|
||||
"tools": [
|
||||
@ -165,6 +171,9 @@ class AgentConfig:
|
||||
max_skills_per_turn=sk.get("max_skills_per_turn", 5),
|
||||
enable_global=sk.get("enable_global", True),
|
||||
enable_org=sk.get("enable_org", True),
|
||||
enable_pipeline=sk.get("enable_pipeline", True),
|
||||
enable_role=sk.get("enable_role", True),
|
||||
enable_project=sk.get("enable_project", True),
|
||||
enable_user=sk.get("enable_user", True),
|
||||
),
|
||||
tools=[ToolDefinition(**t) if isinstance(t, dict) else t for t in data.get("tools", [])],
|
||||
|
||||
@ -1,33 +1,55 @@
|
||||
"""
|
||||
pipeline-core: Skill Loader v2 — 三级技能隔离
|
||||
pipeline-core: Skill Loader — 六级可插拔技能隔离
|
||||
|
||||
技能目录结构（每级可独立存在，同名高优先级覆盖低优先级）：
|
||||
|
||||
技能目录结构：
|
||||
skills/
|
||||
global/ ← 全局技能（所有人可见）
|
||||
skill-name/
|
||||
SKILL.md
|
||||
orgs/ ← 组织技能
|
||||
{org_id}/
|
||||
skill-name/
|
||||
SKILL.md
|
||||
users/ ← 用户个人技能
|
||||
{user_id}/
|
||||
skill-name/
|
||||
SKILL.md
|
||||
global/ ← 通用技能（core 内核，所有产线所有角色）
|
||||
skill-name/SKILL.md
|
||||
pipelines/ ← 产线技能
|
||||
{pipeline_id}/
|
||||
common/ ← 产线通用技能（该产线所有角色可见）
|
||||
skill-name/SKILL.md
|
||||
roles/ ← 角色技能（产线内角色专属）
|
||||
{role}/
|
||||
skill-name/SKILL.md
|
||||
projects/ ← 项目/会话技能（临时、项目特定）
|
||||
{project_id}/
|
||||
skill-name/SKILL.md
|
||||
orgs/{org_id}/ ← 组织技能
|
||||
users/{user_id}/ ← 用户个人技能
|
||||
|
||||
加载优先级：用户 > 组织 > 全局
|
||||
同名技能：用户级覆盖组织级覆盖全局级
|
||||
加载优先级（低 → 高，同名后者覆盖前者）：
|
||||
global → org → pipeline(common) → role → project → user
|
||||
"""
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Dict, List, Optional
|
||||
|
||||
logger = logging.getLogger("pipeline.skill_loader")
|
||||
|
||||
# scope 优先级（数值越大优先级越高）
|
||||
SCOPE_PRIORITY = {
|
||||
"global": 0,
|
||||
"org": 1,
|
||||
"pipeline": 2,
|
||||
"role": 3,
|
||||
"project": 4,
|
||||
"user": 5,
|
||||
}
|
||||
|
||||
SCOPE_TAG = {
|
||||
"global": "通用",
|
||||
"org": "组织",
|
||||
"pipeline": "产线",
|
||||
"role": "角色",
|
||||
"project": "项目",
|
||||
"user": "个人",
|
||||
}
|
||||
|
||||
|
||||
# ── 技能数据模型 ──
|
||||
|
||||
@dataclass
|
||||
@ -37,15 +59,14 @@ class Skill:
|
||||
path: str # SKILL.md 绝对路径
|
||||
description: str = ""
|
||||
content: str = ""
|
||||
scope: str = "global" # global / org / user
|
||||
scope_id: str = "" # org_id 或 user_id
|
||||
scope: str = "global" # global/org/pipeline/role/project/user
|
||||
scope_id: str = "" # org_id / pipeline_id / pipeline_id:role / project_id / user_id
|
||||
trigger_keywords: List[str] = field(default_factory=list)
|
||||
version: str = "1.0.0"
|
||||
|
||||
@classmethod
|
||||
def from_file(cls, filepath: str, scope: str = "global",
|
||||
scope_id: str = "") -> "Skill":
|
||||
"""从 SKILL.md 文件加载技能"""
|
||||
name = os.path.basename(os.path.dirname(filepath))
|
||||
content = ""
|
||||
description = ""
|
||||
@ -59,13 +80,11 @@ class Skill:
|
||||
logger.warning(f"Skill read failed: {filepath} err={e}")
|
||||
return cls(name=name, path=filepath, scope=scope, scope_id=scope_id)
|
||||
|
||||
# 解析 YAML frontmatter
|
||||
fm = _parse_frontmatter(content)
|
||||
if fm:
|
||||
description = fm.get("description", description)
|
||||
trigger_keywords = fm.get("trigger_keywords", trigger_keywords)
|
||||
version = fm.get("version", version)
|
||||
# frontmatter 中的 scope 优先
|
||||
scope = fm.get("scope", scope)
|
||||
|
||||
if not description:
|
||||
@ -82,8 +101,7 @@ class Skill:
|
||||
)
|
||||
|
||||
def to_prompt_block(self) -> str:
|
||||
"""生成注入到 system prompt 的技能文本块"""
|
||||
scope_tag = {"global": "全局", "org": "组织", "user": "个人"}.get(self.scope, "")
|
||||
scope_tag = SCOPE_TAG.get(self.scope, self.scope)
|
||||
return f"""## [{scope_tag}] {self.name}
|
||||
{self.description}
|
||||
|
||||
@ -92,7 +110,6 @@ class Skill:
|
||||
|
||||
|
||||
def _parse_frontmatter(content: str) -> Optional[dict]:
|
||||
"""解析 YAML frontmatter"""
|
||||
if not content.startswith("---"):
|
||||
return None
|
||||
end = content.find("---", 3)
|
||||
@ -113,22 +130,23 @@ def _parse_frontmatter(content: str) -> Optional[dict]:
|
||||
return result
|
||||
|
||||
|
||||
# ── 三级技能加载器 ──
|
||||
# ── 六级技能加载器 ──
|
||||
|
||||
class SkillLoader:
|
||||
"""三级技能加载器：global → org → user"""
|
||||
"""六级技能加载器：global → org → pipeline → role → project → user"""
|
||||
|
||||
def __init__(self, base_dir: str = ""):
|
||||
self.base_dir = base_dir
|
||||
# 三级存储: {scope: {name: Skill}}
|
||||
self._global: Dict[str, Skill] = {}
|
||||
self._orgs: Dict[str, Dict[str, Skill]] = {} # {org_id: {name: Skill}}
|
||||
self._users: Dict[str, Dict[str, Skill]] = {} # {user_id: {name: Skill}}
|
||||
self._orgs: Dict[str, Dict[str, Skill]] = {} # {org_id: {name: Skill}}
|
||||
self._pipelines: Dict[str, Dict[str, Skill]] = {} # {pipeline_id: {name: Skill}}
|
||||
self._roles: Dict[str, Dict[str, Skill]] = {} # {pipeline_id:role: {name: Skill}}
|
||||
self._projects: Dict[str, Dict[str, Skill]] = {} # {project_id: {name: Skill}}
|
||||
self._users: Dict[str, Dict[str, Skill]] = {} # {user_id: {name: Skill}}
|
||||
if base_dir:
|
||||
self.reload()
|
||||
|
||||
def set_base_dir(self, base_dir: str):
|
||||
"""设置技能根目录并扫描"""
|
||||
self.base_dir = base_dir
|
||||
if base_dir:
|
||||
self.reload()
|
||||
@ -136,41 +154,74 @@ class SkillLoader:
|
||||
# ── 目录扫描 ──
|
||||
|
||||
def reload(self):
|
||||
"""重新扫描所有技能目录"""
|
||||
self._global = {}
|
||||
self._orgs = {}
|
||||
self._pipelines = {}
|
||||
self._roles = {}
|
||||
self._projects = {}
|
||||
self._users = {}
|
||||
|
||||
if not self.base_dir or not os.path.isdir(self.base_dir):
|
||||
return
|
||||
|
||||
# 全局技能: skills/global/
|
||||
# 通用: skills/global/
|
||||
global_dir = os.path.join(self.base_dir, "global")
|
||||
if os.path.isdir(global_dir):
|
||||
self._global = self._scan_scope_dir(global_dir, "global")
|
||||
|
||||
# 组织技能: skills/orgs/{org_id}/
|
||||
# 组织: skills/orgs/{org_id}/
|
||||
orgs_dir = os.path.join(self.base_dir, "orgs")
|
||||
if os.path.isdir(orgs_dir):
|
||||
for org_id in os.listdir(orgs_dir):
|
||||
org_path = os.path.join(orgs_dir, org_id)
|
||||
if os.path.isdir(org_path):
|
||||
self._orgs[org_id] = self._scan_scope_dir(org_path, "org", org_id)
|
||||
p = os.path.join(orgs_dir, org_id)
|
||||
if os.path.isdir(p):
|
||||
self._orgs[org_id] = self._scan_scope_dir(p, "org", org_id)
|
||||
|
||||
# 用户技能: skills/users/{user_id}/
|
||||
# 产线: skills/pipelines/{pipeline_id}/common/ + roles/{role}/
|
||||
pipelines_dir = os.path.join(self.base_dir, "pipelines")
|
||||
if os.path.isdir(pipelines_dir):
|
||||
for pid in os.listdir(pipelines_dir):
|
||||
pdir = os.path.join(pipelines_dir, pid)
|
||||
if not os.path.isdir(pdir):
|
||||
continue
|
||||
# 产线通用技能
|
||||
common_dir = os.path.join(pdir, "common")
|
||||
if os.path.isdir(common_dir):
|
||||
self._pipelines[pid] = self._scan_scope_dir(common_dir, "pipeline", pid)
|
||||
# 角色技能
|
||||
roles_dir = os.path.join(pdir, "roles")
|
||||
if os.path.isdir(roles_dir):
|
||||
for role in os.listdir(roles_dir):
|
||||
rdir = os.path.join(roles_dir, role)
|
||||
if os.path.isdir(rdir):
|
||||
key = f"{pid}:{role}"
|
||||
self._roles[key] = self._scan_scope_dir(rdir, "role", key)
|
||||
|
||||
# 项目/会话: skills/projects/{project_id}/
|
||||
projects_dir = os.path.join(self.base_dir, "projects")
|
||||
if os.path.isdir(projects_dir):
|
||||
for project_id in os.listdir(projects_dir):
|
||||
p = os.path.join(projects_dir, project_id)
|
||||
if os.path.isdir(p):
|
||||
self._projects[project_id] = self._scan_scope_dir(p, "project", project_id)
|
||||
|
||||
# 用户: skills/users/{user_id}/
|
||||
users_dir = os.path.join(self.base_dir, "users")
|
||||
if os.path.isdir(users_dir):
|
||||
for user_id in os.listdir(users_dir):
|
||||
user_path = os.path.join(users_dir, user_id)
|
||||
if os.path.isdir(user_path):
|
||||
self._users[user_id] = self._scan_scope_dir(user_path, "user", user_id)
|
||||
p = os.path.join(users_dir, user_id)
|
||||
if os.path.isdir(p):
|
||||
self._users[user_id] = self._scan_scope_dir(p, "user", user_id)
|
||||
|
||||
logger.debug(f"SkillLoader reloaded: global={len(self._global)}, "
|
||||
f"orgs={len(self._orgs)}, users={len(self._users)}")
|
||||
f"orgs={sum(len(s) for s in self._orgs.values())}, "
|
||||
f"pipelines={sum(len(s) for s in self._pipelines.values())}, "
|
||||
f"roles={sum(len(s) for s in self._roles.values())}, "
|
||||
f"projects={sum(len(s) for s in self._projects.values())}, "
|
||||
f"users={sum(len(s) for s in self._users.values())}")
|
||||
|
||||
def _scan_scope_dir(self, scope_dir: str, scope: str,
|
||||
scope_id: str = "") -> Dict[str, Skill]:
|
||||
"""扫描单个 scope 目录下的所有技能"""
|
||||
skills = {}
|
||||
for entry in os.listdir(scope_dir):
|
||||
entry_path = os.path.join(scope_dir, entry)
|
||||
@ -180,7 +231,6 @@ class SkillLoader:
|
||||
if os.path.isfile(skill_file):
|
||||
skill = Skill.from_file(skill_file, scope, scope_id)
|
||||
skills[skill.name] = skill
|
||||
# 也支持子目录嵌套
|
||||
for sub in os.listdir(entry_path):
|
||||
sub_path = os.path.join(entry_path, sub)
|
||||
if os.path.isdir(sub_path):
|
||||
@ -192,24 +242,40 @@ class SkillLoader:
|
||||
|
||||
# ── 查询 ──
|
||||
|
||||
def get_merged(self, org_id: str = "", user_id: str = "") -> Dict[str, Skill]:
|
||||
"""按优先级合并：全局 + 组织 + 用户。同名技能后者覆盖前者。"""
|
||||
merged = dict(self._global) # 全局为基础
|
||||
def get_merged(self, pipeline_id: str = "", role: str = "",
|
||||
project_id: str = "", org_id: str = "",
|
||||
user_id: str = "") -> Dict[str, Skill]:
|
||||
"""按优先级合并六级技能。同名技能高优先级覆盖低优先级。
|
||||
|
||||
优先级（低→高）：global → org → pipeline → role → project → user
|
||||
"""
|
||||
merged = dict(self._global)
|
||||
|
||||
# 组织技能覆盖
|
||||
if org_id and org_id in self._orgs:
|
||||
merged.update(self._orgs[org_id])
|
||||
|
||||
# 用户技能覆盖（最高优先级）
|
||||
if pipeline_id and pipeline_id in self._pipelines:
|
||||
merged.update(self._pipelines[pipeline_id])
|
||||
|
||||
if pipeline_id and role:
|
||||
role_key = f"{pipeline_id}:{role}"
|
||||
if role_key in self._roles:
|
||||
merged.update(self._roles[role_key])
|
||||
|
||||
if project_id and project_id in self._projects:
|
||||
merged.update(self._projects[project_id])
|
||||
|
||||
if user_id and user_id in self._users:
|
||||
merged.update(self._users[user_id])
|
||||
|
||||
return merged
|
||||
|
||||
def get_by_trigger(self, user_input: str, org_id: str = "",
|
||||
user_id: str = "", max_skills: int = 5) -> List[Skill]:
|
||||
"""根据用户输入匹配相关技能（基于 trigger_keywords）"""
|
||||
all_skills = list(self.get_merged(org_id, user_id).values())
|
||||
def get_by_trigger(self, user_input: str, pipeline_id: str = "",
|
||||
role: str = "", project_id: str = "",
|
||||
org_id: str = "", user_id: str = "",
|
||||
max_skills: int = 5) -> List[Skill]:
|
||||
"""根据用户输入匹配相关技能（trigger_keywords + scope 加权）"""
|
||||
all_skills = list(self.get_merged(pipeline_id, role, project_id, org_id, user_id).values())
|
||||
scored = []
|
||||
user_lower = user_input.lower()
|
||||
for skill in all_skills:
|
||||
@ -219,45 +285,41 @@ class SkillLoader:
|
||||
score += 1
|
||||
if skill.name.lower() in user_lower:
|
||||
score += 2
|
||||
# scope 加分：用户 > 组织 > 全局
|
||||
if skill.scope == "user":
|
||||
score += 3
|
||||
elif skill.scope == "org":
|
||||
score += 2
|
||||
elif skill.scope == "global":
|
||||
score += 1
|
||||
# scope 优先级加权
|
||||
score += SCOPE_PRIORITY.get(skill.scope, 0)
|
||||
if score > 0:
|
||||
scored.append((score, skill))
|
||||
scored.sort(key=lambda x: x[0], reverse=True)
|
||||
return [s[1] for s in scored[:max_skills]]
|
||||
|
||||
def get_skill_count(self, org_id: str = "", user_id: str = "") -> int:
|
||||
"""可见技能总数"""
|
||||
return len(self.get_merged(org_id, user_id))
|
||||
def get_skill_count(self, pipeline_id: str = "", role: str = "",
|
||||
project_id: str = "", org_id: str = "",
|
||||
user_id: str = "") -> int:
|
||||
return len(self.get_merged(pipeline_id, role, project_id, org_id, user_id))
|
||||
|
||||
def list_by_scope(self) -> Dict[str, int]:
|
||||
"""各 scope 的技能数量"""
|
||||
return {
|
||||
"global": len(self._global),
|
||||
"orgs": sum(len(s) for s in self._orgs.values()),
|
||||
"pipelines": sum(len(s) for s in self._pipelines.values()),
|
||||
"roles": sum(len(s) for s in self._roles.values()),
|
||||
"projects": sum(len(s) for s in self._projects.values()),
|
||||
"users": sum(len(s) for s in self._users.values()),
|
||||
}
|
||||
|
||||
# ── Prompt 构建 ──
|
||||
|
||||
def build_prompt_block(self, org_id: str = "", user_id: str = "",
|
||||
user_input: str = None, max_skills: int = 5) -> str:
|
||||
"""构建注入 system prompt 的技能段落。
|
||||
|
||||
有 user_input 时按 trigger 匹配，否则取全部（受限 max_skills）。
|
||||
"""
|
||||
def build_prompt_block(self, pipeline_id: str = "", role: str = "",
|
||||
project_id: str = "", org_id: str = "",
|
||||
user_id: str = "", user_input: Optional[str] = None,
|
||||
max_skills: int = 5) -> str:
|
||||
if user_input:
|
||||
skills = self.get_by_trigger(user_input, org_id, user_id, max_skills)
|
||||
skills = self.get_by_trigger(user_input, pipeline_id, role,
|
||||
project_id, org_id, user_id, max_skills)
|
||||
else:
|
||||
merged = self.get_merged(org_id, user_id)
|
||||
# 优先用户级
|
||||
merged = self.get_merged(pipeline_id, role, project_id, org_id, user_id)
|
||||
skills = sorted(merged.values(),
|
||||
key=lambda s: {"user": 0, "org": 1, "global": 2}.get(s.scope, 9))
|
||||
key=lambda s: SCOPE_PRIORITY.get(s.scope, 9))
|
||||
skills = skills[:max_skills]
|
||||
|
||||
if not skills:
|
||||
@ -265,14 +327,15 @@ class SkillLoader:
|
||||
|
||||
blocks = ["## 可用技能\n"]
|
||||
for s in skills:
|
||||
blocks.append(f"- **[{s.scope}] {s.name}**: {s.description[:200]}")
|
||||
blocks.append(f"- **[{SCOPE_TAG.get(s.scope, s.scope)}] {s.name}**: {s.description[:200]}")
|
||||
blocks.append("")
|
||||
return "\n".join(blocks)
|
||||
|
||||
def build_full_prompt_block(self, skill_names: List[str],
|
||||
org_id: str = "", user_id: str = "") -> str:
|
||||
"""注入指定技能的完整内容"""
|
||||
merged = self.get_merged(org_id, user_id)
|
||||
pipeline_id: str = "", role: str = "",
|
||||
project_id: str = "", org_id: str = "",
|
||||
user_id: str = "") -> str:
|
||||
merged = self.get_merged(pipeline_id, role, project_id, org_id, user_id)
|
||||
blocks = []
|
||||
for name in skill_names:
|
||||
if name in merged:
|
||||
@ -286,7 +349,6 @@ _loader: Optional[SkillLoader] = None
|
||||
|
||||
|
||||
def get_skill_loader(base_dir: str = "") -> SkillLoader:
|
||||
"""获取全局技能加载器单例。首次调用可传入 base_dir。"""
|
||||
global _loader
|
||||
if _loader is None:
|
||||
_loader = SkillLoader(base_dir)
|
||||
|
||||
83
pipeline_core/slash.py
Normal file
83
pipeline_core/slash.py
Normal file
@ -0,0 +1,83 @@
|
||||
"""
|
||||
pipeline-core: SlashCommand — 可插拔 slash 命令注册表
|
||||
|
||||
slash 命令 = 产线/角色/通用 的快捷指令，用户以 "/" 开头输入时直接触发，
|
||||
不进入 LLM 循环（类似 Hermes CLI 的 /new /model /tools /skills）。
|
||||
|
||||
命令可插拔：通用命令在 core 注册，产线命令在能力包（sdlc_ability）注册，
|
||||
不同产线可有不同的 slash 命名。
|
||||
|
||||
依赖方向：pipeline-core（本文件，纯定义 + 注册表）← pipeline-service（注册通用命令）
|
||||
← sdlc_ability（注册 SDLC 命令）
|
||||
"""
|
||||
|
||||
import logging
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Callable, Dict, List, Optional
|
||||
|
||||
logger = logging.getLogger("pipeline.slash")
|
||||
|
||||
|
||||
@dataclass
|
||||
class SlashCommand:
|
||||
"""单个 slash 命令。handler 签名：async def handler(args, ctx) -> str
|
||||
- args: 命令后的参数字符串（不含命令名）
|
||||
- ctx: AgentExecutor 上下文 dict（project_id/user_id/pipeline_id/workspace_dir/config/executor）
|
||||
"""
|
||||
name: str # 不含 "/" 前缀，如 "new"
|
||||
description: str = ""
|
||||
handler: Optional[Callable] = None
|
||||
scope: str = "global" # global / pipeline / role
|
||||
scope_id: str = "" # pipeline_id 或 pipeline_id:role
|
||||
args_hint: str = "" # 参数提示，如 "[task_id]"
|
||||
|
||||
|
||||
# ── 全局注册表 ──
|
||||
|
||||
_commands: Dict[str, SlashCommand] = {}
|
||||
|
||||
|
||||
def register_slash_command(cmd: SlashCommand):
|
||||
"""注册一个 slash 命令（幂等：重复注册覆盖）。"""
|
||||
_commands[cmd.name] = cmd
|
||||
logger.debug(f"SlashCommand registered: /{cmd.name} ({cmd.scope})")
|
||||
|
||||
|
||||
def get_slash_commands(pipeline_id: str = "", role: str = "") -> Dict[str, SlashCommand]:
|
||||
"""取当前上下文可见的 slash 命令（通用 + 产线 + 角色）。"""
|
||||
result: Dict[str, SlashCommand] = {}
|
||||
for name, cmd in _commands.items():
|
||||
if cmd.scope == "global":
|
||||
result[name] = cmd
|
||||
elif cmd.scope == "pipeline" and pipeline_id and cmd.scope_id == pipeline_id:
|
||||
result[name] = cmd
|
||||
elif cmd.scope == "role" and pipeline_id and role \
|
||||
and cmd.scope_id == f"{pipeline_id}:{role}":
|
||||
result[name] = cmd
|
||||
return result
|
||||
|
||||
|
||||
def resolve_slash_command(text: str, pipeline_id: str = "",
|
||||
role: str = "") -> Optional[SlashCommand]:
|
||||
"""从用户输入解析 slash 命令（text 以 "/" 开头）。"""
|
||||
if not text or not text.startswith("/"):
|
||||
return None
|
||||
body = text[1:].strip()
|
||||
if not body:
|
||||
return None
|
||||
# 命令名 = 第一个空格前的 token
|
||||
name = body.split()[0] if " " in body else body
|
||||
visible = get_slash_commands(pipeline_id, role)
|
||||
return visible.get(name)
|
||||
|
||||
|
||||
def parse_slash_args(text: str) -> str:
|
||||
"""提取命令名之后的参数字符串。"""
|
||||
body = text[1:].strip() if text.startswith("/") else text.strip()
|
||||
parts = body.split(None, 1)
|
||||
return parts[1] if len(parts) > 1 else ""
|
||||
|
||||
|
||||
def list_commands() -> List[SlashCommand]:
|
||||
"""列出所有已注册命令。"""
|
||||
return list(_commands.values())
|
||||
Loading…
x
Reference in New Issue
Block a user