指南 07 · 自定义节点¶
内置节点不够用时，框架允许你用「声明 + 实现 + 注册」三步扩展自己的节点。节点会进入注册中心，与内置节点同等使用。
1. 节点形态回顾¶
每个节点是「静态声明 + 运行时实现」：
- 声明（类属性
declaration = NodeDeclaration(...)）：名字、描述、输入输出 schema、是否输出节点。 - 实现（
__call__(inputs, ctx)）：只管「算」，返回声明中outputs的键。
from agentframework.nodes.base import FieldSchema, Node, NodeDeclaration
from agentframework.nodes.registry import register
@register
class MyNode(Node):
declaration = NodeDeclaration(
name="MyNode",
description="示例节点",
inputs={"text": FieldSchema(type="str", required=False)},
outputs={"result": FieldSchema(type="str")},
)
def __call__(self, inputs, ctx):
# 只管算：读 inputs、写 ctx（消息/变量/快照），返回 outputs 键
return {"result": inputs.get("text", "默认值")}
约定（由编译管线保证，你无需自己实现）：
- 输入由管线按声明解析（常量 /
wf.ref/ state 同名 channel）后注入inputs。 - 返回值必须覆盖声明中全部 outputs 键，缺键报错。
- 实现内不直接读写 state channel；要访问运行依赖（消息 / 变量 / 流式回调 / 快照）经
ctx。
2. 声明字段类型¶
FieldSchema(type=..., description=..., required=...)，type 是字符串：
- 基础类型：
str/int/float/bool/list/dict - 语义类型最小集：
chatHistory（对话历史，运行期只校验容器是 list） any：通配（不校验）
3. 运行时 ctx 提供的能力¶
你的节点实现里常用的 ctx 能力：
class MyNode(Node):
def __call__(self, inputs, ctx):
ctx.messages # 本次请求的对话历史（BaseMessage 列表）
ctx.variables # 内部控制变量表（可读写）
ctx.emit_stream(text) # 产生流式增量 → 网关 answer 事件（输出节点用）
ctx.add_detail(
module_name=self.name, module_type=self.declaration.name, data={...}
) # 记录节点快照 → detail/flowNodeStatus
...
4. 输出节点标记¶
若你的节点内容应进最终 answer（例如某种 Answer），把声明标为输出节点：
declaration = NodeDeclaration(
name="MyOutputNode",
...,
is_output=True, # 输出节点：内容进最终 answer（严格收口）
)
实例也可用 output 参数覆盖默认（参考 ChatLLM 的 is_output 属性读 params["output"]）。
5. 接入工作流并使用¶
自定义节点注册后，按节点类型名分发；实例化 + add_node 与内置节点一致：
from agentframework import Workflow
wf = Workflow(name="demo")
wf.add_node(MyNode(name="mine", text="hello")) # 构造参数进 params，按声明解析
wf.add_node(Answer(name="ans", text=wf.ref("mine.result")))
wf.add_edge("mine", "ans")
6. 建议（工程规范）¶
- 公共方法/
__call__写 docstring（含示例:最小用法）。 - 需要把节点作为框架交付的一部分时，在
nodes/__init__.py导出，并补对应单测。 - 若要新增独立能力模块（非单个节点），按仓库
doc/Agent框架骨架设计.md的分层落位，遵守 src layout 与命名约定。