plaita-nodes

plaita 的通用节点集（infra 级）：把 plaita 声明式流程接到真实世界——Agent CLI、 本地命令、微信人工确认、通知、文件，以及 REST API / SQL / 邮件 / IM webhook 等 外部系统，外加 gate / rate_limit / report 等流程控制原子。

背景：ADR-2026-08-27 编排双轨收敛 ——编排内核收敛到 plaita，Agent 执行层统一走 agentproc；本仓是这层决议的节点承载。

节点一览

连接器族（api_request 起的 8 行）经 credential 字段按名引用，plaita.credentials 解密读取（编排台「凭据」页创建）。hitl/hitl_await 共享发送协议层 （_hitl_client.py）：发消息/图片降级细节一处维护。

节点经 pyproject 的 [project.entry-points."plaita.nodes"] 自动注册；plaita_nodes.register_all() 可手动注册。

原子节点设计原则

抽象一条硬标准——原子性（一个节点只做一件不可再分的事）、通用性 （不绑定业务语义与特定凭证）、普适性（覆盖一类外部交互）：

• Agent ≠ LLM：agentrun 是多步工具循环的 Agent 原子（重）；llm 是单次 补全的 LLM 原子（轻）。流程里"摘要/改写/抽取/分类"用 llm，"多步编码/ 工具任务"用 agentrun。
• 决策 ≠ 生成：decision 是单步、封闭决策空间的判断原子（路由/分类/ 打分），输出类型化决策 + 置信度，choice 必落在决策空间内；开放文本生成 归 llm。理念对标 System One 决策模型（如 TypeSafe Jev）——流程里 "该走哪个分支"用 decision，"写一段话"用 llm。
• 纯文本变换（判决提取、frontmatter 解析等）不做节点——注册为表达式 F.* 函数（ExpressionRegistry.register），在 assignment 里一行使用。
• 业务领域的状态机（如内容池销账）属于业务仓，不放本仓。

快速上手

# monorepo 内可编辑安装（plaita / agentproc 均为兄弟仓）
pip install -e ../plaita[http] -e ../agentproc/sdk/python -e .

from plaita import Flow

flow = Flow.from_string(open("flow.json").read())
result = flow.run(track="default", platforms=["twitter"], topic="", repo="/path/to/repo")

JSON 用法示例（agentrun + 模板表达式）：

{
  "type": "agentrun", "id": "brief",
  "agent": "glm-52",
  "prompt": "{% $F.concat($INPUT.item.brief_prompt) %}",
  "next": "write"
}

agents.json / providers.json 兼容性

配置搜索顺序与 flowcast 一致：~/.flowx → ~/.flowcast → <repo>/.flowcast（深合并）。

与 flowcast 的两处行为差异（有意为之）：

1. agents.json 里的 env 字段 flowcast 白名单会静默丢弃，本仓按配置透传 （如 glm-52 的 RECURSIVE_MAX_TOKENS）。
2. recursive 直路径 flowcast 默认无超时，本仓 timeout_secs 默认 1800。

设计边界

• 安全：任何日志不打 apiKey / ANTHROPIC_AUTH_TOKEN。
• dry-run：所有有副作用的节点尊重 globalContext.dry_run——agentrun/capture/gate/hitl/hitl_await 与全部连接器（webhook×4 / generic_webhook / api_request / sql_query / email_send）dry 下不解析凭据、不连网，返回带 dry_run 标记的 fake 结果；writefile 照常写（草稿便于检查）。
• 新增执行器：在 agentproc executor 层扩展 + config.EXECUTOR_ALIASES 加映射，本仓节点无需改动。

断点续跑（Distributed 模式）

• hitl（阻塞版，Normal 模式）：execute 内轮询到底——进程挂了，等待即丢。

• hitl_await（挂起版，Distributed 模式）：发消息即返回 pending 并快照挂起——等微信回复期间进程可崩溃/重启（ADR-2026-08-27 phase 2）。

• hitl_poller（恢复桥，独立进程）：轮询 hitl-server 的待确认 session，回复到达即向 EventBus 发布 hitl_reply 事件唤醒挂起节点：

  python -m plaita_nodes.hitl_poller \
    --hitl-url http://127.0.0.1:8081 \
    --sessions sessions.json \
    --bus redis://127.0.0.1:6379/0 \
    [--interval 5]
  

  sessions.json 为 {execution_id: {"session_id": ..., "node_id": ...}} 映射， 由调用方挂起后落盘；事件发布（replied/timeout）后对应条目自动移除。 --bus memory 仅限调试——事件只发布在 poller 进程内，没有消费者；生产走 redis。

开发

pip install -e ".[dev]"
pytest

变更摘要

• 0.7.0（2026-09-30）：修复、契约收口与结构收敛。
  • 修复：gate.max_retries 假重试（Popen 在循环外导致从未真正重跑，现每轮重新执行命令）；移除 sql_query 遗留调试输出（params 泄漏风险）；hitl_poller 缺 os import（默认 --hitl-url 路径 NameError）。
  • dry-run 契约：webhooks×4 / generic_webhook / api_request / email_send / sql_query 补齐——dry 下不解析凭据、不连网，返回带 dry_run 标记的 fake 结果。
  • 结构：新增 _hitl_client.py 共享发送协议层，hitl / hitl_await 去重 ~30 行（HitlError 移至该层，.hitl 保持再导出）。
  • 功能：rate_limit 新增 acquire 原子 check+record（flock 互斥），check/record 两步间不再 crash/并发双发。
  • 工程：新增 Tests CI（3.10–3.12 矩阵，plaita/agentproc 兄弟仓 checkout 后 editable 安装）；jevlike extras（torch）显式声明；README「断点续跑」小节沉淀 hitl_poller 运维用法 + 冒烟测试。
  • 守卫：__all__ 漂移修复（report_append→append_entry）并补齐新节点导出；新增 pyproject entry-points ↔ _ALL_NODES ↔ __all__ 一致性测试；补 rate_limit / report / gate 重试测试；README 节点表补全 22 节点。
  • 追踪：report 节点的内核 workaround 状态立案 plaita#16。
• 0.6.0（2026-09-29）：新增 github_comment / git_publish / parse_json 三节点——出害口消毒+去重、幂等发布、LLM 输出健壮解析（issue-pipeline #43 解析策略沉淀）。修复 __version__ 漂移（0.4.0 → 与 pyproject 同步）。