Compare commits
1 Commits
main
...
feature/ag
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
4c9be85684 |
11
README.md
11
README.md
@ -23,6 +23,7 @@ pam_deploy_graph/
|
||||
fake_runner.py # 测试用 runner,不访问真实环境
|
||||
output_parser.py # 解析 key=value、MCP JSON、待确认回滚标记
|
||||
skill_policy.py # 从 PAM_AUTO_DEPLY_SKILL.md 加载 Skill 策略
|
||||
tool_catalog.py # 统一 action tool schema、tool 摘要和计划动作归一化
|
||||
config_writer.py # 生成脚本 action 所需 config 文件
|
||||
checkpoint_store.py # 业务 checkpoint JSON 读写
|
||||
params_loader.py # 读取 JSON 或 config.txt 风格参数文件
|
||||
@ -68,6 +69,7 @@ packaging/
|
||||
- 实现人工确认入口:`confirm --decision approve|reject` 只处理待确认回滚。
|
||||
- 实现 checkpoint 自动保存和 `resume` 续跑:全局步骤、成功 IP、单 IP 已完成 action 会跳过。
|
||||
- 实现 LLM structured output 骨架:意图识别、参数抽取、部署计划生成。
|
||||
- 增加 LLM 双模式决策:`fixed_runtime` 和 `agentic_skill`,并把模式决策结构化输出纳入 analyze/chat 主链路。
|
||||
- 实现 OpenAI-compatible 真实 LLM client,支持 `base_url` / `model` 配置,`api_key` 可为空。
|
||||
- 固化真实 LLM 提示词:意图识别、参数抽取、部署计划生成均要求 JSON structured output。
|
||||
- 增加规则 fallback `RuleBasedLlmClient`,用于本地开发和测试。
|
||||
@ -82,6 +84,11 @@ packaging/
|
||||
- chat 在开发环境可选启用 `rich` / `prompt_toolkit`;PyInstaller 打包环境默认使用普通文本输入,避免交互兼容问题。
|
||||
- chat 执行前会归一化参数并展示实际写入脚本配置的值;`script_only` / `hybrid_node_mcp` 会提前检查 `ZIP_FILE_PATH` 是否存在。
|
||||
- chat 执行中会播报每个 action 的开始、完成或失败;action 执行失败会停在当前 checkpoint,不再误报 LangGraph 不可用。
|
||||
- `SkillPolicy` 不再只是加载元数据;已接入 LLM prompt、planned actions 归一化和 runtime 动作裁剪。
|
||||
- 新增统一 action tool schema,脚本 action 与 MCP action 通过同一套受控 tool 描述暴露给 LLM。
|
||||
- `AgentState` 已引入 `execution_mode`、`planned_actions`、`mode_reason` 等字段,支持“按计划动作子集执行”。
|
||||
- 增加 action 后 LLM/规则诊断,可通过 `--analyze-actions` 或 `llm action-analysis on` 显式开启。
|
||||
- 添加基础测试,当前本地结果为 `54 passed, 2 skipped`。
|
||||
- 每个 action 完成后都会进入一次 LLM/规则审核;如果审核建议停止,流程会暂停并给出建议,等待用户 `resume`。
|
||||
- `--analyze-actions` 和 `llm action-analysis on` 改为只控制是否把详细审核结果写入 `events`,不再控制审核是否执行。
|
||||
- chat 会播报 action 审核开始、审核完成和审核失败,避免黑盒执行。
|
||||
@ -269,6 +276,9 @@ python -m pam_deploy_graph.cli chat --config doc_scripts/config.txt.example --st
|
||||
|
||||
```text
|
||||
PAM> 请用 MCP 预演部署 HET PAM Node 版本 2.0.5,不要动环境
|
||||
PAM> analyze 请按 skill 自主编排并自动选择工具,帮我排查 HET PAM Node 部署异常
|
||||
- mode: agentic_skill
|
||||
- mode_reason: 用户明确要求按 skill 自主编排,或任务更偏探索/诊断。
|
||||
PAM> preview
|
||||
PAM> set VERSION_NUMBER=2.0.6
|
||||
PAM> load params runtime/override.txt
|
||||
@ -290,6 +300,7 @@ PAM> resume
|
||||
PAM> exit
|
||||
```
|
||||
|
||||
`chat` 默认仍要求在会话内显式输入 `run`,并确认参数、目标 IP 范围和最终执行后才会执行 action。输入 `你好`、`hello` 这类问候不会触发 LLM/结构化分析;需要分析部署需求时可直接描述部署任务,或显式使用 `analyze <需求>`。分析阶段现在会额外输出 `mode`、`mode_reason` 和 `planned_actions`,用于区分 `fixed_runtime` 与 `agentic_skill`。如果某个 IP 失败,会通过 LangGraph interrupt 暂停并提示输入 `approve` 或 `reject [原因]`,确认后恢复同一个图线程继续执行。`chat` 也支持 `--llm-base-url` / `--llm-api-key` / `--llm-model`、`--mcp-config` 和 `--analyze-actions`。
|
||||
`chat` 默认仍要求在会话内显式输入 `run`,并确认参数、目标 IP 范围和最终执行后才会执行 action。输入 `你好`、`hello` 这类问候不会触发 LLM/结构化分析;需要分析部署需求时可直接描述部署任务,或显式使用 `analyze <需求>`。每个 action 完成后都会自动进入一次 LLM/规则审核,并播报审核开始/结束;如果审核建议停止或审核本身失败,流程会暂停并输出建议,等待用户决定是否 `resume`。`--analyze-actions` 仅控制详细审核结果是否写入 `events`。执行中可按 `Ctrl+C` 中断,chat 会保存当前 checkpoint 并把流程标记为 `user_interrupted`。`set KEY=VALUE` 和 `load params <路径>` 会把更新同步到当前运行 state、`config.txt` 和 checkpoint。`chat` 也支持 `--llm-base-url` / `--llm-api-key` / `--llm-model` / `--llm-action-analysis-prompt-file`、`--mcp-config` 和 `--analyze-actions`。
|
||||
|
||||
预演:
|
||||
|
||||
61
docs/todo.md
61
docs/todo.md
@ -22,3 +22,64 @@
|
||||
- [x] 对 LLM 输入做脱敏,禁止把 `CLIENT_SECRET`、token、Authorization、完整日志原文发送给模型。
|
||||
- [x] 每个 action 都会执行审核;`--analyze-actions` 或 `llm action-analysis on` 只控制是否把详细审核结果写入 `events`。
|
||||
- [x] 支持通过 `--llm-action-analysis-prompt-file`、环境变量或 chat 命令热加载自定义 action 审核提示词。
|
||||
- [x] 通过 `--analyze-actions` 或 `llm action-analysis on` 显式开启,真实部署默认不启用。
|
||||
|
||||
## 双模式 Agent 改造
|
||||
|
||||
- [x] 明确双模式入口:`fixed_runtime` 和 `agentic_skill`,由 LLM 基于用户意图、风险和任务类型先做模式决策。
|
||||
- [x] 为 LLM 增加模式决策 structured output schema,至少输出:`mode`、`reason`、`risk_level`、`requires_confirmation`。
|
||||
- [x] 保留固定 runtime 为默认主链路:标准部署、高风险动作、回滚和批量升级优先走确定性流程。
|
||||
- [x] 仅在诊断类、探索类、半结构化任务,或用户明确要求“按 skill 自主编排”时进入 `agentic_skill` 模式。
|
||||
|
||||
## Skill 驱动执行
|
||||
|
||||
- [x] 将 `PAM_AUTO_DEPLY_SKILL.md` 从“只加载元数据”升级为真正驱动执行的规则源。
|
||||
- [x] 让 `SkillPolicy` 进入 LLM prompt,明确 allowed actions、required params、required confirmations、forbidden actions。
|
||||
- [x] 让 `SkillPolicy` 进入 runtime/graph 路由,用于裁剪不可执行 action,而不是只挂在 `PamDeployAgent.skill_policy` 上。
|
||||
- [ ] 支持把 skill 中的执行顺序、确认点、回滚约束映射到 LangGraph 节点和边。
|
||||
- [ ] 评估是否需要把 markdown skill 拆成“机器可读配置 + 人类可读说明”双文件结构,降低解析歧义。
|
||||
|
||||
## Tool Schema 收口
|
||||
|
||||
- [x] 不给 LLM 原始 shell 或 powershell 执行权限,只允许调用受控 typed tools。
|
||||
- [x] 把现有脚本 action 统一抽象为标准 tool schema,例如 `create_version`、`upload_package`、`publish_version`、`get_online_ips`、`upgrade_ip`、`verify_ip`、`rollback_ip`。
|
||||
- [x] 把 MCP action 统一抽象为同一套 tool schema,避免“大模型看见的是脚本”和“大模型看见的是 MCP tool”两套接口。
|
||||
- [ ] 为每个 tool 明确入参、出参、是否幂等、是否高风险、是否需要人工确认。
|
||||
- [ ] 在 tool 层增加白名单、步数限制、超时限制和调用次数限制,避免 agentic 模式失控。
|
||||
|
||||
## MCP 与 LLM 协同
|
||||
|
||||
- [ ] 明确 MCP 在 agentic 模式中的角色:作为 LLM 可调用 tool,而不是仅作为固定 runtime 的后端。
|
||||
- [ ] 为 MCP tool 增加面向 LLM 的描述信息,至少包含用途、必填参数、成功返回字段、失败语义。
|
||||
- [ ] 支持在 agentic 模式下先 `list_tools` 再做工具匹配,但最终仍映射回受控 action/tool schema。
|
||||
- [ ] 评估是否需要增加 MCP tool 结果摘要层,避免把原始复杂返回直接喂回 LLM。
|
||||
|
||||
## LangGraph Workflow 收敛
|
||||
|
||||
- [ ] 合并当前 `graph.py` 和 `langgraph_runtime.py` 的职责,避免维护两套相近的部署图工厂。
|
||||
- [ ] 在 LangGraph 中补齐完整工作流:`analyze -> clarify -> decide_mode -> confirm -> execute -> diagnose -> resume`。
|
||||
- [ ] 将人工确认点从“只覆盖 rollback”扩展到模式切换、高风险 tool 调用、关键参数缺失修正等场景。
|
||||
- [ ] 为 `agentic_skill` 模式增加 LangGraph tool loop 节点,而不是仅复用当前固定 action 图。
|
||||
- [ ] 明确单进程 LangGraph checkpointer 和跨进程业务 checkpoint 的职责边界,避免双状态源混乱。
|
||||
|
||||
## Chat 交互升级
|
||||
|
||||
- [x] 在 chat 中增加“当前模式”展示,让用户知道当前是固定 runtime 还是 agentic skill。
|
||||
- [ ] 在 chat 中增加模式切换确认,例如从固定 runtime 切到 agentic 模式前提示风险和限制。
|
||||
- [ ] 在 chat 中展示大模型最近一次 tool 决策摘要,包括原因、目标 action/tool 和关键参数。
|
||||
- [ ] 为 agentic 模式增加中途打断、继续、回退到固定 runtime 的命令能力。
|
||||
|
||||
## 安全与可观测性
|
||||
|
||||
- [ ] 为 agentic 模式增加完整审计日志:用户输入、LLM 决策、tool 调用、tool 返回、确认记录。
|
||||
- [ ] 在高风险 tool 调用前增加统一确认策略,不允许 LLM 自行越过人工确认直接执行。
|
||||
- [ ] 对发送给 LLM 的 tool 返回做脱敏和截断,避免把脚本原始日志、敏感配置和 token 直接暴露给模型。
|
||||
- [ ] 增加失败保护:当 LLM structured output 非法、模式决策异常或 tool loop 超限时自动降级回固定 runtime。
|
||||
|
||||
## 测试与验收
|
||||
|
||||
- [x] 增加双模式决策测试,覆盖固定 runtime 与 agentic skill 的分流条件。
|
||||
- [x] 增加 skill 约束测试,确认 forbidden action、required confirmation、required params 会真正生效。
|
||||
- [ ] 增加 agentic tool loop 测试,验证 LLM 只能调用白名单工具,不能执行任意命令。
|
||||
- [ ] 增加 MCP + LLM 协同测试,验证 MCP tools 可以被统一 schema 包装并参与自主编排。
|
||||
- [ ] 补充文档,明确当前项目是“固定 runtime 为主,agentic skill 为辅”的演进路线,而不是直接替代现有部署流。
|
||||
|
||||
@ -17,11 +17,23 @@ from .checkpoint_store import save_checkpoint
|
||||
from .config_writer import write_config
|
||||
from .constants import DEFAULT_PARAMS, GLOBAL_ACTION_SEQUENCE, IP_ACTION_SEQUENCE, REQUIRED_PARAMS
|
||||
from .fake_runner import FakeActionRunner
|
||||
from .llm import LlmClient, RuleBasedLlmClient, validate_deploy_plan, validate_intent_result
|
||||
from .llm import LlmClient, RuleBasedLlmClient, validate_deploy_plan, validate_intent_result, validate_mode_decision
|
||||
from .mcp_runner import McpActionRunner
|
||||
from .models import (
|
||||
ActionResult,
|
||||
AgentExecutionMode,
|
||||
AgentState,
|
||||
ExecutionStrategy,
|
||||
LlmDeployPlan,
|
||||
LlmIntentResult,
|
||||
LlmModeDecision,
|
||||
LlmParamResult,
|
||||
SkillPolicy,
|
||||
)
|
||||
from .models import ActionResult, AgentState, ExecutionStrategy, LlmActionAnalysis, LlmDeployPlan, LlmIntentResult, LlmParamResult
|
||||
from .script_runner import ScriptActionRunner, select_script_entry
|
||||
from .skill_policy import load_skill_policy
|
||||
from .tool_catalog import normalize_planned_actions, tool_summaries
|
||||
|
||||
REQUIRED_ACTION_VALUES = {
|
||||
"upload-package": ("HASH_CODE",),
|
||||
@ -76,23 +88,67 @@ class PamDeployAgent:
|
||||
strategy: ExecutionStrategy,
|
||||
) -> LlmDeployPlan:
|
||||
"""根据参数、意图和执行策略生成部署计划。"""
|
||||
plan = self.llm_client.generate_plan(params=params, intent=intent, strategy=strategy)
|
||||
plan = self.llm_client.generate_plan(
|
||||
params=params,
|
||||
intent=intent,
|
||||
strategy=strategy,
|
||||
skill_policy=self._skill_policy_payload(self.skill_policy),
|
||||
tool_summaries=self.tool_summaries(strategy),
|
||||
)
|
||||
validate_deploy_plan(plan)
|
||||
return plan
|
||||
|
||||
def decide_execution_mode(
|
||||
self,
|
||||
*,
|
||||
text: str,
|
||||
params: dict[str, Any],
|
||||
intent: str,
|
||||
strategy: ExecutionStrategy,
|
||||
) -> LlmModeDecision:
|
||||
"""根据用户请求、skill 约束和工具能力决定执行模式。"""
|
||||
allowed_modes = list(self.skill_policy.allowed_execution_modes)
|
||||
decision = self.llm_client.decide_execution_mode(
|
||||
text=text,
|
||||
params=params,
|
||||
intent=intent,
|
||||
strategy=strategy,
|
||||
allowed_modes=allowed_modes,
|
||||
tool_summaries=self.tool_summaries(strategy),
|
||||
)
|
||||
validate_mode_decision(decision, allowed_modes)
|
||||
return decision
|
||||
|
||||
def tool_summaries(self, strategy: ExecutionStrategy) -> list[dict[str, str]]:
|
||||
"""返回当前 skill 和策略下允许暴露给 LLM 的 tool 摘要。"""
|
||||
return tool_summaries(self.skill_policy, strategy)
|
||||
|
||||
def analyze_request(self, text: str, base_params: dict[str, Any] | None = None) -> dict[str, Any]:
|
||||
"""完成意图识别、参数抽取和计划生成,供 analyze/chat 使用。"""
|
||||
intent = self.understand_request(text)
|
||||
params = self.extract_params(text, base_params)
|
||||
strategy = self._choose_strategy(intent.strategy_preference)
|
||||
plan = self.generate_plan(
|
||||
params={**DEFAULT_PARAMS, **params.extracted_params},
|
||||
merged_params = {**DEFAULT_PARAMS, **params.extracted_params}
|
||||
mode_decision = self.decide_execution_mode(
|
||||
text=text,
|
||||
params=merged_params,
|
||||
intent=intent.intent,
|
||||
strategy=strategy,
|
||||
)
|
||||
plan = self.generate_plan(
|
||||
params=merged_params,
|
||||
intent=intent.intent,
|
||||
strategy=strategy,
|
||||
)
|
||||
plan.planned_actions = normalize_planned_actions(
|
||||
plan.planned_actions,
|
||||
policy=self.skill_policy,
|
||||
mode=mode_decision.mode,
|
||||
)
|
||||
return {
|
||||
"intent": intent,
|
||||
"params": params,
|
||||
"mode_decision": mode_decision,
|
||||
"plan": plan,
|
||||
}
|
||||
|
||||
@ -116,12 +172,17 @@ class PamDeployAgent:
|
||||
*,
|
||||
params: dict[str, Any],
|
||||
execution_strategy: ExecutionStrategy = "hybrid_node_mcp",
|
||||
execution_mode: AgentExecutionMode = "fixed_runtime",
|
||||
run_id: str | None = None,
|
||||
script_entry: str | None = None,
|
||||
config_path: str | None = None,
|
||||
trace_file_path: str | None = None,
|
||||
checkpoint_path: str | None = None,
|
||||
target_ips: list[str] | None = None,
|
||||
planned_actions: list[str] | None = None,
|
||||
mode_reason: str = "",
|
||||
mode_risk_level: str = "medium",
|
||||
mode_requires_confirmation: bool = True,
|
||||
) -> AgentState:
|
||||
"""创建一次运行所需的 AgentState,并写入脚本配置文件。"""
|
||||
normalized = self.normalize_params(params)
|
||||
@ -131,6 +192,11 @@ class PamDeployAgent:
|
||||
actual_config_path = _absolute_path(config_path or runtime_dir / f"config_{actual_run_id}.txt")
|
||||
actual_trace_path = _absolute_path(trace_file_path or Path("logs") / f"api_trace_{actual_run_id}.log")
|
||||
write_config(normalized, actual_config_path)
|
||||
normalized_actions = normalize_planned_actions(
|
||||
planned_actions or [],
|
||||
policy=self.skill_policy,
|
||||
mode=execution_mode,
|
||||
)
|
||||
return AgentState(
|
||||
run_id=actual_run_id,
|
||||
params=normalized,
|
||||
@ -142,6 +208,11 @@ class PamDeployAgent:
|
||||
trace_file_path=str(actual_trace_path),
|
||||
checkpoint_path=checkpoint_path or "",
|
||||
target_ips=target_ips or [],
|
||||
execution_mode=execution_mode,
|
||||
planned_actions=normalized_actions,
|
||||
mode_reason=mode_reason,
|
||||
mode_risk_level=mode_risk_level, # type: ignore[arg-type]
|
||||
mode_requires_confirmation=mode_requires_confirmation,
|
||||
)
|
||||
|
||||
def pause_state(
|
||||
@ -192,6 +263,7 @@ class PamDeployAgent:
|
||||
lines = [
|
||||
"## PAM 部署预览",
|
||||
"",
|
||||
f"- 执行模式: fixed_runtime",
|
||||
f"- 执行策略: {strategy}",
|
||||
f"- PAM_HOME: {home_backend}",
|
||||
f"- PAM_NODE: {node_backend}",
|
||||
@ -220,6 +292,7 @@ class PamDeployAgent:
|
||||
|
||||
def next_global_action(self, state: AgentState) -> str | None:
|
||||
"""返回下一个未完成的全局 action。"""
|
||||
for action in self._planned_global_actions(state):
|
||||
if state.paused:
|
||||
return None
|
||||
for action in GLOBAL_ACTION_SEQUENCE:
|
||||
@ -353,6 +426,9 @@ class PamDeployAgent:
|
||||
if state.pending_confirmation or state.paused:
|
||||
self._save_checkpoint(state)
|
||||
return None
|
||||
planned_ip_actions = self._planned_ip_actions(state)
|
||||
if not planned_ip_actions:
|
||||
return None
|
||||
self._resolve_target_ips(state)
|
||||
for ip in state.target_ips:
|
||||
ip_state = state.ip_states.get(ip)
|
||||
@ -378,7 +454,7 @@ class PamDeployAgent:
|
||||
state.ip_states[ip] = ip_state
|
||||
|
||||
completed_steps = ip_state.setdefault("completed_steps", [])
|
||||
for action in IP_ACTION_SEQUENCE:
|
||||
for action in planned_ip_actions:
|
||||
if action not in completed_steps:
|
||||
return ip, action
|
||||
|
||||
@ -801,7 +877,9 @@ class PamDeployAgent:
|
||||
"""生成给 LLM action 分析使用的脱敏状态摘要。"""
|
||||
return {
|
||||
"run_id": state.run_id,
|
||||
"execution_mode": state.execution_mode,
|
||||
"execution_strategy": state.execution_strategy,
|
||||
"planned_actions": state.planned_actions,
|
||||
"completed_global_steps": state.completed_global_steps,
|
||||
"online_ip_count": len(state.online_ips),
|
||||
"target_ips": state.target_ips,
|
||||
@ -851,7 +929,9 @@ class PamDeployAgent:
|
||||
lines = [
|
||||
"## PAM 智能部署报告",
|
||||
"",
|
||||
f"- 执行模式: {state.execution_mode}",
|
||||
f"- 执行策略: {state.execution_strategy}",
|
||||
f"- 模式原因: {state.mode_reason or '-'}",
|
||||
f"- 机场: {state.params['AIRPORT_CODE']}",
|
||||
f"- 应用: {state.params['APP_NAME']}",
|
||||
f"- 模块: {state.params['MODULE_NAME']}",
|
||||
@ -861,6 +941,7 @@ class PamDeployAgent:
|
||||
f"- 成功: {success}",
|
||||
f"- 失败: {failed}",
|
||||
f"- 待确认: {state.pending_confirmation or '-'}",
|
||||
f"- 计划动作: {', '.join(state.planned_actions) if state.planned_actions else '-'}",
|
||||
f"- 暂停状态: {'是' if state.paused else '否'}",
|
||||
f"- 暂停原因: {state.pause_reason or '-'}",
|
||||
"",
|
||||
@ -879,6 +960,33 @@ class PamDeployAgent:
|
||||
)
|
||||
return "\n".join(lines)
|
||||
|
||||
def _planned_global_actions(self, state: AgentState) -> list[str]:
|
||||
"""返回当前 state 下应执行的全局 action 列表。"""
|
||||
if state.planned_actions:
|
||||
return [action for action in state.planned_actions if action in GLOBAL_ACTION_SEQUENCE]
|
||||
return list(GLOBAL_ACTION_SEQUENCE)
|
||||
|
||||
def _planned_ip_actions(self, state: AgentState) -> list[str]:
|
||||
"""返回当前 state 下应执行的逐 IP action 列表。"""
|
||||
if state.planned_actions:
|
||||
return [action for action in state.planned_actions if action in IP_ACTION_SEQUENCE]
|
||||
return list(IP_ACTION_SEQUENCE)
|
||||
|
||||
def _skill_policy_payload(self, policy: SkillPolicy) -> dict[str, Any]:
|
||||
"""把 SkillPolicy 转成可发送给 LLM 的简化 JSON。"""
|
||||
return {
|
||||
"name": policy.name,
|
||||
"description": policy.description,
|
||||
"allowed_execution_modes": list(policy.allowed_execution_modes),
|
||||
"allowed_modes": list(policy.allowed_modes),
|
||||
"allowed_actions": list(policy.allowed_actions),
|
||||
"required_confirmations": list(policy.required_confirmations),
|
||||
"required_params": list(policy.required_params),
|
||||
"action_sequence": list(policy.action_sequence),
|
||||
"ip_action_sequence": list(policy.ip_action_sequence),
|
||||
"forbidden_actions": list(policy.forbidden_actions),
|
||||
}
|
||||
|
||||
|
||||
def _absolute_path(path: str | Path) -> Path:
|
||||
"""把传给脚本的文件路径转换为绝对路径,避免 cwd 切换后读错文件。"""
|
||||
|
||||
@ -29,12 +29,17 @@ def build_langgraph(agent: PamDeployAgent | None = None, flow: GraphFlow = "depl
|
||||
agent_state = runtime.create_state(
|
||||
params=state["params"],
|
||||
execution_strategy=state.get("execution_strategy", "hybrid_node_mcp"),
|
||||
execution_mode=state.get("execution_mode", "fixed_runtime"),
|
||||
run_id=state.get("run_id"),
|
||||
script_entry=state.get("script_entry"),
|
||||
config_path=state.get("config_path"),
|
||||
trace_file_path=state.get("trace_file_path"),
|
||||
checkpoint_path=state.get("checkpoint_path"),
|
||||
target_ips=state.get("target_ips"),
|
||||
planned_actions=state.get("planned_actions"),
|
||||
mode_reason=state.get("mode_reason", ""),
|
||||
mode_risk_level=state.get("mode_risk_level", "medium"),
|
||||
mode_requires_confirmation=state.get("mode_requires_confirmation", True),
|
||||
)
|
||||
return {"agent_state": agent_state}
|
||||
|
||||
|
||||
@ -18,6 +18,7 @@ from .langgraph_runtime import LangGraphDeploymentRuntime, LangGraphRunResult
|
||||
from .llm import build_llm_client
|
||||
from .llm.rule_based import RuleBasedLlmClient
|
||||
from .mcp_factory import build_mcp_runner_from_config
|
||||
from .models import AgentExecutionMode, AgentState, ExecutionStrategy
|
||||
from .models import AgentState, ExecutionStrategy
|
||||
from .params_loader import load_params_file
|
||||
|
||||
@ -69,6 +70,7 @@ class InteractiveCliSession:
|
||||
self.agent = agent
|
||||
self.params = dict(params)
|
||||
self.strategy = strategy
|
||||
self.execution_mode: AgentExecutionMode = "fixed_runtime"
|
||||
self.checkpoint_path = checkpoint_path or _default_checkpoint_path()
|
||||
self.target_ips = list(target_ips or [])
|
||||
self.input = _build_prompt_input(input_func)
|
||||
@ -186,9 +188,11 @@ class InteractiveCliSession:
|
||||
self.last_analysis = result
|
||||
param_result = result["params"]
|
||||
intent_result = result["intent"]
|
||||
mode_decision = result["mode_decision"]
|
||||
plan = result["plan"]
|
||||
self.params = dict(param_result.extracted_params)
|
||||
self.strategy = _choose_strategy(intent_result.strategy_preference, self.strategy)
|
||||
self.execution_mode = mode_decision.mode
|
||||
|
||||
user_ips = param_result.extracted_control.get("user_specified_ips")
|
||||
if isinstance(user_ips, list):
|
||||
@ -198,8 +202,12 @@ class InteractiveCliSession:
|
||||
safe_payload = redact_mapping({key: asdict(value) for key, value in result.items()})
|
||||
self.output("已生成结构化理解:")
|
||||
self.output(f"- intent: {intent_result.intent}")
|
||||
self.output(f"- mode: {mode_decision.mode}")
|
||||
self.output(f"- strategy: {self.strategy}")
|
||||
self.output(f"- mode_reason: {mode_decision.reason}")
|
||||
self.output(f"- summary: {plan.summary}")
|
||||
if plan.planned_actions:
|
||||
self.output("- planned_actions: " + ", ".join(plan.planned_actions))
|
||||
if param_result.missing_required_params:
|
||||
self.output("- missing: " + ", ".join(param_result.missing_required_params))
|
||||
if self.target_ips:
|
||||
@ -322,6 +330,7 @@ class InteractiveCliSession:
|
||||
self.checkpoint_path = str(checkpoint)
|
||||
self.params = dict(self.state.params)
|
||||
self.strategy = self.state.execution_strategy
|
||||
self.execution_mode = self.state.execution_mode
|
||||
self.target_ips = list(self.state.target_ips)
|
||||
self.graph_runtime = None
|
||||
self.output(f"已加载 checkpoint: {checkpoint}")
|
||||
@ -381,8 +390,13 @@ class InteractiveCliSession:
|
||||
self.state = self.agent.create_state(
|
||||
params=self.params,
|
||||
execution_strategy=self.strategy,
|
||||
execution_mode=self.execution_mode,
|
||||
checkpoint_path=self.checkpoint_path,
|
||||
target_ips=self.target_ips,
|
||||
planned_actions=self._current_planned_actions(),
|
||||
mode_reason=self._current_mode_reason(),
|
||||
mode_risk_level=self._current_mode_risk_level(),
|
||||
mode_requires_confirmation=self._current_mode_requires_confirmation(),
|
||||
)
|
||||
self.graph_runtime = None
|
||||
self._execute_current_state()
|
||||
@ -407,6 +421,7 @@ class InteractiveCliSession:
|
||||
return
|
||||
self.state = load_agent_state(checkpoint)
|
||||
self.state.checkpoint_path = self.state.checkpoint_path or str(checkpoint)
|
||||
self.execution_mode = self.state.execution_mode
|
||||
if self.state.paused:
|
||||
self.state = self.agent.resume_state(self.state)
|
||||
if self.graph_runtime and self.graph_runtime.waiting_confirmation:
|
||||
@ -534,6 +549,36 @@ class InteractiveCliSession:
|
||||
if self.state.pending_confirmation:
|
||||
self._print_confirmation()
|
||||
|
||||
def _current_planned_actions(self) -> list[str]:
|
||||
"""返回当前分析结果中的 planned actions;没有分析结果时返回空列表。"""
|
||||
if not self.last_analysis:
|
||||
return []
|
||||
plan = self.last_analysis.get("plan")
|
||||
if plan is None:
|
||||
return []
|
||||
return list(getattr(plan, "planned_actions", []))
|
||||
|
||||
def _current_mode_reason(self) -> str:
|
||||
"""返回最近一次模式决策原因。"""
|
||||
if not self.last_analysis:
|
||||
return ""
|
||||
decision = self.last_analysis.get("mode_decision")
|
||||
return str(getattr(decision, "reason", ""))
|
||||
|
||||
def _current_mode_risk_level(self) -> str:
|
||||
"""返回最近一次模式决策风险等级。"""
|
||||
if not self.last_analysis:
|
||||
return "medium"
|
||||
decision = self.last_analysis.get("mode_decision")
|
||||
return str(getattr(decision, "risk_level", "medium"))
|
||||
|
||||
def _current_mode_requires_confirmation(self) -> bool:
|
||||
"""返回最近一次模式决策是否要求确认。"""
|
||||
if not self.last_analysis:
|
||||
return True
|
||||
decision = self.last_analysis.get("mode_decision")
|
||||
return bool(getattr(decision, "requires_confirmation", True))
|
||||
|
||||
def _confirm(self, *, approved: bool, note: str = "") -> None:
|
||||
"""处理 approve/reject 命令。"""
|
||||
if self.state is None:
|
||||
|
||||
@ -4,7 +4,7 @@ from .base import LlmClient
|
||||
from .factory import build_llm_client
|
||||
from .openai_compatible import OpenAICompatibleLlmClient
|
||||
from .rule_based import RuleBasedLlmClient
|
||||
from .validators import validate_deploy_plan, validate_intent_result
|
||||
from .validators import validate_deploy_plan, validate_intent_result, validate_mode_decision
|
||||
|
||||
__all__ = [
|
||||
"LlmClient",
|
||||
@ -13,4 +13,5 @@ __all__ = [
|
||||
"build_llm_client",
|
||||
"validate_deploy_plan",
|
||||
"validate_intent_result",
|
||||
"validate_mode_decision",
|
||||
]
|
||||
|
||||
@ -10,6 +10,7 @@ from pam_deploy_graph.models import (
|
||||
LlmActionAnalysis,
|
||||
LlmDeployPlan,
|
||||
LlmIntentResult,
|
||||
LlmModeDecision,
|
||||
LlmParamResult,
|
||||
)
|
||||
|
||||
@ -31,10 +32,25 @@ class LlmClient(Protocol):
|
||||
params: dict[str, Any],
|
||||
intent: str,
|
||||
strategy: ExecutionStrategy,
|
||||
skill_policy: dict[str, Any],
|
||||
tool_summaries: list[dict[str, Any]],
|
||||
) -> LlmDeployPlan:
|
||||
"""根据参数和意图生成部署计划。"""
|
||||
...
|
||||
|
||||
def decide_execution_mode(
|
||||
self,
|
||||
*,
|
||||
text: str,
|
||||
params: dict[str, Any],
|
||||
intent: str,
|
||||
strategy: ExecutionStrategy,
|
||||
allowed_modes: list[str],
|
||||
tool_summaries: list[dict[str, Any]],
|
||||
) -> LlmModeDecision:
|
||||
"""决定进入固定 runtime 还是 agentic skill 模式。"""
|
||||
...
|
||||
|
||||
def analyze_action_result(
|
||||
self,
|
||||
*,
|
||||
|
||||
@ -20,10 +20,10 @@ from pam_deploy_graph.constants import (
|
||||
REQUIRED_PARAMS,
|
||||
SENSITIVE_KEYS,
|
||||
)
|
||||
from pam_deploy_graph.models import ExecutionStrategy, LlmDeployPlan, LlmIntentResult, LlmParamResult
|
||||
from pam_deploy_graph.models import ExecutionStrategy, LlmDeployPlan, LlmIntentResult, LlmModeDecision, LlmParamResult
|
||||
from pam_deploy_graph.models import ActionResult, LlmActionAnalysis
|
||||
|
||||
from .prompts import ACTION_ANALYSIS_PROMPT, INTENT_PROMPT, PARAM_PROMPT, PLAN_PROMPT, SYSTEM_PROMPT
|
||||
from .prompts import ACTION_ANALYSIS_PROMPT, INTENT_PROMPT, MODE_PROMPT, PARAM_PROMPT, PLAN_PROMPT, SYSTEM_PROMPT
|
||||
|
||||
JsonTransport = Callable[[str, dict[str, str], dict[str, Any], float], dict[str, Any]]
|
||||
|
||||
@ -107,6 +107,8 @@ class OpenAICompatibleLlmClient:
|
||||
params: dict[str, Any],
|
||||
intent: str,
|
||||
strategy: ExecutionStrategy,
|
||||
skill_policy: dict[str, Any],
|
||||
tool_summaries: list[dict[str, Any]],
|
||||
) -> LlmDeployPlan:
|
||||
"""调用 LLM 生成部署计划。"""
|
||||
payload = self._complete_json(
|
||||
@ -115,6 +117,8 @@ class OpenAICompatibleLlmClient:
|
||||
"params": _redact_sensitive(params),
|
||||
"intent": intent,
|
||||
"execution_strategy": strategy,
|
||||
"skill_policy": skill_policy,
|
||||
"tool_summaries": tool_summaries,
|
||||
"allowed_actions": list(ALLOWED_ACTIONS),
|
||||
"global_action_sequence": list(GLOBAL_ACTION_SEQUENCE),
|
||||
"ip_action_sequence": list(IP_ACTION_SEQUENCE),
|
||||
@ -129,6 +133,35 @@ class OpenAICompatibleLlmClient:
|
||||
execution_strategy=_string(payload, "execution_strategy", strategy), # type: ignore[arg-type]
|
||||
)
|
||||
|
||||
def decide_execution_mode(
|
||||
self,
|
||||
*,
|
||||
text: str,
|
||||
params: dict[str, Any],
|
||||
intent: str,
|
||||
strategy: ExecutionStrategy,
|
||||
allowed_modes: list[str],
|
||||
tool_summaries: list[dict[str, Any]],
|
||||
) -> LlmModeDecision:
|
||||
"""调用 LLM 决定本次任务进入固定 runtime 或 agentic skill。"""
|
||||
payload = self._complete_json(
|
||||
MODE_PROMPT,
|
||||
{
|
||||
"user_text": text,
|
||||
"params": _redact_sensitive(params),
|
||||
"intent": intent,
|
||||
"execution_strategy": strategy,
|
||||
"allowed_modes": allowed_modes,
|
||||
"tool_summaries": tool_summaries,
|
||||
},
|
||||
)
|
||||
return LlmModeDecision(
|
||||
mode=_string(payload, "mode", "fixed_runtime"), # type: ignore[arg-type]
|
||||
reason=_string(payload, "reason", ""),
|
||||
risk_level=_string(payload, "risk_level", "medium"), # type: ignore[arg-type]
|
||||
requires_confirmation=bool(payload.get("requires_confirmation", True)),
|
||||
)
|
||||
|
||||
def analyze_action_result(
|
||||
self,
|
||||
*,
|
||||
|
||||
@ -24,6 +24,23 @@ INTENT_PROMPT = """根据用户输入识别意图和执行偏好。
|
||||
}
|
||||
"""
|
||||
|
||||
MODE_PROMPT = """根据用户输入、参数、执行策略和可用工具,决定本次任务应进入固定 runtime 还是 agentic skill 模式。
|
||||
|
||||
输出 JSON schema:
|
||||
{
|
||||
"mode": "fixed_runtime|agentic_skill",
|
||||
"reason": "...",
|
||||
"risk_level": "low|medium|high",
|
||||
"requires_confirmation": true
|
||||
}
|
||||
|
||||
决策原则:
|
||||
- 标准部署、批量升级、回滚、高风险变更默认优先 fixed_runtime。
|
||||
- 诊断类、探索类、半结构化任务,或用户明确要求“按 skill 自主编排/自主调用工具”时,可选 agentic_skill。
|
||||
- 不能因为用户提到 MCP 就自动选择 agentic_skill;MCP 也可以作为 fixed_runtime 的后端。
|
||||
- 只有在允许模式集合中选择 mode。
|
||||
"""
|
||||
|
||||
PARAM_PROMPT = """从用户输入中抽取 PAM 部署参数和控制信息。
|
||||
|
||||
输出 JSON schema:
|
||||
|
||||
@ -9,13 +9,14 @@ from __future__ import annotations
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
from pam_deploy_graph.constants import GLOBAL_ACTION_SEQUENCE, REQUIRED_PARAMS
|
||||
from pam_deploy_graph.constants import GLOBAL_ACTION_SEQUENCE, IP_ACTION_SEQUENCE, REQUIRED_PARAMS
|
||||
from pam_deploy_graph.models import (
|
||||
ActionResult,
|
||||
ExecutionStrategy,
|
||||
LlmActionAnalysis,
|
||||
LlmDeployPlan,
|
||||
LlmIntentResult,
|
||||
LlmModeDecision,
|
||||
LlmParamResult,
|
||||
)
|
||||
|
||||
@ -116,6 +117,8 @@ class RuleBasedLlmClient:
|
||||
params: dict[str, Any],
|
||||
intent: str,
|
||||
strategy: ExecutionStrategy,
|
||||
skill_policy: dict[str, Any],
|
||||
tool_summaries: list[dict[str, Any]],
|
||||
) -> LlmDeployPlan:
|
||||
"""生成确定性的部署计划和风险提示。"""
|
||||
if strategy == "hybrid_node_mcp":
|
||||
@ -139,14 +142,62 @@ class RuleBasedLlmClient:
|
||||
if strategy == "hybrid_node_mcp":
|
||||
risk_notes.append("PAM_HOME 当前没有 MCP 能力,HOME 阶段仍会调用脚本 action。")
|
||||
|
||||
if intent == "query_node_ips":
|
||||
planned_actions = ["get-token", "get-node-url", "get-online-ips"]
|
||||
elif intent == "rollback":
|
||||
planned_actions = ["rollback-ip", "verify-ip", "download-log"]
|
||||
elif intent == "deploy":
|
||||
planned_actions = [*GLOBAL_ACTION_SEQUENCE, *IP_ACTION_SEQUENCE]
|
||||
else:
|
||||
planned_actions = list(GLOBAL_ACTION_SEQUENCE)
|
||||
|
||||
return LlmDeployPlan(
|
||||
summary=summary,
|
||||
risk_notes=risk_notes,
|
||||
planned_actions=list(GLOBAL_ACTION_SEQUENCE),
|
||||
planned_actions=planned_actions,
|
||||
requires_confirmation=intent in ("deploy", "query_node_ips", "rollback"),
|
||||
execution_strategy=strategy,
|
||||
)
|
||||
|
||||
def decide_execution_mode(
|
||||
self,
|
||||
*,
|
||||
text: str,
|
||||
params: dict[str, Any],
|
||||
intent: str,
|
||||
strategy: ExecutionStrategy,
|
||||
allowed_modes: list[str],
|
||||
tool_summaries: list[dict[str, Any]],
|
||||
) -> LlmModeDecision:
|
||||
"""根据关键词规则决定进入固定 runtime 或 agentic skill。"""
|
||||
lowered = text.lower()
|
||||
requested_agentic = any(
|
||||
word in lowered for word in ("自主编排", "按 skill", "自动选择工具", "自动决策", "toolcall", "agentic")
|
||||
)
|
||||
diagnostic_intent = any(word in lowered for word in ("诊断", "排查", "分析异常", "帮我看看", "explore"))
|
||||
high_risk_intent = intent in ("deploy", "rollback") or any(word in lowered for word in ("批量", "升级", "回滚"))
|
||||
|
||||
mode = "fixed_runtime"
|
||||
reason = "标准部署和高风险动作默认走固定 runtime。"
|
||||
risk_level = "high" if high_risk_intent else "medium"
|
||||
requires_confirmation = True
|
||||
|
||||
if requested_agentic or diagnostic_intent:
|
||||
mode = "agentic_skill"
|
||||
reason = "用户明确要求按 skill 自主编排,或任务更偏探索/诊断。"
|
||||
risk_level = "medium"
|
||||
|
||||
if mode not in allowed_modes:
|
||||
mode = allowed_modes[0] if allowed_modes else "fixed_runtime"
|
||||
reason = "原始模式不在 skill 允许集合内,已回退到允许模式。"
|
||||
|
||||
return LlmModeDecision(
|
||||
mode=mode, # type: ignore[arg-type]
|
||||
reason=reason,
|
||||
risk_level=risk_level, # type: ignore[arg-type]
|
||||
requires_confirmation=requires_confirmation,
|
||||
)
|
||||
|
||||
def analyze_action_result(
|
||||
self,
|
||||
*,
|
||||
|
||||
@ -3,9 +3,11 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from pam_deploy_graph.constants import ALLOWED_ACTIONS
|
||||
from pam_deploy_graph.models import LlmDeployPlan, LlmIntentResult
|
||||
from pam_deploy_graph.models import LlmDeployPlan, LlmIntentResult, LlmModeDecision
|
||||
|
||||
VALID_INTENTS = {"deploy", "show_usage", "preview", "query_node_ips", "rollback"}
|
||||
VALID_EXECUTION_MODES = {"fixed_runtime", "agentic_skill"}
|
||||
VALID_RISK_LEVELS = {"low", "medium", "high"}
|
||||
FORBIDDEN_TEXT = ("bash ", "powershell ", "deploy.sh", "deploy.ps1", "CLIENT_SECRET=")
|
||||
|
||||
|
||||
@ -27,3 +29,13 @@ def validate_deploy_plan(plan: LlmDeployPlan) -> None:
|
||||
forbidden = [item for item in FORBIDDEN_TEXT if item.lower() in lowered]
|
||||
if forbidden:
|
||||
raise ValueError(f"计划包含禁止出现的可执行文本: {', '.join(forbidden)}")
|
||||
|
||||
|
||||
def validate_mode_decision(result: LlmModeDecision, allowed_modes: list[str] | None = None) -> None:
|
||||
"""校验模式决策结果是否合法。"""
|
||||
if result.mode not in VALID_EXECUTION_MODES:
|
||||
raise ValueError(f"非法执行模式: {result.mode}")
|
||||
if result.risk_level not in VALID_RISK_LEVELS:
|
||||
raise ValueError(f"非法风险等级: {result.risk_level}")
|
||||
if allowed_modes and result.mode not in allowed_modes:
|
||||
raise ValueError(f"模式 {result.mode} 不在允许集合内")
|
||||
|
||||
@ -7,10 +7,13 @@ from typing import Any, Literal
|
||||
|
||||
BackendName = Literal["mcp", "script", "fake"]
|
||||
ExecutionStrategy = Literal["hybrid_node_mcp", "script_only", "fake"]
|
||||
AgentExecutionMode = Literal["fixed_runtime", "agentic_skill"]
|
||||
IntentName = Literal["deploy", "show_usage", "preview", "query_node_ips", "rollback"]
|
||||
ModePreference = Literal["MCP", "API脚本", "未指定"]
|
||||
StrategyPreference = Literal["hybrid_node_mcp", "script_only", "fake", "未指定"]
|
||||
ActionAnalysisSeverity = Literal["info", "low", "medium", "high"]
|
||||
ModeDecisionRisk = Literal["low", "medium", "high"]
|
||||
ToolScope = Literal["global", "ip"]
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
@ -29,6 +32,21 @@ class ActionResult:
|
||||
error_summary: str = ""
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class ActionToolSpec:
|
||||
"""面向 LLM 和 runtime 的统一 action tool 描述。"""
|
||||
|
||||
name: str
|
||||
action: str
|
||||
scope: ToolScope
|
||||
description: str
|
||||
risk_level: ModeDecisionRisk = "medium"
|
||||
requires_confirmation: bool = False
|
||||
required_runtime_fields: tuple[str, ...] = ()
|
||||
required_param_fields: tuple[str, ...] = ()
|
||||
preferred_backend: str = ""
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class SkillPolicy:
|
||||
"""从 Skill 文档提取出的部署策略约束。"""
|
||||
@ -36,6 +54,7 @@ class SkillPolicy:
|
||||
name: str
|
||||
source_path: str
|
||||
description: str = ""
|
||||
allowed_execution_modes: tuple[AgentExecutionMode, ...] = ("fixed_runtime", "agentic_skill")
|
||||
allowed_modes: tuple[str, ...] = ("MCP", "API脚本")
|
||||
allowed_actions: tuple[str, ...] = ()
|
||||
required_confirmations: tuple[str, ...] = (
|
||||
@ -89,6 +108,16 @@ class LlmDeployPlan:
|
||||
execution_strategy: StrategyPreference = "未指定"
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class LlmModeDecision:
|
||||
"""LLM 给出的执行模式决策。"""
|
||||
|
||||
mode: AgentExecutionMode = "fixed_runtime"
|
||||
reason: str = ""
|
||||
risk_level: ModeDecisionRisk = "medium"
|
||||
requires_confirmation: bool = True
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class LlmActionAnalysis:
|
||||
"""LLM 或规则对单次 action 结果的诊断建议。"""
|
||||
@ -117,6 +146,7 @@ class AgentState:
|
||||
trace_file_path: str = ""
|
||||
node_mcp_server_name: str = ""
|
||||
node_mcp_tool_names: dict[str, str] = field(default_factory=dict)
|
||||
execution_mode: AgentExecutionMode = "fixed_runtime"
|
||||
completed_global_steps: list[str] = field(default_factory=list)
|
||||
hash_code: str = ""
|
||||
node_url: str = ""
|
||||
@ -127,6 +157,10 @@ class AgentState:
|
||||
last_success_step: str = ""
|
||||
last_failed_step: str = ""
|
||||
checkpoint_path: str = ""
|
||||
planned_actions: list[str] = field(default_factory=list)
|
||||
mode_reason: str = ""
|
||||
mode_risk_level: ModeDecisionRisk = "medium"
|
||||
mode_requires_confirmation: bool = True
|
||||
paused: bool = False
|
||||
pause_reason: str = ""
|
||||
review_context: dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
@ -2,6 +2,7 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
from .constants import (
|
||||
@ -15,7 +16,7 @@ from .models import SkillPolicy
|
||||
|
||||
|
||||
def load_skill_policy(path: str | Path) -> SkillPolicy:
|
||||
"""读取 Skill markdown 头部信息,并填充 action/参数策略。"""
|
||||
"""读取 Skill markdown,并提取真正参与执行的策略约束。"""
|
||||
skill_path = Path(path)
|
||||
text = skill_path.read_text(encoding="utf-8")
|
||||
name = "pam-auto-deply"
|
||||
@ -30,13 +31,121 @@ def load_skill_policy(path: str | Path) -> SkillPolicy:
|
||||
elif line.startswith("description:"):
|
||||
description = line.split(":", 1)[1].strip()
|
||||
|
||||
parsed_actions = _parse_allowed_actions(text)
|
||||
action_sequence = _parse_action_sequence(text)
|
||||
ip_action_sequence = tuple(action for action in action_sequence if action in IP_ACTION_SEQUENCE) or IP_ACTION_SEQUENCE
|
||||
global_sequence = tuple(action for action in action_sequence if action in GLOBAL_ACTION_SEQUENCE) or GLOBAL_ACTION_SEQUENCE
|
||||
required_confirmations = _parse_required_confirmations(text)
|
||||
forbidden_actions = _parse_forbidden_actions(text)
|
||||
required_params = _parse_required_params(text)
|
||||
|
||||
return SkillPolicy(
|
||||
name=name,
|
||||
source_path=str(skill_path),
|
||||
description=description,
|
||||
allowed_actions=ALLOWED_ACTIONS,
|
||||
required_params=REQUIRED_PARAMS,
|
||||
allowed_actions=parsed_actions or ALLOWED_ACTIONS,
|
||||
required_params=required_params or REQUIRED_PARAMS,
|
||||
optional_params=DEFAULT_PARAMS.copy(),
|
||||
action_sequence=GLOBAL_ACTION_SEQUENCE,
|
||||
ip_action_sequence=IP_ACTION_SEQUENCE,
|
||||
required_confirmations=required_confirmations or ("params", "target_scope", "rollback"),
|
||||
action_sequence=global_sequence,
|
||||
ip_action_sequence=ip_action_sequence,
|
||||
forbidden_actions=forbidden_actions or ("script-main-flow", "auto-rollback", "modify-deploy-scripts"),
|
||||
)
|
||||
|
||||
|
||||
def _parse_allowed_actions(text: str) -> tuple[str, ...]:
|
||||
"""从 skill 文档的 action 表中提取允许的 action。"""
|
||||
section = _section_body(text, "### 6.3 可用 action")
|
||||
if not section:
|
||||
return ()
|
||||
actions: list[str] = []
|
||||
for raw_line in section.splitlines():
|
||||
line = raw_line.strip()
|
||||
if not line.startswith("|"):
|
||||
continue
|
||||
parts = [item.strip() for item in line.strip("|").split("|")]
|
||||
if not parts or parts[0] in ("action", "---"):
|
||||
continue
|
||||
action = parts[0].strip("` ")
|
||||
if action in ALLOWED_ACTIONS and action not in actions:
|
||||
actions.append(action)
|
||||
return tuple(actions)
|
||||
|
||||
|
||||
def _parse_action_sequence(text: str) -> tuple[str, ...]:
|
||||
"""从主流程章节提取推荐执行顺序。"""
|
||||
section = _section_body(text, "### 4.1 正式部署主流程")
|
||||
if not section:
|
||||
return ()
|
||||
found: list[str] = []
|
||||
for action in [*GLOBAL_ACTION_SEQUENCE, *IP_ACTION_SEQUENCE]:
|
||||
if re.search(rf"\b{re.escape(action)}\b", section) and action not in found:
|
||||
found.append(action)
|
||||
return tuple(found)
|
||||
|
||||
|
||||
def _parse_required_confirmations(text: str) -> tuple[str, ...]:
|
||||
"""从强制确认点章节提取确认类型。"""
|
||||
section = _section_body(text, "### 4.2 主流程中的强制确认点")
|
||||
if not section:
|
||||
return ()
|
||||
confirmations: list[str] = []
|
||||
keyword_map = {
|
||||
"参数确认": "params",
|
||||
"目标 ip": "target_scope",
|
||||
"部署范围": "target_scope",
|
||||
"回滚": "rollback",
|
||||
"间隔策略": "interval_policy",
|
||||
}
|
||||
lowered = section.lower()
|
||||
for keyword, name in keyword_map.items():
|
||||
if keyword in lowered or keyword in section:
|
||||
confirmations.append(name)
|
||||
return tuple(dict.fromkeys(confirmations))
|
||||
|
||||
|
||||
def _parse_forbidden_actions(text: str) -> tuple[str, ...]:
|
||||
"""从禁止事项章节提取禁止项。"""
|
||||
section = _section_body(text, "### 7.5 明确禁止的做法")
|
||||
if not section:
|
||||
return ()
|
||||
forbidden: list[str] = []
|
||||
if "脚本主流程" in section:
|
||||
forbidden.append("script-main-flow")
|
||||
if "自动执行回滚" in section or "自动回滚" in section:
|
||||
forbidden.append("auto-rollback")
|
||||
if "修改脚本" in section or "自动生成" in section:
|
||||
forbidden.append("modify-deploy-scripts")
|
||||
return tuple(dict.fromkeys(forbidden))
|
||||
|
||||
|
||||
def _parse_required_params(text: str) -> tuple[str, ...]:
|
||||
"""从参数表提取必填脚本字段。"""
|
||||
section = _section_body(text, "### 3.1 必填业务参数")
|
||||
if not section:
|
||||
return ()
|
||||
params: list[str] = []
|
||||
for raw_line in section.splitlines():
|
||||
line = raw_line.strip()
|
||||
if not line.startswith("|"):
|
||||
continue
|
||||
parts = [item.strip() for item in line.strip("|").split("|")]
|
||||
if len(parts) < 3 or parts[0] in ("规范字段", "---"):
|
||||
continue
|
||||
script_field = parts[1].strip("` ")
|
||||
required_flag = parts[2]
|
||||
if script_field in REQUIRED_PARAMS and required_flag == "是":
|
||||
params.append(script_field)
|
||||
return tuple(params)
|
||||
|
||||
|
||||
def _section_body(text: str, heading: str) -> str:
|
||||
"""提取指定 markdown heading 到下一同级 heading 之间的正文。"""
|
||||
marker = f"{heading}\n"
|
||||
if marker not in text:
|
||||
return ""
|
||||
_, tail = text.split(marker, 1)
|
||||
matches = list(re.finditer(r"^###\s+", tail, flags=re.MULTILINE))
|
||||
if matches:
|
||||
return tail[: matches[0].start()]
|
||||
return tail
|
||||
|
||||
@ -55,6 +55,41 @@ def test_run_deploy_flow_success(tmp_path: Path):
|
||||
assert all(item["status"] == "SUCCESS" for item in state.ip_states.values())
|
||||
|
||||
|
||||
def test_agentic_state_uses_planned_actions_subset(tmp_path: Path):
|
||||
agent = PamDeployAgent(fake_runner=FakeActionRunner())
|
||||
state = agent.create_state(
|
||||
params=PARAMS,
|
||||
execution_strategy="fake",
|
||||
execution_mode="agentic_skill",
|
||||
planned_actions=["get-token", "get-node-url", "get-online-ips"],
|
||||
config_path=str(tmp_path / "config.txt"),
|
||||
)
|
||||
|
||||
agent.run_deploy_flow(state)
|
||||
|
||||
assert state.execution_mode == "agentic_skill"
|
||||
assert state.planned_actions == ["get-token", "get-node-url", "get-online-ips"]
|
||||
assert state.completed_global_steps == ["get-token", "get-node-url", "get-online-ips"]
|
||||
assert state.ip_states == {}
|
||||
|
||||
|
||||
def test_fixed_runtime_state_also_respects_planned_actions_subset(tmp_path: Path):
|
||||
agent = PamDeployAgent(fake_runner=FakeActionRunner())
|
||||
state = agent.create_state(
|
||||
params=PARAMS,
|
||||
execution_strategy="fake",
|
||||
execution_mode="fixed_runtime",
|
||||
planned_actions=["get-token", "get-node-url", "get-online-ips"],
|
||||
config_path=str(tmp_path / "config.txt"),
|
||||
)
|
||||
|
||||
agent.run_deploy_flow(state)
|
||||
|
||||
assert state.execution_mode == "fixed_runtime"
|
||||
assert state.completed_global_steps == ["get-token", "get-node-url", "get-online-ips"]
|
||||
assert state.ip_states == {}
|
||||
|
||||
|
||||
def test_create_state_writes_absolute_script_config_path_and_normalized_zip(tmp_path: Path):
|
||||
package_path = tmp_path / "pkg.zip"
|
||||
params = {**PARAMS, "ZIP_FILE_PATH": str(package_path)}
|
||||
|
||||
@ -44,7 +44,29 @@ def test_analyze_request_returns_structured_objects():
|
||||
)
|
||||
payload = {key: asdict(value) for key, value in result.items()}
|
||||
assert payload["intent"]["intent"] == "preview"
|
||||
assert payload["mode_decision"]["mode"] == "fixed_runtime"
|
||||
assert payload["plan"]["execution_strategy"] == "hybrid_node_mcp"
|
||||
assert payload["plan"]["planned_actions"][:3] == ["get-token", "create-version", "upload-package"]
|
||||
|
||||
|
||||
def test_rule_based_mode_decision_can_choose_agentic_skill():
|
||||
agent = PamDeployAgent()
|
||||
result = agent.analyze_request(
|
||||
"请按 skill 自主编排并自动选择工具,帮我排查 HET PAM Node 部署异常",
|
||||
{
|
||||
"HOME_BASE_URL": "https://x",
|
||||
"CLIENT_ID": "id",
|
||||
"CLIENT_SECRET": "s",
|
||||
"AIRPORT_CODE": "HET",
|
||||
"APP_NAME": "PAM",
|
||||
"MODULE_NAME": "Node",
|
||||
"VERSION_NUMBER": "2.0.5",
|
||||
"ZIP_FILE_PATH": "C:/pkg.zip",
|
||||
},
|
||||
)
|
||||
|
||||
assert result["mode_decision"].mode == "agentic_skill"
|
||||
assert "自主编排" in result["mode_decision"].reason
|
||||
|
||||
|
||||
def test_analyze_payload_can_be_redacted():
|
||||
|
||||
@ -6,6 +6,11 @@ from pam_deploy_graph.skill_policy import load_skill_policy
|
||||
def test_load_skill_policy_from_doc():
|
||||
policy = load_skill_policy(Path("doc_scripts/PAM_AUTO_DEPLY_SKILL.md"))
|
||||
assert policy.name == "pam-auto-deply"
|
||||
assert "fixed_runtime" in policy.allowed_execution_modes
|
||||
assert "get-token" in policy.allowed_actions
|
||||
assert "CLIENT_SECRET" in policy.required_params
|
||||
|
||||
assert "params" in policy.required_confirmations
|
||||
assert "rollback" in policy.required_confirmations
|
||||
assert "script-main-flow" in policy.forbidden_actions
|
||||
assert policy.action_sequence[0] == "get-token"
|
||||
assert "upgrade-ip" in policy.ip_action_sequence
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user