From a11904b7c51b51365946286d0f50cc258658a09e Mon Sep 17 00:00:00 2001 From: dark Date: Mon, 1 Jun 2026 11:21:42 +0800 Subject: [PATCH] =?UTF-8?q?docs/build:=20=E8=A1=A5=E9=BD=90=E4=B8=AD?= =?UTF-8?q?=E6=96=87=E6=B3=A8=E9=87=8A=E3=80=81=E6=B5=81=E7=A8=8B=E5=9B=BE?= =?UTF-8?q?=E5=92=8C=20Linux=20=E8=A7=A3=E5=8E=8B=E5=8D=B3=E7=94=A8?= =?UTF-8?q?=E6=89=93=E5=8C=85=E8=84=9A=E6=9C=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 为 pam_deploy_graph 生产代码补充中文模块、类、函数/方法文档字符串 - 将原有英文说明和主要英文异常提示改为中文 - 新增当前整体逻辑结构流程图文档，覆盖模块结构、执行链路、action 路由、人工确认和 checkpoint 续跑 - 新增 Linux 自带运行环境打包脚本，使用 PyInstaller 生成解压即用目录和 tar.gz - 新增 Linux 打包说明，包含构建命令、运行方式、依赖说明和包大小评估 - 同步 README，补充流程图、打包方式、产物路径和大小预估 - 更新相关测试断言以匹配中文错误提示 --- README.md | 37 ++++++ docs/current_logic_flow.md | 137 ++++++++++++++++++++++ packaging/README_linux_package.md | 55 +++++++++ packaging/build_linux_self_contained.sh | 117 ++++++++++++++++++ packaging/pyinstaller_entry.py | 7 ++ pam_deploy_graph/__init__.py | 3 +- pam_deploy_graph/action_router.py | 16 ++- pam_deploy_graph/agent.py | 56 ++++++--- pam_deploy_graph/checkpoint_store.py | 7 +- pam_deploy_graph/cli.py | 6 +- pam_deploy_graph/config_writer.py | 4 +- pam_deploy_graph/constants.py | 11 +- pam_deploy_graph/fake_runner.py | 10 +- pam_deploy_graph/graph.py | 13 +- pam_deploy_graph/interactive.py | 23 +++- pam_deploy_graph/llm/__init__.py | 2 +- pam_deploy_graph/llm/base.py | 7 +- pam_deploy_graph/llm/factory.py | 5 +- pam_deploy_graph/llm/openai_compatible.py | 39 ++++-- pam_deploy_graph/llm/prompts.py | 2 +- pam_deploy_graph/llm/rule_based.py | 15 ++- pam_deploy_graph/llm/validators.py | 12 +- pam_deploy_graph/mcp_client.py | 34 +++--- pam_deploy_graph/mcp_runner.py | 17 ++- pam_deploy_graph/models.py | 14 ++- pam_deploy_graph/output_parser.py | 13 +- pam_deploy_graph/params_loader.py | 4 +- pam_deploy_graph/script_runner.py | 11 +- pam_deploy_graph/skill_policy.py | 4 +- tests/test_graph.py | 2 +- tests/test_llm_structured.py | 2 +- 31 files changed, 587 insertions(+), 98 deletions(-) create mode 100644 docs/current_logic_flow.md create mode 100644 packaging/README_linux_package.md create mode 100644 packaging/build_linux_self_contained.sh create mode 100644 packaging/pyinstaller_entry.py diff --git a/README.md b/README.md index 2d06bd2..411ce56 100644 --- a/README.md +++ b/README.md @@ -39,6 +39,13 @@ tests/ test_script_runner.py test_skill_policy.py test_interactive_cli.py + +docs/ + current_logic_flow.md # 当前整体逻辑结构流程图 + +packaging/ + build_linux_self_contained.sh # Linux 解压即用包构建脚本 + README_linux_package.md # Linux 打包说明和包大小评估 ``` ## 当前进度 @@ -138,6 +145,36 @@ agent = PamDeployAgent(mcp_runner=runner) ## 使用方式 +整体逻辑结构流程图： + +```text +docs/current_logic_flow.md +``` + +Linux 解压即用打包： + +```bash +bash packaging/build_linux_self_contained.sh +``` + +构建产物会输出到： + +```text +dist/linux_self_contained/pam-deploy-agent-linux-x86_64/ +dist/linux_self_contained/pam-deploy-agent-linux-x86_64.tar.gz +``` + +目标机器解压后运行： + +```bash +tar -xzf pam-deploy-agent-linux-x86_64.tar.gz +cd pam-deploy-agent-linux-x86_64 +./run.sh --help +./run.sh chat --config doc_scripts/config.txt.example --strategy fake --checkpoint runtime/checkpoints/demo.json +``` + +包大小以构建脚本末尾打印的 `du` 结果为准。按当前依赖估算：默认包含 MCP 依赖时压缩包约 60-110 MB、解压后约 160-300 MB；使用 `PACKAGE_EXTRAS=` 构建最小包时压缩包约 45-75 MB、解压后约 120-200 MB。 + 交互式对话框： ```bash diff --git a/docs/current_logic_flow.md b/docs/current_logic_flow.md new file mode 100644 index 0000000..b3aae77 --- /dev/null +++ b/docs/current_logic_flow.md @@ -0,0 +1,137 @@ +# 当前整体逻辑结构流程图 + +本文描述当前 PAM 部署 Agent 的主要模块、运行路径、人工确认点和断点续跑逻辑。 + +## 模块结构 + +```mermaid +flowchart TD + U[用户/上层系统] --> CLI[cli.py 命令行入口] + U --> CHAT[interactive.py 交互式 chat] + + CLI --> PARAMS[params_loader.py 读取参数] + CHAT --> PARAMS + CLI --> LLMF[llm.factory 构造 LLM client] + CHAT --> LLMF + + LLMF --> RULE[RuleBasedLlmClient 规则 fallback] + LLMF --> REAL[OpenAICompatibleLlmClient 真实 LLM] + REAL --> PROMPTS[llm.prompts 结构化提示词] + + CLI --> AGENT[PamDeployAgent] + CHAT --> AGENT + PARAMS --> AGENT + RULE --> AGENT + REAL --> AGENT + + AGENT --> ROUTER[ActionRouter] + ROUTER --> SCRIPT[ScriptActionRunner] + ROUTER --> MCP[McpActionRunner] + ROUTER --> FAKE[FakeActionRunner] + + SCRIPT --> DEPLOY[doc_scripts/deploy.sh 或 deploy.ps1] + MCP --> MCPCLIENT[mcp_client.py: Session/Function adapter] + FAKE --> FIXTURE[测试 fixture 或默认 fake 返回值] + + AGENT --> CHECKPOINT[checkpoint_store.py] + AGENT --> REPORT[render_report 部署报告] +``` + +## analyze/chat 理解和计划链路 + +```mermaid +flowchart TD + A[用户输入自然语言] --> B[understand_request 识别意图] + B --> C[validate_intent_result 校验意图] + C --> D[extract_params 抽取参数和控制信息] + D --> E[选择执行策略] + E --> F[generate_plan 生成部署计划] + F --> G[validate_deploy_plan 校验 action 和危险文本] + G --> H[输出结构化理解/计划] + H --> I{用户是否执行} + I -- 否 --> X[仅预演或继续对话] + I -- 是 --> J[run / yes 后进入部署执行] +``` + +## 部署执行主流程 + +```mermaid +flowchart TD + A[create_state 创建运行状态] --> B[normalize_params 合并默认参数并校验必填项] + B --> C[write_config 写脚本配置文件] + C --> D[build_action_backends 生成 action 路由表] + D --> E[run_deploy_flow] + + E --> F{是否存在 pending_confirmation} + F -- 是 --> P[暂停并保存 checkpoint] + F -- 否 --> G[run_global_flow 全局阶段] + + G --> G1[get-token] + G1 --> G2[create-version] + G2 --> G3[upload-package] + G3 --> G4[publish-version] + G4 --> G5[get-node-url] + G5 --> G6[get-online-ips] + G6 --> G7[create-download-task] + G7 --> G8[poll-download-progress] + G8 --> H[run_ip_flow 逐 IP 阶段] + + H --> I[resolve_target_ips 计算目标 IP] + I --> J[upgrade-ip] + J --> K[poll-upgrade-progress] + K --> L[start-ip] + L --> M[verify-ip] + M --> N[download-log] + N --> O{还有下一个 IP} + O -- 是 --> J + O -- 否 --> R[render_report 输出报告] +``` + +## action 路由规则 + +```mermaid +flowchart LR + A[action] --> B{execution_strategy} + B -- fake --> F[fake runner 执行所有 action] + B -- script_only --> S[脚本执行所有 action] + B -- hybrid_node_mcp --> C{action 类型} + C -- PAM_HOME action --> HS[脚本执行] + C -- PAM_NODE action --> NM[MCP tool 执行] +``` + +## 失败、人工确认和续跑 + +```mermaid +flowchart TD + A[逐 IP action 执行] --> B{action 失败或业务校验失败} + B -- 否 --> C[记录 completed_steps 并保存 checkpoint] + B -- 是 --> D[记录 ip_state 为 FAILED] + D --> E[download-log 尽力下载日志] + E --> F[设置 pending_confirmation=rollback-ip:IP] + F --> G[保存 checkpoint 并暂停] + + G --> H{用户决定} + H -- approve --> I[confirm_pending 执行 rollback-ip] + I --> J{rollback 是否成功} + J -- 是 --> K[清空 pending_confirmation] + J -- 否 --> L[保持 pending_confirmation，等待再次处理] + H -- reject --> M[标记 REJECTED_BY_OPERATOR 并清空 pending_confirmation] + + K --> N[resume 续跑] + M --> N + N --> O[跳过已完成全局步骤、成功 IP 和单 IP 已完成 action] +``` + +## checkpoint 续跑语义 + +- `completed_global_steps`：全局阶段已经完成的 action 会跳过。 +- `ip_states[ip].status == SUCCESS`：成功 IP 会跳过。 +- `ip_states[ip].completed_steps`：同一个 IP 已完成的 action 会跳过。 +- `pending_confirmation`：存在待确认事项时，部署流程不继续执行，必须先 `approve` 或 `reject`。 +- checkpoint 为了真实续跑会保存完整参数，请放在受控目录中。 + +## 真实外部能力接入点 + +- 真实 LLM：`llm.openai_compatible.OpenAICompatibleLlmClient`，通过 `PAM_LLM_BASE_URL`、`PAM_LLM_API_KEY`、`PAM_LLM_MODEL` 或 CLI 参数配置。 +- 真实 MCP：外部建立 MCP session 后，用 `SessionMcpToolClient` 包装，再传给 `McpActionRunner`。 +- 真实脚本：PAM_HOME action 通过 `doc_scripts/deploy.sh` 或 `deploy.ps1` 调用。 diff --git a/packaging/README_linux_package.md b/packaging/README_linux_package.md new file mode 100644 index 0000000..86a009b --- /dev/null +++ b/packaging/README_linux_package.md @@ -0,0 +1,55 @@ +# Linux 解压即用打包说明 + +## 目标 + +`build_linux_self_contained.sh` 会在 Linux x86_64 构建机上生成一个自带 Python 运行时和 Python 依赖的发布包。目标机器解压后可以直接运行，不需要额外安装 Python 或 pip 依赖。 + +## 构建机要求 + +- Linux x86_64 +- `bash` +- `python3`，需要带 `venv` 模块 +- 构建时需要访问 Python 包索引，用于安装 PyInstaller 和项目依赖 + +该脚本不支持在 Windows 上交叉构建 Linux 包。建议在和目标环境 glibc 版本相同或更旧的 Linux 发行版上构建，以提高二进制兼容性。 + +## 构建命令 + +```bash +bash packaging/build_linux_self_contained.sh +``` + +默认会安装 `.[mcp]`，即包含 MCP 可选依赖。如果只想打最小包： + +```bash +PACKAGE_EXTRAS= bash packaging/build_linux_self_contained.sh +``` + +构建产物： + +```text +dist/linux_self_contained/pam-deploy-agent-linux-x86_64/ +dist/linux_self_contained/pam-deploy-agent-linux-x86_64.tar.gz +``` + +## 解压后运行 + +```bash +tar -xzf pam-deploy-agent-linux-x86_64.tar.gz +cd pam-deploy-agent-linux-x86_64 +./run.sh --help +./run.sh chat --config doc_scripts/config.txt.example --strategy fake --checkpoint runtime/checkpoints/demo.json +``` + +`run.sh` 会切换到发布目录再启动可执行程序，因此默认的 `doc_scripts/...` 相对路径可以正常工作。 + +## 包大小评估 + +最终大小以脚本末尾打印的 `du` 结果为准。按当前依赖结构预估： + +| 构建方式 | 压缩包大小 | 解压后大小 | 主要影响因素 | +| --- | ---: | ---: | --- | +| 默认包含 MCP 依赖 | 60-110 MB | 160-300 MB | PyInstaller 运行时、langgraph、mcp 及其依赖 | +| `PACKAGE_EXTRAS=` 最小包 | 45-75 MB | 120-200 MB | PyInstaller 运行时、langgraph 及基础依赖 | + +当前项目源码和脚本文档本身体积很小，包大小主要由 Python 运行时、PyInstaller bootloader、第三方依赖和动态库决定。 diff --git a/packaging/build_linux_self_contained.sh b/packaging/build_linux_self_contained.sh new file mode 100644 index 0000000..07141fd --- /dev/null +++ b/packaging/build_linux_self_contained.sh @@ -0,0 +1,117 @@ +#!/usr/bin/env bash +# 构建 Linux 解压即用包：包含 Python 运行时、依赖、CLI 可执行程序和脚本文档。 + +set -euo pipefail + +ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +cd "$ROOT_DIR" + +PYTHON_BIN="${PYTHON_BIN:-python3}" +APP_NAME="pam-deploy-agent" +RELEASE_NAME="${APP_NAME}-linux-x86_64" +PACKAGE_EXTRAS="${PACKAGE_EXTRAS:-mcp}" +BUILD_DIR="${BUILD_DIR:-$ROOT_DIR/build/linux_self_contained}" +DIST_DIR="${DIST_DIR:-$ROOT_DIR/dist/linux_self_contained}" +RELEASE_DIR="$DIST_DIR/$RELEASE_NAME" +ARCHIVE_PATH="$DIST_DIR/${RELEASE_NAME}.tar.gz" + +if [[ "$(uname -s)" != "Linux" ]]; then + echo "该脚本需要在 Linux x86_64 构建机上运行。" + exit 1 +fi + +if ! command -v "$PYTHON_BIN" >/dev/null 2>&1; then + echo "未找到 Python: $PYTHON_BIN" + exit 1 +fi + +echo "==> 清理旧构建目录" +rm -rf "$BUILD_DIR" "$RELEASE_DIR" "$ARCHIVE_PATH" +mkdir -p "$BUILD_DIR" "$DIST_DIR" + +echo "==> 创建构建虚拟环境" +"$PYTHON_BIN" -m venv "$BUILD_DIR/venv" +source "$BUILD_DIR/venv/bin/activate" + +echo "==> 安装构建依赖" +python -m pip install --upgrade pip setuptools wheel +python -m pip install pyinstaller + +echo "==> 安装项目依赖" +if [[ -n "$PACKAGE_EXTRAS" ]]; then + python -m pip install -e ".[${PACKAGE_EXTRAS}]" +else + python -m pip install -e . +fi + +echo "==> 使用 PyInstaller 生成自带 Python 运行时的可执行目录" +python -m PyInstaller \ + --clean \ + --noconfirm \ + --name "$APP_NAME" \ + --onedir \ + --console \ + --distpath "$BUILD_DIR/pyinstaller_dist" \ + --workpath "$BUILD_DIR/pyinstaller_build" \ + --specpath "$BUILD_DIR" \ + --collect-submodules pam_deploy_graph \ + --collect-submodules langgraph \ + --hidden-import pam_deploy_graph.cli \ + packaging/pyinstaller_entry.py + +echo "==> 组装发布目录" +mkdir -p "$RELEASE_DIR" +cp -a "$BUILD_DIR/pyinstaller_dist/$APP_NAME/." "$RELEASE_DIR/" +cp -a doc_scripts "$RELEASE_DIR/doc_scripts" +cp -a README.md "$RELEASE_DIR/README.md" +cp -a LICENSE "$RELEASE_DIR/LICENSE" + +cat > "$RELEASE_DIR/run.sh" <<'RUN_SCRIPT' +#!/usr/bin/env bash +set -euo pipefail +DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +cd "$DIR" +exec "$DIR/pam-deploy-agent" "$@" +RUN_SCRIPT +chmod +x "$RELEASE_DIR/run.sh" + +cat > "$RELEASE_DIR/使用说明.txt" <<'USAGE_TEXT' +PAM 部署 Agent Linux 解压即用包 + +使用方式： + ./run.sh --help + ./run.sh chat --config doc_scripts/config.txt.example --strategy fake --checkpoint runtime/checkpoints/demo.json + ./run.sh analyze --config doc_scripts/config.txt.example --text "请用 MCP 预演部署 HET PAM Node 版本 2.0.5，不要动环境" + +说明： + 1. 该包已包含 Python 运行时和 Python 依赖，目标机器不需要额外安装 Python 包。 + 2. 真实 LLM 仍需通过 PAM_LLM_BASE_URL、PAM_LLM_API_KEY、PAM_LLM_MODEL 或 CLI 参数配置。 + 3. 真实 MCP session 仍需由你在外部接入后传给 Agent；当前包已包含 MCP client adapter。 + 4. checkpoint 会保存完整运行参数，请放在受控目录。 +USAGE_TEXT + +echo "==> 生成 tar.gz" +tar -C "$DIST_DIR" -czf "$ARCHIVE_PATH" "$RELEASE_NAME" + +format_bytes() { + local bytes="$1" + python - "$bytes" <<'PY' +import sys +value = float(sys.argv[1]) +units = ["B", "KB", "MB", "GB"] +for unit in units: + if value < 1024 or unit == units[-1]: + print(f"{value:.1f} {unit}") + break + value /= 1024 +PY +} + +EXTRACTED_BYTES="$(du -sb "$RELEASE_DIR" | awk '{print $1}')" +ARCHIVE_BYTES="$(du -sb "$ARCHIVE_PATH" | awk '{print $1}')" + +echo "==> 构建完成" +echo "发布目录: $RELEASE_DIR" +echo "压缩包: $ARCHIVE_PATH" +echo "解压后大小: $(format_bytes "$EXTRACTED_BYTES")" +echo "压缩包大小: $(format_bytes "$ARCHIVE_BYTES")" diff --git a/packaging/pyinstaller_entry.py b/packaging/pyinstaller_entry.py new file mode 100644 index 0000000..6189c6f --- /dev/null +++ b/packaging/pyinstaller_entry.py @@ -0,0 +1,7 @@ +"""PyInstaller 打包入口。""" + +from pam_deploy_graph.cli import main + + +if __name__ == "__main__": + main() diff --git a/pam_deploy_graph/__init__.py b/pam_deploy_graph/__init__.py index 71b73f5..7e20b82 100644 --- a/pam_deploy_graph/__init__.py +++ b/pam_deploy_graph/__init__.py @@ -1,6 +1,5 @@ -"""PAM deploy agent package.""" +"""PAM 部署 Agent 包入口。""" from .agent import PamDeployAgent __all__ = ["PamDeployAgent"] - diff --git a/pam_deploy_graph/action_router.py b/pam_deploy_graph/action_router.py index c86d195..22a7f33 100644 --- a/pam_deploy_graph/action_router.py +++ b/pam_deploy_graph/action_router.py @@ -1,4 +1,4 @@ -"""Action routing for HOME script actions and NODE MCP actions.""" +"""按照执行策略把 action 路由到脚本、MCP 或 fake runner。""" from __future__ import annotations @@ -7,6 +7,7 @@ from .models import AgentState, BackendName, ExecutionStrategy, ActionResult def build_action_backends(strategy: ExecutionStrategy) -> dict[str, BackendName]: + """根据执行策略生成每个 action 对应的后端类型。""" if strategy == "fake": return {action: "fake" for action in ALLOWED_ACTIONS} if strategy == "script_only": @@ -15,19 +16,23 @@ def build_action_backends(strategy: ExecutionStrategy) -> dict[str, BackendName] routes: dict[str, BackendName] = {action: "script" for action in HOME_ACTIONS} routes.update({action: "mcp" for action in NODE_ACTIONS}) return routes - raise ValueError(f"Unknown execution strategy: {strategy}") + raise ValueError(f"未知执行策略: {strategy}") class ActionRouter: + """统一的 action 调度器，屏蔽脚本、MCP 和 fake 后端差异。""" + def __init__(self, *, script_runner, mcp_runner=None, fake_runner=None) -> None: + """保存各类 runner，运行时按 state 中的路由表选择后端。""" self.script_runner = script_runner self.mcp_runner = mcp_runner self.fake_runner = fake_runner def run_action(self, state: AgentState, action: str, **kwargs) -> ActionResult: + """执行一个 action，并返回统一的 ActionResult。""" backend = state.action_backends.get(action) if not backend: - raise ValueError(f"Action is not routed: {action}") + raise ValueError(f"action 未配置路由: {action}") if backend == "script": return self.script_runner.run( action, @@ -39,9 +44,8 @@ class ActionRouter: ) if backend == "mcp": if self.mcp_runner is None: - raise RuntimeError(f"MCP runner is required for action: {action}") + raise RuntimeError(f"action 需要 MCP runner: {action}") return self.mcp_runner.run(action, params=state.params, **kwargs) if self.fake_runner is None: - raise RuntimeError(f"Fake runner is required for action: {action}") + raise RuntimeError(f"action 需要 fake runner: {action}") return self.fake_runner.run(action, params=state.params, **kwargs) - diff --git a/pam_deploy_graph/agent.py b/pam_deploy_graph/agent.py index 7bbb758..b472bf6 100644 --- a/pam_deploy_graph/agent.py +++ b/pam_deploy_graph/agent.py @@ -1,7 +1,7 @@ -"""PAM deploy Agent runtime. +"""PAM 部署 Agent 运行时。 -This is intentionally runnable without langgraph installed. The same nodes can -be wired into LangGraph later via pam_deploy_graph.graph. +本模块不强依赖 LangGraph，可独立运行；同一组节点也可在 +`pam_deploy_graph.graph` 中接入 LangGraph。 """ from __future__ import annotations @@ -23,6 +23,8 @@ from .skill_policy import load_skill_policy class PamDeployAgent: + """PAM 部署主 Agent，串联 LLM、action 路由、确认和续跑状态。""" + def __init__( self, *, @@ -32,6 +34,7 @@ class PamDeployAgent: fake_runner: FakeActionRunner | None = None, llm_client: LlmClient | None = None, ) -> None: + """初始化策略、脚本 runner、MCP runner、fake runner 和 LLM client。""" self.skill_policy = load_skill_policy(skill_path) self.script_base_dir = Path(script_base_dir) self.script_runner = ScriptActionRunner(self.script_base_dir) @@ -45,11 +48,13 @@ class PamDeployAgent: ) def understand_request(self, text: str) -> LlmIntentResult: + """调用 LLM 识别用户意图，并执行基础校验。""" result = self.llm_client.understand_request(text) validate_intent_result(result) return result def extract_params(self, text: str, base_params: dict[str, Any] | None = None) -> LlmParamResult: + """从自然语言中抽取部署参数和控制参数。""" return self.llm_client.extract_params(text, base_params) def generate_plan( @@ -59,11 +64,13 @@ class PamDeployAgent: intent: str, strategy: ExecutionStrategy, ) -> LlmDeployPlan: + """根据参数、意图和执行策略生成部署计划。""" plan = self.llm_client.generate_plan(params=params, intent=intent, strategy=strategy) validate_deploy_plan(plan) return plan 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) @@ -79,13 +86,15 @@ class PamDeployAgent: } def normalize_params(self, params: dict[str, Any]) -> dict[str, Any]: + """合并默认参数并校验必填参数是否齐全。""" normalized = {**DEFAULT_PARAMS, **params} missing = [key for key in REQUIRED_PARAMS if not normalized.get(key)] if missing: - raise ValueError(f"Missing required params: {', '.join(missing)}") + raise ValueError(f"缺少必填参数: {', '.join(missing)}") return normalized def _choose_strategy(self, preference: str) -> ExecutionStrategy: + """把 LLM 给出的策略偏好转换为内部执行策略。""" if preference in ("hybrid_node_mcp", "script_only", "fake"): return preference # type: ignore[return-value] return "hybrid_node_mcp" @@ -102,6 +111,7 @@ class PamDeployAgent: checkpoint_path: str | None = None, target_ips: list[str] | None = None, ) -> AgentState: + """创建一次运行所需的 AgentState，并写入脚本配置文件。""" normalized = self.normalize_params(params) actual_run_id = run_id or time.strftime("%Y%m%d_%H%M%S") actual_script_entry = script_entry or select_script_entry() @@ -123,6 +133,7 @@ class PamDeployAgent: ) def preview(self, params: dict[str, Any], strategy: ExecutionStrategy = "hybrid_node_mcp") -> str: + """渲染部署预演，展示参数和 action 路由。""" normalized = self.normalize_params(params) routes = build_action_backends(strategy) if strategy == "hybrid_node_mcp": @@ -153,6 +164,7 @@ class PamDeployAgent: return "\n".join(lines) def run_global_flow(self, state: AgentState) -> AgentState: + """执行全局部署阶段，并跳过 checkpoint 中已完成的步骤。""" for action in GLOBAL_ACTION_SEQUENCE: if action in state.completed_global_steps: continue @@ -171,7 +183,7 @@ class PamDeployAgent: if not result.ok: state.last_failed_step = action self._save_checkpoint(state) - raise RuntimeError(f"{action} failed: {result.error_summary}") + raise RuntimeError(f"{action} 执行失败: {result.error_summary}") self._apply_result(state, action, result.values) state.completed_global_steps.append(action) state.last_success_step = action @@ -179,6 +191,7 @@ class PamDeployAgent: return state def run_deploy_flow(self, state: AgentState) -> AgentState: + """执行完整部署流程：全局阶段后进入逐 IP 阶段。""" if state.pending_confirmation: self._save_checkpoint(state) return state @@ -187,6 +200,7 @@ class PamDeployAgent: return state def run_ip_flow(self, state: AgentState) -> AgentState: + """执行逐 IP 部署流程，失败时停在人工确认点。""" if state.pending_confirmation: self._save_checkpoint(state) return state @@ -248,6 +262,7 @@ class PamDeployAgent: return state def build_confirmation_request(self, state: AgentState) -> dict[str, Any]: + """把 pending_confirmation 转换为面向用户的确认请求。""" if not state.pending_confirmation: return {} kind, _, value = state.pending_confirmation.partition(":") @@ -268,11 +283,12 @@ class PamDeployAgent: } def confirm_pending(self, state: AgentState, *, approved: bool, operator_note: str = "") -> AgentState: + """处理人工确认结果；当前支持失败 IP 的回滚确认。""" request = self.build_confirmation_request(state) if not request: - raise ValueError("No pending confirmation") + raise ValueError("当前没有待确认事项") if request["type"] != "rollback-ip": - raise ValueError(f"Unsupported confirmation type: {request['type']}") + raise ValueError(f"不支持的确认类型: {request['type']}") ip = request["ip"] ip_state = state.ip_states[ip] @@ -317,6 +333,7 @@ class PamDeployAgent: return state def _apply_result(self, state: AgentState, action: str, values: dict[str, Any]) -> None: + """把全局 action 返回值写回 AgentState。""" if "HASH_CODE" in values: state.hash_code = str(values["HASH_CODE"]) if "NODE_URL" in values: @@ -329,6 +346,7 @@ class PamDeployAgent: state.target_ips = state.target_ips or state.online_ips.copy() def _resolve_target_ips(self, state: AgentState) -> None: + """根据在线 IP 和用户指定 IP 计算最终目标 IP。""" if not state.target_ips: state.target_ips = state.online_ips.copy() return @@ -347,6 +365,7 @@ class PamDeployAgent: ) def _business_failed(self, action: str, values: dict[str, Any]) -> bool: + """识别 exit code 之外的业务失败条件。""" if action == "verify-ip": success = values.get("SUCCESS") if success is None: @@ -355,10 +374,12 @@ class PamDeployAgent: return False def _apply_ip_result(self, ip_state: dict[str, Any], action: str, values: dict[str, Any]) -> None: + """把逐 IP action 返回值写回单 IP 状态。""" if action == "download-log": ip_state["log_file"] = str(values.get("LOG_FILE", "")) def _record_ip_failure(self, state: AgentState, ip: str, action: str, reason: str) -> None: + """记录单 IP 失败，并设置待回滚确认状态。""" ip_state = state.ip_states[ip] stop_first = action in ("start-ip", "verify-ip") ip_state.update( @@ -373,15 +394,16 @@ class PamDeployAgent: state.last_failed_step = action state.events.append( { - "type": "CONFIRMATION_REQUIRED", - "stage": "rollback-ip", - "ip": ip, - "stop_first": stop_first, - "message": f"{action} failed; rollback confirmation required", - } - ) + "type": "CONFIRMATION_REQUIRED", + "stage": "rollback-ip", + "ip": ip, + "stop_first": stop_first, + "message": f"{action} 执行失败，需要确认是否回滚", + } + ) def _download_log_best_effort(self, state: AgentState, ip: str) -> None: + """失败后尽力下载日志，日志失败不覆盖原失败原因。""" result = self.router.run_action(state, "download-log", ip=ip) ip_state = state.ip_states[ip] if result.ok: @@ -392,7 +414,7 @@ class PamDeployAgent: "stage": "download-log", "backend": result.backend, "ip": ip, - "message": "best effort log downloaded", + "message": "已尽力下载日志", } ) else: @@ -402,15 +424,17 @@ class PamDeployAgent: "stage": "download-log", "backend": result.backend, "ip": ip, - "message": result.error_summary or "best effort log download failed", + "message": result.error_summary or "尽力下载日志失败", } ) def _save_checkpoint(self, state: AgentState) -> None: + """如果配置了 checkpoint 路径，则保存完整运行状态。""" if state.checkpoint_path: save_checkpoint(state, state.checkpoint_path, redact=False) def render_report(self, state: AgentState) -> str: + """渲染当前部署状态报告。""" success = sum(1 for item in state.ip_states.values() if item.get("status") == "SUCCESS") failed = sum(1 for item in state.ip_states.values() if item.get("status") == "FAILED") lines = [ diff --git a/pam_deploy_graph/checkpoint_store.py b/pam_deploy_graph/checkpoint_store.py index a010c79..8f4dca7 100644 --- a/pam_deploy_graph/checkpoint_store.py +++ b/pam_deploy_graph/checkpoint_store.py @@ -1,4 +1,4 @@ -"""Business checkpoint JSON storage.""" +"""业务 checkpoint 的 JSON 存储与恢复工具。""" from __future__ import annotations @@ -12,6 +12,7 @@ from .models import AgentState def redact_mapping(value: Any) -> Any: + """递归脱敏 dict/list 中的敏感字段。""" if isinstance(value, dict): result = {} for key, item in value.items(): @@ -26,6 +27,7 @@ def redact_mapping(value: Any) -> Any: def save_checkpoint(state: Any, path: str | Path, *, redact: bool = True) -> Path: + """保存 checkpoint；真实续跑场景可关闭脱敏以保留必要参数。""" checkpoint_path = Path(path) checkpoint_path.parent.mkdir(parents=True, exist_ok=True) payload = asdict(state) if is_dataclass(state) else state @@ -39,14 +41,17 @@ def save_checkpoint(state: Any, path: str | Path, *, redact: bool = True) -> Pat def load_checkpoint(path: str | Path) -> dict[str, Any]: + """从 JSON 文件读取原始 checkpoint 字典。""" return json.loads(Path(path).read_text(encoding="utf-8")) def agent_state_from_mapping(payload: dict[str, Any]) -> AgentState: + """把 checkpoint 字典转换回 AgentState，忽略未知字段。""" allowed_fields = {item.name for item in fields(AgentState)} state_payload = {key: value for key, value in payload.items() if key in allowed_fields} return AgentState(**state_payload) def load_agent_state(path: str | Path) -> AgentState: + """读取 checkpoint 文件并恢复 AgentState。""" return agent_state_from_mapping(load_checkpoint(path)) diff --git a/pam_deploy_graph/cli.py b/pam_deploy_graph/cli.py index 9ea22a2..6000650 100644 --- a/pam_deploy_graph/cli.py +++ b/pam_deploy_graph/cli.py @@ -1,4 +1,4 @@ -"""Command line interface for the PAM deploy agent.""" +"""PAM 部署 Agent 的命令行入口。""" from __future__ import annotations @@ -14,17 +14,20 @@ from .params_loader import load_params_file def add_llm_args(parser: argparse.ArgumentParser) -> None: + """为子命令追加真实 LLM 配置参数。""" parser.add_argument("--llm-base-url") parser.add_argument("--llm-api-key") parser.add_argument("--llm-model") def require_confirm(args: argparse.Namespace) -> None: + """真实执行前强制要求命令行显式传入 --confirm。""" if not getattr(args, "confirm", False): raise SystemExit("Refusing to execute actions without --confirm.") def print_pause_payload(agent: PamDeployAgent, state) -> None: + """输出 checkpoint 和待确认信息，便于用户续跑或确认。""" if state.pending_confirmation: print(json.dumps({"confirmation": agent.build_confirmation_request(state)}, ensure_ascii=False, indent=2)) if state.checkpoint_path: @@ -32,6 +35,7 @@ def print_pause_payload(agent: PamDeployAgent, state) -> None: def main() -> None: + """解析 CLI 参数并分发到对应命令。""" parser = argparse.ArgumentParser(prog="pam-deploy-agent") sub = parser.add_subparsers(dest="command", required=True) diff --git a/pam_deploy_graph/config_writer.py b/pam_deploy_graph/config_writer.py index 38de0fe..53c8e4d 100644 --- a/pam_deploy_graph/config_writer.py +++ b/pam_deploy_graph/config_writer.py @@ -1,4 +1,4 @@ -"""Write script config files for PAM HOME action calls.""" +"""为 PAM_HOME 脚本 action 写入 config.txt 风格配置文件。""" from __future__ import annotations @@ -21,9 +21,9 @@ CONFIG_KEYS = ( def write_config(params: dict[str, Any], path: str | Path) -> Path: + """按脚本约定的字段顺序生成配置文件，并返回最终路径。""" config_path = Path(path) config_path.parent.mkdir(parents=True, exist_ok=True) lines = [f"{key}={params.get(key, '')}" for key in CONFIG_KEYS] config_path.write_text("\n".join(lines) + "\n", encoding="utf-8") return config_path - diff --git a/pam_deploy_graph/constants.py b/pam_deploy_graph/constants.py index 4d2b0ef..64485e8 100644 --- a/pam_deploy_graph/constants.py +++ b/pam_deploy_graph/constants.py @@ -1,5 +1,6 @@ -"""Constants for PAM deploy action routing.""" +"""PAM 部署流程中的 action、参数和敏感字段常量。""" +# PAM_HOME 侧只能通过脚本执行的 action。 HOME_ACTIONS = ( "get-token", "create-version", @@ -8,6 +9,7 @@ HOME_ACTIONS = ( "get-node-url", ) +# PAM_NODE 侧可通过 MCP 或脚本执行的 action。 NODE_ACTIONS = ( "get-online-ips", "create-download-task", @@ -21,6 +23,7 @@ NODE_ACTIONS = ( "rollback-ip", ) +# 全局阶段按顺序执行，完成后才能进入逐 IP 阶段。 GLOBAL_ACTION_SEQUENCE = ( "get-token", "create-version", @@ -32,6 +35,7 @@ GLOBAL_ACTION_SEQUENCE = ( "poll-download-progress", ) +# 单个工作站 IP 的部署阶段顺序。 IP_ACTION_SEQUENCE = ( "upgrade-ip", "poll-upgrade-progress", @@ -40,8 +44,10 @@ IP_ACTION_SEQUENCE = ( "download-log", ) +# Agent 允许规划和执行的完整 action 集合。 ALLOWED_ACTIONS = HOME_ACTIONS + NODE_ACTIONS +# 创建运行状态前必须具备的部署参数。 REQUIRED_PARAMS = ( "HOME_BASE_URL", "CLIENT_ID", @@ -53,12 +59,14 @@ REQUIRED_PARAMS = ( "ZIP_FILE_PATH", ) +# 用户未显式提供时使用的默认参数。 DEFAULT_PARAMS = { "ACTION_TYPE": "FULL", "TIMEOUT": 120, "LOG_NAME": "app.log", } +# 日志、报告和 LLM 输入中需要脱敏的字段。 SENSITIVE_KEYS = { "CLIENT_SECRET", "TOKEN", @@ -66,4 +74,3 @@ SENSITIVE_KEYS = { "access_token", "ACCESS_TOKEN", } - diff --git a/pam_deploy_graph/fake_runner.py b/pam_deploy_graph/fake_runner.py index 90bd789..d5b72f0 100644 --- a/pam_deploy_graph/fake_runner.py +++ b/pam_deploy_graph/fake_runner.py @@ -1,4 +1,4 @@ -"""Fake action runner for graph and agent tests.""" +"""供本地测试和预演使用的 fake action runner。""" from __future__ import annotations @@ -8,11 +8,15 @@ from .models import ActionResult class FakeActionRunner: + """返回确定性 action 结果，避免测试触碰真实 PAM 环境。""" + def __init__(self, fixtures: dict[str, dict[str, Any]] | None = None) -> None: + """保存可覆盖默认行为的测试 fixture，并记录调用历史。""" self.fixtures = fixtures or {} self.calls: list[tuple[str, dict[str, Any]]] = [] def run(self, action: str, *, params: dict[str, Any], **kwargs: Any) -> ActionResult: + """执行 fake action，优先使用 fixture，否则使用内置默认结果。""" self.calls.append((action, kwargs)) values = self._fixture_for(action, kwargs) if not values: @@ -26,10 +30,11 @@ class FakeActionRunner: values=values, exit_code=0 if ok else 1, raw_output=str(values), - error_summary="" if ok else str(values.get("MESSAGE", "Fake action failed")), + error_summary="" if ok else str(values.get("MESSAGE", "fake action 执行失败")), ) def _default_values(self, action: str, kwargs: dict[str, Any]) -> dict[str, Any]: + """为常见部署 action 构造稳定的默认返回值。""" if action == "get-token": return {"ACTION": action, "TOKEN": "***"} if action == "upload-package": @@ -57,6 +62,7 @@ class FakeActionRunner: return {"ACTION": action, "RESULT": "OK"} def _fixture_for(self, action: str, kwargs: dict[str, Any]) -> dict[str, Any]: + """按 action 或 action:ip 查找测试 fixture。""" ip = kwargs.get("ip") ip_key = f"{action}:{ip}" if ip else "" if ip_key and ip_key in self.fixtures: diff --git a/pam_deploy_graph/graph.py b/pam_deploy_graph/graph.py index 8642c15..18ed3f9 100644 --- a/pam_deploy_graph/graph.py +++ b/pam_deploy_graph/graph.py @@ -1,4 +1,4 @@ -"""LangGraph integration for the PAM deploy Agent.""" +"""PAM 部署 Agent 的 LangGraph 集成入口。""" from __future__ import annotations @@ -10,17 +10,18 @@ GraphFlow = Literal["global", "deploy"] def build_langgraph(agent: PamDeployAgent | None = None, flow: GraphFlow = "deploy"): + """把现有 Agent 节点组装成 LangGraph StateGraph。""" try: from langgraph.graph import END, START, StateGraph - except ImportError as exc: # pragma: no cover - depends on optional package + except ImportError as exc: # pragma: no cover - 依赖可选安装状态 raise RuntimeError( - "langgraph is not installed. Install project dependencies with " - "`pip install -e .`." + "未安装 langgraph。请先执行 `pip install -e .` 安装项目依赖。" ) from exc runtime = agent or PamDeployAgent() def create_state_node(state: dict[str, Any]) -> dict[str, Any]: + """根据输入参数创建 AgentState。""" agent_state = runtime.create_state( params=state["params"], execution_strategy=state.get("execution_strategy", "hybrid_node_mcp"), @@ -33,14 +34,17 @@ def build_langgraph(agent: PamDeployAgent | None = None, flow: GraphFlow = "depl return {"agent_state": agent_state} def run_global_node(state: dict[str, Any]) -> dict[str, Any]: + """运行全局部署阶段。""" agent_state = runtime.run_global_flow(state["agent_state"]) return {"agent_state": agent_state} def run_ip_node(state: dict[str, Any]) -> dict[str, Any]: + """运行逐 IP 部署阶段。""" agent_state = runtime.run_ip_flow(state["agent_state"]) return {"agent_state": agent_state} def report_node(state: dict[str, Any]) -> dict[str, Any]: + """渲染最终部署报告。""" return {"report": runtime.render_report(state["agent_state"])} graph = StateGraph(dict) @@ -61,6 +65,7 @@ def build_langgraph(agent: PamDeployAgent | None = None, flow: GraphFlow = "depl def build_graph_or_none(agent: PamDeployAgent | None = None, flow: GraphFlow = "deploy"): + """在未安装 LangGraph 时返回 None，便于调用方降级。""" try: return build_langgraph(agent=agent, flow=flow) except RuntimeError: diff --git a/pam_deploy_graph/interactive.py b/pam_deploy_graph/interactive.py index 847e3f5..e492183 100644 --- a/pam_deploy_graph/interactive.py +++ b/pam_deploy_graph/interactive.py @@ -1,4 +1,4 @@ -"""Interactive CLI session for the PAM deploy agent.""" +"""PAM 部署 Agent 的常驻式交互 CLI 会话。""" from __future__ import annotations @@ -32,6 +32,8 @@ COMMAND_HELP = """可用命令： class InteractiveCliSession: + """维护一次交互式 CLI 会话的参数、状态和命令处理逻辑。""" + def __init__( self, *, @@ -43,6 +45,7 @@ class InteractiveCliSession: input_func: InputFunc = input, output_func: OutputFunc = print, ) -> None: + """初始化会话上下文和输入输出函数。""" self.agent = agent self.params = dict(params) self.strategy = strategy @@ -54,7 +57,8 @@ class InteractiveCliSession: self.last_analysis: dict[str, Any] | None = None def run(self) -> None: - self.output("PAM Deploy Agent interactive session") + """启动 REPL 循环，直到用户 exit 或输入流结束。""" + self.output("PAM 部署 Agent 交互式会话") self.output("输入 help 查看命令，输入 exit 退出。") self._load_existing_checkpoint_if_any() while True: @@ -67,6 +71,7 @@ class InteractiveCliSession: return def handle_line(self, line: str) -> bool: + """处理用户输入的一行命令；返回 False 表示退出会话。""" text = line.strip() if not text: return True @@ -112,6 +117,7 @@ class InteractiveCliSession: return True def _analyze(self, text: str) -> None: + """分析自然语言需求，并更新会话中的参数、策略和目标 IP。""" if not text: self.output("请输入要分析的自然语言需求，例如：analyze 请用 MCP 预演部署 HET。") return @@ -141,6 +147,7 @@ class InteractiveCliSession: self.output(_format_redacted_params(safe_payload["params"]["extracted_params"])) def _set_param(self, assignment: str) -> None: + """处理 `set KEY=VALUE` 命令，更新当前会话参数。""" if "=" not in assignment: self.output("格式：set KEY=VALUE") return @@ -153,6 +160,7 @@ class InteractiveCliSession: self.output(f"已设置 {key}") def _run_deploy(self) -> None: + """在用户确认后创建状态并执行完整部署流程。""" if self.state and self.state.pending_confirmation: self._print_confirmation() return @@ -170,6 +178,7 @@ class InteractiveCliSession: self._execute_current_state() def _resume(self) -> None: + """从内存状态或 checkpoint 文件继续执行部署流程。""" if self.state is None: checkpoint = Path(self.checkpoint_path) if not checkpoint.exists(): @@ -180,6 +189,7 @@ class InteractiveCliSession: self._execute_current_state() def _execute_current_state(self) -> None: + """执行当前 state，并输出报告、确认提示和 checkpoint 路径。""" if self.state is None: self.output("当前没有运行状态。") return @@ -190,6 +200,7 @@ class InteractiveCliSession: self.output(f"checkpoint: {self.state.checkpoint_path or self.checkpoint_path}") def _status(self) -> None: + """输出当前运行状态；没有 state 时输出 checkpoint 路径。""" if self.state is None: self.output("当前还没有运行状态。") self.output(f"checkpoint: {self.checkpoint_path}") @@ -199,6 +210,7 @@ class InteractiveCliSession: self._print_confirmation() def _confirm(self, *, approved: bool, note: str = "") -> None: + """处理 approve/reject 命令。""" if self.state is None: checkpoint = Path(self.checkpoint_path) if checkpoint.exists(): @@ -217,6 +229,7 @@ class InteractiveCliSession: self._print_confirmation() def _print_confirmation(self) -> None: + """输出当前待人工确认事项。""" if self.state is None: return request = self.agent.build_confirmation_request(self.state) @@ -233,6 +246,7 @@ class InteractiveCliSession: self.output("输入 approve 执行回滚，或 reject [原因] 拒绝回滚。") def _ask_yes_no(self, prompt: str) -> bool: + """读取一次 yes/no 确认，只有 yes/y 视为确认。""" try: answer = self.input(prompt).strip().lower() except EOFError: @@ -240,6 +254,7 @@ class InteractiveCliSession: return answer in ("yes", "y") def _load_existing_checkpoint_if_any(self) -> None: + """会话启动时自动加载已存在的 checkpoint。""" checkpoint = Path(self.checkpoint_path) if not checkpoint.exists(): return @@ -260,6 +275,7 @@ def run_interactive_chat( input_func: InputFunc = input, output_func: OutputFunc = print, ) -> InteractiveCliSession: + """创建并运行交互式 CLI 会话，返回会话对象便于测试。""" session = InteractiveCliSession( agent=agent, params=params, @@ -274,16 +290,19 @@ def run_interactive_chat( def _default_checkpoint_path() -> str: + """生成默认 chat checkpoint 路径。""" return str(Path("runtime") / "checkpoints" / f"chat_{time.strftime('%Y%m%d_%H%M%S')}.json") def _choose_strategy(preference: str, default: ExecutionStrategy) -> ExecutionStrategy: + """根据 LLM 偏好更新执行策略，非法值保留默认策略。""" if preference in ("hybrid_node_mcp", "script_only", "fake"): return preference # type: ignore[return-value] return default def _format_redacted_params(params: dict[str, Any]) -> str: + """把脱敏后的参数字典格式化为多行文本。""" lines = ["当前参数："] for key in sorted(params): lines.append(f"- {key}: {params[key]}") diff --git a/pam_deploy_graph/llm/__init__.py b/pam_deploy_graph/llm/__init__.py index 7e70597..a704e26 100644 --- a/pam_deploy_graph/llm/__init__.py +++ b/pam_deploy_graph/llm/__init__.py @@ -1,4 +1,4 @@ -"""LLM integration surfaces for PAM deploy Agent.""" +"""PAM 部署 Agent 的 LLM 集成导出入口。""" from .base import LlmClient from .factory import build_llm_client diff --git a/pam_deploy_graph/llm/base.py b/pam_deploy_graph/llm/base.py index 2716115..ee5c2ca 100644 --- a/pam_deploy_graph/llm/base.py +++ b/pam_deploy_graph/llm/base.py @@ -1,4 +1,4 @@ -"""Shared LLM client protocol.""" +"""LLM client 需要实现的共享协议。""" from __future__ import annotations @@ -8,10 +8,14 @@ from pam_deploy_graph.models import ExecutionStrategy, LlmDeployPlan, LlmIntentR class LlmClient(Protocol): + """Agent 使用的最小 LLM 能力接口。""" + def understand_request(self, text: str) -> LlmIntentResult: + """识别用户自然语言请求的意图。""" ... def extract_params(self, text: str, base_params: dict[str, Any] | None = None) -> LlmParamResult: + """从自然语言中抽取部署参数。""" ... def generate_plan( @@ -21,4 +25,5 @@ class LlmClient(Protocol): intent: str, strategy: ExecutionStrategy, ) -> LlmDeployPlan: + """根据参数和意图生成部署计划。""" ... diff --git a/pam_deploy_graph/llm/factory.py b/pam_deploy_graph/llm/factory.py index 7654943..a4e2c6b 100644 --- a/pam_deploy_graph/llm/factory.py +++ b/pam_deploy_graph/llm/factory.py @@ -1,4 +1,4 @@ -"""LLM client factory for CLI and embedding code.""" +"""供 CLI 和外部嵌入使用的 LLM client 工厂。""" from __future__ import annotations @@ -15,6 +15,7 @@ def build_llm_client( api_key: str | None = None, model: str | None = None, ) -> LlmClient: + """根据显式参数或环境变量构造 LLM client。""" actual_base_url = base_url or os.getenv("PAM_LLM_BASE_URL", "") actual_api_key = api_key or os.getenv("PAM_LLM_API_KEY", "") actual_model = model or os.getenv("PAM_LLM_MODEL", "") @@ -30,7 +31,7 @@ def build_llm_client( if not actual_model: missing.append("model") if missing: - raise ValueError(f"Incomplete LLM config: missing {', '.join(missing)}") + raise ValueError(f"LLM 配置不完整，缺少: {', '.join(missing)}") return OpenAICompatibleLlmClient( base_url=actual_base_url, diff --git a/pam_deploy_graph/llm/openai_compatible.py b/pam_deploy_graph/llm/openai_compatible.py index fae48de..73b865c 100644 --- a/pam_deploy_graph/llm/openai_compatible.py +++ b/pam_deploy_graph/llm/openai_compatible.py @@ -1,8 +1,7 @@ -"""OpenAI-compatible HTTP LLM client. +"""OpenAI-compatible HTTP LLM client。 -The client targets providers exposing a `/chat/completions` endpoint with -OpenAI-style request and response shapes. It intentionally uses only the Python -standard library so the runtime can stay dependency-light. +该 client 面向暴露 `/chat/completions` 的模型服务，并使用 OpenAI 风格的 +请求/响应结构。实现只依赖 Python 标准库，便于控制运行时依赖体积。 """ from __future__ import annotations @@ -28,6 +27,8 @@ JsonTransport = Callable[[str, dict[str, str], dict[str, Any], float], dict[str, class OpenAICompatibleLlmClient: + """通过 OpenAI-compatible HTTP 接口获取结构化 LLM 输出。""" + def __init__( self, *, @@ -38,12 +39,13 @@ class OpenAICompatibleLlmClient: temperature: float = 0, transport: JsonTransport | None = None, ) -> None: + """保存连接参数、模型参数和可替换的 HTTP transport。""" if not base_url: - raise ValueError("LLM base_url is required") + raise ValueError("必须配置 LLM base_url") if not api_key: - raise ValueError("LLM api_key is required") + raise ValueError("必须配置 LLM api_key") if not model: - raise ValueError("LLM model is required") + raise ValueError("必须配置 LLM model") self.base_url = base_url.rstrip("/") self.api_key = api_key self.model = model @@ -52,6 +54,7 @@ class OpenAICompatibleLlmClient: self.transport = transport or _default_transport def understand_request(self, text: str) -> LlmIntentResult: + """调用 LLM 识别用户意图。""" payload = self._complete_json(INTENT_PROMPT, {"user_text": text}) return LlmIntentResult( intent=_string(payload, "intent", "deploy"), # type: ignore[arg-type] @@ -64,6 +67,7 @@ class OpenAICompatibleLlmClient: ) def extract_params(self, text: str, base_params: dict[str, Any] | None = None) -> LlmParamResult: + """调用 LLM 抽取参数，并避免把敏感值发送进 prompt。""" original_base = dict(base_params or {}) safe_base = _redact_sensitive(original_base) payload = self._complete_json( @@ -102,6 +106,7 @@ class OpenAICompatibleLlmClient: intent: str, strategy: ExecutionStrategy, ) -> LlmDeployPlan: + """调用 LLM 生成部署计划。""" payload = self._complete_json( PLAN_PROMPT, { @@ -115,7 +120,7 @@ class OpenAICompatibleLlmClient: ) planned_actions = _string_list(payload.get("planned_actions")) or list(GLOBAL_ACTION_SEQUENCE) return LlmDeployPlan( - summary=_string(payload, "summary", "PAM deployment plan"), + summary=_string(payload, "summary", "PAM 部署计划"), risk_notes=_string_list(payload.get("risk_notes")), planned_actions=planned_actions, requires_confirmation=bool(payload.get("requires_confirmation", True)), @@ -123,6 +128,7 @@ class OpenAICompatibleLlmClient: ) def _complete_json(self, instruction: str, input_payload: dict[str, Any]) -> dict[str, Any]: + """发送 chat/completions 请求，并解析 JSON 对象响应。""" request_payload = { "model": self.model, "temperature": self.temperature, @@ -149,7 +155,7 @@ class OpenAICompatibleLlmClient: content = _message_content(response) parsed = _loads_json_object(content) if not isinstance(parsed, dict): - raise ValueError("LLM response must be a JSON object") + raise ValueError("LLM 响应必须是 JSON object") return parsed @@ -159,6 +165,7 @@ def _default_transport( payload: dict[str, Any], timeout_sec: float, ) -> dict[str, Any]: + """使用标准库 urllib 发送 JSON POST 请求。""" request = urllib.request.Request( url, data=json.dumps(payload).encode("utf-8"), @@ -169,11 +176,12 @@ def _default_transport( raw = response.read().decode("utf-8") decoded = json.loads(raw) if not isinstance(decoded, dict): - raise ValueError("LLM HTTP response must be a JSON object") + raise ValueError("LLM HTTP 响应必须是 JSON object") return decoded def _chat_completions_url(base_url: str) -> str: + """把 base_url 规范化为 chat/completions endpoint。""" clean = base_url.rstrip("/") if clean.endswith("/chat/completions"): return clean @@ -181,10 +189,11 @@ def _chat_completions_url(base_url: str) -> str: def _message_content(response: dict[str, Any]) -> Any: + """从 OpenAI-compatible 响应中提取 message.content。""" try: content = response["choices"][0]["message"]["content"] except (KeyError, IndexError, TypeError) as exc: - raise ValueError("LLM response does not contain choices[0].message.content") from exc + raise ValueError("LLM 响应缺少 choices[0].message.content") from exc if isinstance(content, list): parts: list[str] = [] for item in content: @@ -197,14 +206,16 @@ def _message_content(response: dict[str, Any]) -> Any: def _loads_json_object(content: Any) -> Any: + """把 message.content 解析为 JSON 对象。""" if isinstance(content, dict): return content if not isinstance(content, str): - raise ValueError("LLM message content must be JSON text") + raise ValueError("LLM message content 必须是 JSON 文本") return json.loads(content) def _redact_sensitive(value: Any) -> Any: + """递归脱敏 prompt 输入中的敏感字段。""" if isinstance(value, dict): redacted: dict[str, Any] = {} for key, item in value.items(): @@ -219,11 +230,13 @@ def _redact_sensitive(value: Any) -> Any: def _string(payload: dict[str, Any], key: str, default: str) -> str: + """安全读取字符串字段。""" value = payload.get(key, default) return str(value) if value is not None else default def _float(payload: dict[str, Any], key: str, default: float) -> float: + """安全读取浮点数字段。""" try: return float(payload.get(key, default)) except (TypeError, ValueError): @@ -231,10 +244,12 @@ def _float(payload: dict[str, Any], key: str, default: float) -> float: def _dict(value: Any) -> dict[str, Any]: + """确保返回 dict，非法值降级为空 dict。""" return value if isinstance(value, dict) else {} def _string_list(value: Any) -> list[str]: + """确保返回字符串列表。""" if isinstance(value, list): return [str(item) for item in value] if value in (None, ""): diff --git a/pam_deploy_graph/llm/prompts.py b/pam_deploy_graph/llm/prompts.py index 45a594c..a9f291d 100644 --- a/pam_deploy_graph/llm/prompts.py +++ b/pam_deploy_graph/llm/prompts.py @@ -1,4 +1,4 @@ -"""Prompts for structured PAM deployment planning.""" +"""用于 PAM 部署结构化理解和规划的 LLM 提示词。""" SYSTEM_PROMPT = """你是 PAM 智能部署 Agent 的结构化理解与规划组件。 diff --git a/pam_deploy_graph/llm/rule_based.py b/pam_deploy_graph/llm/rule_based.py index b78f859..566371f 100644 --- a/pam_deploy_graph/llm/rule_based.py +++ b/pam_deploy_graph/llm/rule_based.py @@ -1,8 +1,7 @@ -"""Deterministic fallback for LLM structured outputs. +"""LLM 结构化输出的确定性规则 fallback。 -This class is intentionally not a replacement for a real model. It gives the -Agent stable structured outputs for local development and tests. A real LLM -client should implement the same methods. +该类不是对真实模型的替代，只用于本地开发和测试时提供稳定输出。 +真实 LLM client 需要实现相同方法。 """ from __future__ import annotations @@ -45,7 +44,10 @@ KEY_ALIASES = { class RuleBasedLlmClient: + """基于规则的轻量 LLM client fallback。""" + def understand_request(self, text: str) -> LlmIntentResult: + """用关键词规则识别用户意图和执行策略偏好。""" lowered = text.lower() reasons: list[str] = [] intent = "deploy" @@ -87,6 +89,7 @@ class RuleBasedLlmClient: ) def extract_params(self, text: str, base_params: dict[str, Any] | None = None) -> LlmParamResult: + """从 key=value、中文短语和 IP 地址中抽取参数。""" params = dict(base_params or {}) params.update(self._extract_key_values(text)) params.update(self._extract_chinese_patterns(text)) @@ -112,6 +115,7 @@ class RuleBasedLlmClient: intent: str, strategy: ExecutionStrategy, ) -> LlmDeployPlan: + """生成确定性的部署计划和风险提示。""" if strategy == "hybrid_node_mcp": strategy_text = "PAM_HOME 使用脚本 action，PAM_NODE 使用 MCP" elif strategy == "script_only": @@ -142,6 +146,7 @@ class RuleBasedLlmClient: ) def _extract_key_values(self, text: str) -> dict[str, str]: + """抽取 KEY=VALUE 形式的参数。""" params: dict[str, str] = {} for match in re.finditer(r"([A-Za-z_][A-Za-z0-9_]*)\s*=\s*([^\s,;]+)", text): raw_key, value = match.groups() @@ -151,6 +156,7 @@ class RuleBasedLlmClient: return params def _extract_chinese_patterns(self, text: str) -> dict[str, str]: + """抽取常见中文描述中的部署参数。""" patterns = { "AIRPORT_CODE": r"(?:机场|三字码)\s*[:：]?\s*([A-Z]{3})", "APP_NAME": r"(?:应用|应用名)\s*[:：]?\s*([A-Za-z0-9_.-]+)", @@ -164,4 +170,3 @@ class RuleBasedLlmClient: if match: params[key] = match.group(1) return params - diff --git a/pam_deploy_graph/llm/validators.py b/pam_deploy_graph/llm/validators.py index 2549e8b..b68881a 100644 --- a/pam_deploy_graph/llm/validators.py +++ b/pam_deploy_graph/llm/validators.py @@ -1,4 +1,4 @@ -"""Validation and guardrails for LLM structured outputs.""" +"""LLM 结构化输出的校验和安全护栏。""" from __future__ import annotations @@ -10,18 +10,20 @@ FORBIDDEN_TEXT = ("bash ", "powershell ", "deploy.sh", "deploy.ps1", "CLIENT_SEC def validate_intent_result(result: LlmIntentResult) -> None: + """校验意图识别结果是否合法。""" if result.intent not in VALID_INTENTS: - raise ValueError(f"Invalid intent: {result.intent}") + raise ValueError(f"非法意图: {result.intent}") if not 0 <= result.confidence <= 1: - raise ValueError("Intent confidence must be between 0 and 1") + raise ValueError("意图置信度必须在 0 到 1 之间") def validate_deploy_plan(plan: LlmDeployPlan) -> None: + """校验部署计划中的 action 和文本安全性。""" invalid = [action for action in plan.planned_actions if action not in ALLOWED_ACTIONS] if invalid: - raise ValueError(f"Plan contains invalid actions: {', '.join(invalid)}") + raise ValueError(f"计划包含非法 action: {', '.join(invalid)}") combined_text = "\n".join([plan.summary, *plan.risk_notes]) lowered = combined_text.lower() forbidden = [item for item in FORBIDDEN_TEXT if item.lower() in lowered] if forbidden: - raise ValueError(f"Plan contains forbidden executable text: {', '.join(forbidden)}") + raise ValueError(f"计划包含禁止出现的可执行文本: {', '.join(forbidden)}") diff --git a/pam_deploy_graph/mcp_client.py b/pam_deploy_graph/mcp_client.py index 5104e80..8820739 100644 --- a/pam_deploy_graph/mcp_client.py +++ b/pam_deploy_graph/mcp_client.py @@ -1,8 +1,7 @@ -"""MCP client adapters. +"""MCP client 适配器。 -The Agent only needs a synchronous `call_tool(name, arguments)` surface. This -module adapts simple callables or SDK-like sessions to that surface without -forcing the rest of the codebase to import a concrete MCP SDK. +Agent 只依赖同步的 `call_tool(name, arguments)` 接口。本模块把普通 +callable 或 SDK session 适配成这个接口，避免业务代码绑定具体 MCP SDK。 """ from __future__ import annotations @@ -16,16 +15,17 @@ from typing import Any @dataclass(frozen=True) class McpClientConfig: - """Configuration needed after a real MCP session has been created.""" + """真实 MCP session 建立后需要传给 runner 的配置。""" server_name: str = "pam-node" tool_names: dict[str, str] = field(default_factory=dict) @classmethod def from_mapping(cls, payload: dict[str, Any]) -> "McpClientConfig": + """从 JSON 字典构造 MCP client 配置。""" tool_names = payload.get("tool_names") or payload.get("tools") or {} if not isinstance(tool_names, dict): - raise ValueError("MCP tool_names must be an object") + raise ValueError("MCP tool_names 必须是 JSON object") return cls( server_name=str(payload.get("server_name", "pam-node")), tool_names={str(key): str(value) for key, value in tool_names.items()}, @@ -33,43 +33,49 @@ class McpClientConfig: def load_mcp_client_config(path: str | Path) -> McpClientConfig: + """读取 MCP client JSON 配置文件。""" payload = json.loads(Path(path).read_text(encoding="utf-8")) if not isinstance(payload, dict): - raise ValueError("MCP client config must be a JSON object") + raise ValueError("MCP client 配置必须是 JSON object") return McpClientConfig.from_mapping(payload) class FunctionMcpToolClient: - """Wrap a plain Python callable as an MCP tool client.""" + """把普通 Python callable 包装为 MCP tool client。""" def __init__(self, caller: Callable[[str, dict[str, Any]], Any]) -> None: + """保存实际执行工具调用的函数。""" self.caller = caller def call_tool(self, tool_name: str, arguments: dict[str, Any]) -> Any: + """调用底层函数并返回原始结果。""" return self.caller(tool_name, arguments) class SessionMcpToolClient: - """Adapt SDK-like sessions exposing `call_tool`. + """适配暴露 `call_tool` 的 MCP SDK session。 - The adapter accepts common result shapes: + 适配器接受常见返回形态： - - raw dict/list/string - - object with `structuredContent` - - object with `content`, where text content may contain JSON + - 原始 dict/list/string + - 带有 `structuredContent` 的对象 + - 带有 `content` 的对象，其中 text 内容可能是 JSON """ def __init__(self, session: Any) -> None: + """校验并保存 MCP SDK session。""" if not hasattr(session, "call_tool"): - raise TypeError("MCP session must expose call_tool") + raise TypeError("MCP session 必须暴露 call_tool 方法") self.session = session def call_tool(self, tool_name: str, arguments: dict[str, Any]) -> Any: + """调用 SDK session，并把 SDK 返回值归一化。""" result = self.session.call_tool(tool_name, arguments) return normalize_mcp_sdk_result(result) def normalize_mcp_sdk_result(result: Any) -> Any: + """把常见 MCP SDK 返回结构归一化成 dict/list/string。""" if hasattr(result, "structuredContent"): structured = getattr(result, "structuredContent") if structured is not None: diff --git a/pam_deploy_graph/mcp_runner.py b/pam_deploy_graph/mcp_runner.py index 8cff615..a7058eb 100644 --- a/pam_deploy_graph/mcp_runner.py +++ b/pam_deploy_graph/mcp_runner.py @@ -1,4 +1,4 @@ -"""Runner wrapper for PAM_NODE MCP tools.""" +"""PAM_NODE MCP 工具的 action runner 封装。""" from __future__ import annotations @@ -9,7 +9,10 @@ from .output_parser import parse_mcp_result class McpToolClient(Protocol): + """MCP 工具客户端需要实现的最小同步接口。""" + def call_tool(self, tool_name: str, arguments: dict[str, Any]) -> Any: + """调用指定 MCP tool，并返回工具原始输出。""" ... @@ -28,11 +31,14 @@ DEFAULT_NODE_MCP_TOOLS = { class McpActionRunner: + """把 Agent action 转换为 MCP tool 调用。""" + def __init__( self, client: McpToolClient | None = None, tool_names: dict[str, str] | None = None, ) -> None: + """保存 MCP client 和 action 到 tool name 的映射。""" self.client = client self.tool_names = tool_names or DEFAULT_NODE_MCP_TOOLS.copy() @@ -46,11 +52,12 @@ class McpActionRunner: stop_first: bool = False, **_: Any, ) -> ActionResult: + """执行一个 PAM_NODE action，并归一化为 ActionResult。""" if self.client is None: - raise RuntimeError("MCP client is not configured") + raise RuntimeError("尚未配置 MCP client") tool_name = self.tool_names.get(action) if not tool_name: - raise ValueError(f"No MCP tool mapped for action: {action}") + raise ValueError(f"action 未映射 MCP tool: {action}") arguments = self._build_arguments( action, params=params, @@ -60,7 +67,7 @@ class McpActionRunner: ) try: payload = self.client.call_tool(tool_name, arguments) - except Exception as exc: # pragma: no cover - defensive wrapper + except Exception as exc: # pragma: no cover - 防御性异常包装 return parse_mcp_result(action, {}, ok=False, tool_name=tool_name, error=str(exc)) return parse_mcp_result(action, payload, ok=True, tool_name=tool_name) @@ -73,6 +80,7 @@ class McpActionRunner: hash_code: str | None, stop_first: bool, ) -> dict[str, Any]: + """把 Agent 参数转换为 MCP tool 所需的入参。""" arguments = { "homeBaseUrl": params.get("HOME_BASE_URL"), "airportCode": params.get("AIRPORT_CODE"), @@ -90,4 +98,3 @@ class McpActionRunner: if action == "rollback-ip": arguments["stopFirst"] = stop_first return {key: value for key, value in arguments.items() if value not in (None, "")} - diff --git a/pam_deploy_graph/models.py b/pam_deploy_graph/models.py index 94c88bd..d621496 100644 --- a/pam_deploy_graph/models.py +++ b/pam_deploy_graph/models.py @@ -1,4 +1,4 @@ -"""Shared dataclasses for the PAM deploy agent.""" +"""PAM 部署 Agent 共享数据模型。""" from __future__ import annotations @@ -14,6 +14,8 @@ StrategyPreference = Literal["hybrid_node_mcp", "script_only", "fake", "未指 @dataclass(slots=True) class ActionResult: + """单个 action 的统一执行结果。""" + action: str backend: BackendName ok: bool @@ -28,6 +30,8 @@ class ActionResult: @dataclass(slots=True) class SkillPolicy: + """从 Skill 文档提取出的部署策略约束。""" + name: str source_path: str description: str = "" @@ -51,6 +55,8 @@ class SkillPolicy: @dataclass(slots=True) class LlmIntentResult: + """LLM 意图识别结果。""" + intent: IntentName mode_preference: ModePreference = "未指定" strategy_preference: StrategyPreference = "未指定" @@ -62,6 +68,8 @@ class LlmIntentResult: @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) @@ -71,6 +79,8 @@ class LlmParamResult: @dataclass(slots=True) class LlmDeployPlan: + """LLM 生成的部署计划。""" + summary: str risk_notes: list[str] = field(default_factory=list) planned_actions: list[str] = field(default_factory=list) @@ -80,6 +90,8 @@ class LlmDeployPlan: @dataclass(slots=True) class AgentState: + """一次部署运行的完整状态，可序列化到 checkpoint。""" + run_id: str params: dict[str, Any] execution_strategy: ExecutionStrategy diff --git a/pam_deploy_graph/output_parser.py b/pam_deploy_graph/output_parser.py index d3a48d7..6e1f2fc 100644 --- a/pam_deploy_graph/output_parser.py +++ b/pam_deploy_graph/output_parser.py @@ -1,4 +1,4 @@ -"""Normalize script stdout and MCP tool returns into ActionResult objects.""" +"""把脚本 stdout 和 MCP tool 返回值归一化为 ActionResult。""" from __future__ import annotations @@ -14,6 +14,7 @@ KEY_VALUE_RE = re.compile(r"^(?P[A-Za-z_][A-Za-z0-9_]*)=(?P.*)$") def redact_text(text: str) -> str: + """对文本中的敏感字段值进行脱敏。""" redacted = text for key in SENSITIVE_KEYS: redacted = re.sub( @@ -26,6 +27,7 @@ def redact_text(text: str) -> str: def parse_key_values(text: str) -> dict[str, Any]: + """解析脚本输出中的 KEY=VALUE 行，重复 IP 会聚合为列表。""" values: dict[str, Any] = {} for raw_line in text.splitlines(): line = raw_line.strip() @@ -42,6 +44,7 @@ def parse_key_values(text: str) -> dict[str, Any]: def normalize_mcp_values(payload: Any) -> dict[str, Any]: + """把 MCP 返回值归一化为脚本兼容的字段名。""" if isinstance(payload, str): try: payload = json.loads(payload) @@ -75,6 +78,7 @@ def parse_script_result( backend: BackendName = "script", tool_name: str = "", ) -> ActionResult: + """解析脚本执行结果，并识别待人工确认标记。""" raw_output = redact_text("\n".join(part for part in (stdout, stderr) if part)) values = parse_key_values(stdout) pending = PENDING_CONFIRMATION_RE.search(stdout) or PENDING_CONFIRMATION_RE.search(stderr) @@ -105,6 +109,7 @@ def parse_mcp_result( tool_name: str = "", error: str = "", ) -> ActionResult: + """解析 MCP tool 返回值，并包装为统一 ActionResult。""" values = normalize_mcp_values(payload) raw_output = redact_text(json.dumps(payload, ensure_ascii=False, default=str)) return ActionResult( @@ -117,11 +122,12 @@ def parse_mcp_result( stdout=raw_output if ok else "", stderr=redact_text(error), raw_output=raw_output, - error_summary="" if ok else redact_text(error or "MCP tool failed"), + error_summary="" if ok else redact_text(error or "MCP tool 执行失败"), ) def _summarize_error(stderr: str, stdout: str, pending: str) -> str: + """从 stderr/stdout/确认标记中提取简短错误摘要。""" if pending: return pending for text in (stderr, stdout): @@ -129,5 +135,4 @@ def _summarize_error(stderr: str, stdout: str, pending: str) -> str: stripped = line.strip() if stripped: return redact_text(stripped) - return "Action failed" - + return "action 执行失败" diff --git a/pam_deploy_graph/params_loader.py b/pam_deploy_graph/params_loader.py index 759c8c9..d2ca036 100644 --- a/pam_deploy_graph/params_loader.py +++ b/pam_deploy_graph/params_loader.py @@ -1,4 +1,4 @@ -"""Load deploy parameters from JSON or config.txt style files.""" +"""从 JSON 或 config.txt 风格文件读取部署参数。""" from __future__ import annotations @@ -8,6 +8,7 @@ from typing import Any def load_params_file(path: str | Path) -> dict[str, Any]: + """自动识别 JSON/config.txt 格式，并返回参数字典。""" config_path = Path(path) text = config_path.read_text(encoding="utf-8") stripped = text.lstrip() @@ -24,4 +25,3 @@ def load_params_file(path: str | Path) -> dict[str, Any]: key, value = line.split("=", 1) values[key.strip()] = value.strip() return values - diff --git a/pam_deploy_graph/script_runner.py b/pam_deploy_graph/script_runner.py index 3f87872..5f6d27c 100644 --- a/pam_deploy_graph/script_runner.py +++ b/pam_deploy_graph/script_runner.py @@ -1,4 +1,4 @@ -"""Subprocess runner for deploy.sh and deploy.ps1 action calls.""" +"""通过子进程执行 deploy.sh / deploy.ps1 action。""" from __future__ import annotations @@ -11,7 +11,10 @@ from .output_parser import parse_script_result class ScriptActionRunner: + """脚本 action runner，负责构造命令、执行脚本并解析结果。""" + def __init__(self, script_base_dir: str | Path = "doc_scripts") -> None: + """保存脚本所在目录。""" self.script_base_dir = Path(script_base_dir) def run( @@ -27,6 +30,7 @@ class ScriptActionRunner: trace_file_path: str | None = None, timeout_sec: int | None = None, ) -> ActionResult: + """执行一个脚本 action，并返回统一 ActionResult。""" command = self.build_command( action, script_entry=script_entry, @@ -64,6 +68,7 @@ class ScriptActionRunner: stop_first: bool = False, trace_file_path: str | None = None, ) -> list[str]: + """根据脚本类型构造 action 命令行参数。""" if script_entry == "deploy.sh": command = [ "bash", @@ -101,14 +106,14 @@ class ScriptActionRunner: command.append("-RollbackStopFirst") return command - raise ValueError(f"Unsupported script entry: {script_entry}") + raise ValueError(f"不支持的脚本入口: {script_entry}") def select_script_entry(os_name: str | None = None) -> str: + """根据操作系统选择默认脚本入口。""" import platform name = (os_name or platform.system()).lower() if "windows" in name: return "deploy.ps1" return "deploy.sh" - diff --git a/pam_deploy_graph/skill_policy.py b/pam_deploy_graph/skill_policy.py index 5ce3d19..f07ba20 100644 --- a/pam_deploy_graph/skill_policy.py +++ b/pam_deploy_graph/skill_policy.py @@ -1,4 +1,4 @@ -"""Load the PAM deploy Skill document into a compact policy object.""" +"""把 PAM 部署 Skill 文档加载为简化策略对象。""" from __future__ import annotations @@ -15,6 +15,7 @@ from .models import SkillPolicy def load_skill_policy(path: str | Path) -> SkillPolicy: + """读取 Skill markdown 头部信息，并填充 action/参数策略。""" skill_path = Path(path) text = skill_path.read_text(encoding="utf-8") name = "pam-auto-deply" @@ -39,4 +40,3 @@ def load_skill_policy(path: str | Path) -> SkillPolicy: action_sequence=GLOBAL_ACTION_SEQUENCE, ip_action_sequence=IP_ACTION_SEQUENCE, ) - diff --git a/tests/test_graph.py b/tests/test_graph.py index c66881c..cb63d63 100644 --- a/tests/test_graph.py +++ b/tests/test_graph.py @@ -17,7 +17,7 @@ def test_build_graph_or_none_without_langgraph_is_safe(): def test_build_langgraph_error_without_dependency_is_clear(): if importlib.util.find_spec("langgraph"): pytest.skip("langgraph installed") - with pytest.raises(RuntimeError, match="langgraph is not installed"): + with pytest.raises(RuntimeError, match="未安装 langgraph"): build_langgraph() diff --git a/tests/test_llm_structured.py b/tests/test_llm_structured.py index 873aa2f..09f5b89 100644 --- a/tests/test_llm_structured.py +++ b/tests/test_llm_structured.py @@ -69,7 +69,7 @@ def test_plan_guardrails_reject_executable_text(): try: validate_deploy_plan(plan) except ValueError as exc: - assert "forbidden" in str(exc) + assert "禁止" in str(exc) else: raise AssertionError("expected guardrail failure")