Messages & types¶
The envelope and payload types that flow between nodes: messages, headers, working memory, and streaming/planner payload objects.
Typed message and controller models for PenguiFlow.
Bases: BaseModel
Envelope wrapping a payload as it flows through a PenguiFlow graph.
Every value passed between nodes is carried inside a Message (or is a bare
payload wrapped in one by the runtime). It couples the actual data (payload)
with tracing, deadline, and metadata fields that the runtime and middlewares use to
observe, budget, and correlate work across an entire run.
Attributes:
| Name | Type | Description |
|---|---|---|
payload |
Any
|
The actual data being routed to a node. Its type is validated
per-node against the |
headers |
Headers
|
Routing/provenance envelope (tenant, topic, priority) associated with this message. |
trace_id |
str
|
Identifier shared by every message that belongs to the same logical run. Generated automatically per top-level message; propagated unchanged to child messages (streams, subflows, controller hops) so all related work can be correlated in logs, metrics, and cancellation. |
ts |
float
|
Unix timestamp (seconds) recording when this message instance was created. |
deadline_s |
float | None
|
Optional absolute wall-clock deadline (Unix timestamp, seconds).
When set, the runtime treats messages observed after this time as expired:
pending work for the trace is skipped and controller loops emit a
deadline-exceeded |
meta |
dict[str, Any]
|
Free-form metadata dictionary (e.g. debug info, cost tracking, custom context) that is copied forward to derived messages (streamed chunks, controller hops) so it survives fan-out/fan-in and subflow boundaries. |
class-attribute
instance-attribute
¶
trace_id: str = Field(default_factory=lambda: uuid.uuid4().hex, description='Identifier shared by every message belonging to the same logical run; used to correlate work across nodes, streams, and subflows.')
class-attribute
instance-attribute
¶
ts: float = Field(default_factory=time.time, description='Unix timestamp (seconds) when this message instance was created.')
class-attribute
instance-attribute
¶
deadline_s: float | None = Field(default=None, description="Optional absolute wall-clock deadline (Unix timestamp, seconds) after which the runtime treats this trace's work as expired.")
class-attribute
instance-attribute
¶
meta: dict[str, Any] = Field(default_factory=dict, description='Free-form metadata (debug/cost/context) carried forward to derived messages across fan-out/fan-in and subflow boundaries.')
Bases: BaseModel
Routing and provenance envelope attached to every Message.
Headers travels alongside a Message payload and is preserved as messages move
between nodes, fan out, and fan back in. Nodes and middlewares can read it to make
tenant-aware or priority-aware decisions without inspecting the payload itself.
Attributes:
| Name | Type | Description |
|---|---|---|
tenant |
str
|
Identifier of the tenant/customer the message belongs to. Used, for example, by remote/A2A call sites to resolve which tenant a downstream request should be attributed to. |
topic |
str | None
|
Optional free-form routing topic (e.g. a queue or subject name) that application code can use to steer messages without changing payload types. |
priority |
int
|
Optional priority hint for consumers that implement priority-aware scheduling. Higher values are conventionally treated as higher priority; the core runtime does not itself reorder work based on this field. |
Bases: BaseModel
Working memory carried through a controller loop's hops.
WM (working memory) is the payload a controller-style flow passes to itself
across hops (see PenguiFlow's controller postprocessing). It tracks accumulated
facts and enforces hop/token budgets and a wall-clock deadline (via the wrapping
Message.deadline_s): once a budget or the deadline is exceeded, the runtime
replaces the payload with a FinalAnswer and routes it to the Rookery instead of
continuing the loop.
Attributes:
| Name | Type | Description |
|---|---|---|
query |
str
|
The original user query or task description driving this run. |
facts |
list[Any]
|
Accumulated evidence/results gathered across hops so far. |
hops |
int
|
Number of hops completed so far in this controller loop. |
budget_hops |
int | None
|
Maximum number of hops allowed before the runtime forces
termination with a budget-exceeded |
tokens_used |
int
|
Running count of tokens consumed so far across hops. |
budget_tokens |
int | None
|
Maximum tokens allowed before the runtime forces termination
with a budget-exceeded |
confidence |
float
|
Planner-reported confidence score (application-defined scale, typically 0.0-1.0) for the current working state. |
class-attribute
instance-attribute
¶
query: str = Field(description='The original user query or task description driving this run.')
class-attribute
instance-attribute
¶
facts: list[Any] = Field(default_factory=list, description='Accumulated evidence/results gathered across hops so far.')
class-attribute
instance-attribute
¶
hops: int = Field(default=0, description='Number of hops completed so far in this controller loop.')
class-attribute
instance-attribute
¶
budget_hops: int | None = Field(default=8, description='Maximum number of hops allowed before the runtime forces termination; None disables it.')
class-attribute
instance-attribute
¶
tokens_used: int = Field(default=0, description='Running count of tokens consumed so far across hops.')
class-attribute
instance-attribute
¶
budget_tokens: int | None = Field(default=None, description='Maximum tokens allowed before the runtime forces termination; None disables it.')
class-attribute
instance-attribute
¶
confidence: float = Field(default=0.0, description='Planner-reported confidence score for the current working state.')
Bases: BaseModel
A single piece of incrementally streamed output (e.g. LLM tokens).
Nodes that stream partial results (via Context.emit_chunk) produce a sequence of
StreamChunk payloads sharing the same stream_id. Consumers reassemble the
full output by ordering chunks on seq and stopping once a chunk with
done=True is received.
Attributes:
| Name | Type | Description |
|---|---|---|
stream_id |
str
|
Identifier grouping all chunks belonging to the same stream.
Typically defaults to the originating message's |
seq |
int
|
Monotonically increasing sequence number within |
text |
str
|
The partial text content carried by this chunk. |
done |
bool
|
Whether this is the final chunk of the stream. When |
meta |
dict[str, Any]
|
Free-form metadata attached to this specific chunk (distinct from the
wrapping |
class-attribute
instance-attribute
¶
stream_id: str = Field(description='Identifier grouping all chunks belonging to the same stream.')
class-attribute
instance-attribute
¶
seq: int = Field(description='Monotonically increasing sequence number within stream_id, starting at 0.')
class-attribute
instance-attribute
¶
text: str = Field(description='The partial text content carried by this chunk.')
class-attribute
instance-attribute
¶
done: bool = Field(default=False, description='Whether this is the final chunk of the stream.')
class-attribute
instance-attribute
¶
meta: dict[str, Any] = Field(default_factory=dict, description='Free-form metadata attached to this specific chunk.')
Bases: BaseModel
Terminal payload signaling a controller loop (or flow) has produced its result.
When a node returns a Message whose payload is a FinalAnswer, the runtime
routes it directly to the Rookery instead of continuing to iterate, ending the
controller loop. It is also the payload the runtime synthesizes itself when a
deadline or hop/token budget is exceeded (see WM).
Attributes:
| Name | Type | Description |
|---|---|---|
text |
str
|
The final answer text to return to the caller. |
citations |
list[str]
|
Optional list of citation identifiers/sources supporting the answer. |
class-attribute
instance-attribute
¶
text: str = Field(description='The final answer text to return to the caller.')
class-attribute
instance-attribute
¶
citations: list[str] = Field(default_factory=list, description='Optional list of citation identifiers/sources supporting the answer.')
Bases: BaseModel
A planner's decision for the current controller hop.
Produced by a planning/reasoning node (e.g. an LLM-backed controller) to describe
what should happen next: which PlanStep instances to run, why, and whether the
loop is finished.
Attributes:
| Name | Type | Description |
|---|---|---|
steps |
list[PlanStep]
|
Ordered list of |
rationale |
str
|
Human-readable explanation of why these steps (or none) were chosen; useful for logging/debugging planner behavior. |
done |
bool
|
Whether the controller loop should stop after this hop (e.g. because a final answer is ready) rather than continue iterating. |
class-attribute
instance-attribute
¶
steps: list[PlanStep] = Field(description='Ordered list of PlanStep instances to execute for this hop.')
class-attribute
instance-attribute
¶
rationale: str = Field(description='Human-readable explanation of why these steps (or none) were chosen.')
class-attribute
instance-attribute
¶
done: bool = Field(default=False, description='Whether the controller loop should stop after this hop.')
Bases: BaseModel
A single unit of work chosen by a controller/planner for one hop.
A Thought bundles one or more PlanStep instances describing what a
controller loop wants to execute next (e.g. run a retrieval tool, issue a web
search, run SQL, summarize, route, or stop).
Attributes:
| Name | Type | Description |
|---|---|---|
kind |
Literal['retrieve', 'web', 'sql', 'summarize', 'route', 'stop']
|
The category of action this step performs. |
args |
dict[str, Any]
|
Arguments for the step, interpreted by whatever executes it (e.g. a tool or node); shape is application-defined. |
max_concurrency |
int
|
Maximum number of concurrent executions allowed for this step when it fans out to multiple parallel invocations. |
class-attribute
instance-attribute
¶
kind: Literal['retrieve', 'web', 'sql', 'summarize', 'route', 'stop'] = Field(description="Category of action this step performs; 'stop' signals the controller loop should terminate.")
class-attribute
instance-attribute
¶
args: dict[str, Any] = Field(default_factory=dict, description='Arguments for the step, interpreted by whatever executes it.')
class-attribute
instance-attribute
¶
max_concurrency: int = Field(default=1, description='Maximum number of concurrent executions allowed for this step.')