8d390aa416
调整 workflow 执行逻辑：每个 action 完成后统一进入 LLM/规则审核，审核开始/结果可播报，审核阻断时自动暂停并给出建议 增强 chat 交互：支持执行中 Ctrl+C 中断并保存 checkpoint，后续可 resume 继续 增加运行时热更新能力：支持 set KEY=VALUE 和 load params <路径> 同步更新当前 state、config.txt 和 checkpoint 支持自定义 action 审核提示词：新增 --llm-action-analysis-prompt-file / PAM_LLM_ACTION_ANALYSIS_PROMPT_FILE 新增 prompts/action_review.txt，落地保存当前默认审核提示词，便于后续按基线调整 更新 Linux 打包脚本，将 prompts/action_review.txt 一并带入发布包 同步更新 README、流程图、todo 和打包文档，修正 --analyze-actions 语义说明与 chat 最新行为说明
134 lines
4.1 KiB
Python
134 lines
4.1 KiB
Python
"""PAM 部署 Agent 共享数据模型。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Any, Literal
|
|
|
|
BackendName = Literal["mcp", "script", "fake"]
|
|
ExecutionStrategy = Literal["hybrid_node_mcp", "script_only", "fake"]
|
|
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"]
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class ActionResult:
|
|
"""单个 action 的统一执行结果。"""
|
|
|
|
action: str
|
|
backend: BackendName
|
|
ok: bool
|
|
values: dict[str, Any] = field(default_factory=dict)
|
|
exit_code: int = 0
|
|
tool_name: str = ""
|
|
stdout: str = ""
|
|
stderr: str = ""
|
|
raw_output: str = ""
|
|
error_summary: str = ""
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class SkillPolicy:
|
|
"""从 Skill 文档提取出的部署策略约束。"""
|
|
|
|
name: str
|
|
source_path: str
|
|
description: str = ""
|
|
allowed_modes: tuple[str, ...] = ("MCP", "API脚本")
|
|
allowed_actions: tuple[str, ...] = ()
|
|
required_confirmations: tuple[str, ...] = (
|
|
"params",
|
|
"target_scope",
|
|
"rollback",
|
|
)
|
|
required_params: tuple[str, ...] = ()
|
|
optional_params: dict[str, Any] = field(default_factory=dict)
|
|
action_sequence: tuple[str, ...] = ()
|
|
ip_action_sequence: tuple[str, ...] = ()
|
|
forbidden_actions: tuple[str, ...] = (
|
|
"script-main-flow",
|
|
"auto-rollback",
|
|
"modify-deploy-scripts",
|
|
)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class LlmIntentResult:
|
|
"""LLM 意图识别结果。"""
|
|
|
|
intent: IntentName
|
|
mode_preference: ModePreference = "未指定"
|
|
strategy_preference: StrategyPreference = "未指定"
|
|
confidence: float = 0.0
|
|
reasons: list[str] = field(default_factory=list)
|
|
needs_clarification: bool = False
|
|
clarification_questions: list[str] = field(default_factory=list)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class LlmParamResult:
|
|
"""LLM 参数抽取结果。"""
|
|
|
|
extracted_params: dict[str, Any] = field(default_factory=dict)
|
|
extracted_control: dict[str, Any] = field(default_factory=dict)
|
|
missing_required_params: list[str] = field(default_factory=list)
|
|
ambiguous_fields: list[str] = field(default_factory=list)
|
|
sensitive_fields_present: list[str] = field(default_factory=list)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class LlmDeployPlan:
|
|
"""LLM 生成的部署计划。"""
|
|
|
|
summary: str
|
|
risk_notes: list[str] = field(default_factory=list)
|
|
planned_actions: list[str] = field(default_factory=list)
|
|
requires_confirmation: bool = True
|
|
execution_strategy: StrategyPreference = "未指定"
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class LlmActionAnalysis:
|
|
"""LLM 或规则对单次 action 结果的诊断建议。"""
|
|
|
|
action: str
|
|
has_anomaly: bool = False
|
|
severity: ActionAnalysisSeverity = "info"
|
|
possible_reason: str = ""
|
|
suggested_action: str = ""
|
|
requires_confirmation: bool = False
|
|
should_continue: bool = True
|
|
notes: list[str] = field(default_factory=list)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class AgentState:
|
|
"""一次部署运行的完整状态，可序列化到 checkpoint。"""
|
|
|
|
run_id: str
|
|
params: dict[str, Any]
|
|
execution_strategy: ExecutionStrategy
|
|
action_backends: dict[str, BackendName]
|
|
script_entry: str = ""
|
|
script_base_dir: str = "."
|
|
config_path: str = ""
|
|
trace_file_path: str = ""
|
|
node_mcp_server_name: str = ""
|
|
node_mcp_tool_names: dict[str, str] = field(default_factory=dict)
|
|
completed_global_steps: list[str] = field(default_factory=list)
|
|
hash_code: str = ""
|
|
node_url: str = ""
|
|
online_ips: list[str] = field(default_factory=list)
|
|
target_ips: list[str] = field(default_factory=list)
|
|
ip_states: dict[str, dict[str, Any]] = field(default_factory=dict)
|
|
pending_confirmation: str = ""
|
|
last_success_step: str = ""
|
|
last_failed_step: str = ""
|
|
checkpoint_path: str = ""
|
|
paused: bool = False
|
|
pause_reason: str = ""
|
|
review_context: dict[str, Any] = field(default_factory=dict)
|
|
events: list[dict[str, Any]] = field(default_factory=list)
|