Metrics & middleware¶
Observability primitives: the structured flow event, the pluggable middleware hook, and helpers for formatting events and configuring logging.
Observability primitives for PenguiFlow.
dataclass
¶
FlowEvent(event_type: str, ts: float, node_name: str | None, node_id: str | None, trace_id: str | None, attempt: int, latency_ms: float | None, queue_depth_in: int, queue_depth_out: int, outgoing_edges: int, queue_maxsize: int, trace_pending: int | None, trace_inflight: int, trace_cancelled: bool, extra: Mapping[str, Any] = dict())
Structured runtime event emitted around node execution.
Attributes:
| Name | Type | Description |
|---|---|---|
event_type |
str
|
Kind of event (e.g. |
ts |
float
|
Unix timestamp (seconds) at which the event was recorded. |
node_name |
str | None
|
Name of the node associated with the event, if any. |
node_id |
str | None
|
Identifier of the node instance associated with the event, if any. |
trace_id |
str | None
|
Identifier of the trace associated with the event, if any. |
attempt |
int
|
Attempt number for the node execution (1-indexed). |
latency_ms |
float | None
|
Duration of the associated operation in milliseconds, if available. |
queue_depth_in |
int
|
Depth of the node's inbound queue at event time. |
queue_depth_out |
int
|
Depth of the node's outbound queue at event time. |
outgoing_edges |
int
|
Number of outgoing edges from the node. |
queue_maxsize |
int
|
Configured maximum size of the node's queue. |
trace_pending |
int | None
|
Number of pending messages for the trace, if tracked. |
trace_inflight |
int
|
Number of in-flight messages for the trace. |
trace_cancelled |
bool
|
Whether the trace has been cancelled. |
extra |
Mapping[str, Any]
|
Additional event-specific data as an immutable mapping. |
property
¶
error_payload: Mapping[str, Any] | None
Return the structured FlowError payload if present.
Returns:
| Type | Description |
|---|---|
Mapping[str, Any] | None
|
An immutable mapping built from the |
Mapping[str, Any] | None
|
|
property
¶
queue_depth: int
Return the combined depth of incoming and outgoing queues.
Returns:
| Type | Description |
|---|---|
int
|
The sum of |
to_payload() -> dict[str, Any]
Render a dictionary payload suitable for structured logging.
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
A mutable dictionary with core event fields plus any entries from |
dict[str, Any]
|
(and |
metric_samples() -> dict[str, float]
Derive numeric metrics for integrations such as MLflow.
Returns:
| Type | Description |
|---|---|
dict[str, float]
|
A dictionary of numeric samples (queue depths, attempt, trace inflight, |
dict[str, float]
|
trace cancelled as 0.0/1.0, and latency when available). |
tag_values() -> dict[str, str]
Return string tags describing the event.
Returns:
| Type | Description |
|---|---|
dict[str, str]
|
A dictionary of string tags including |
dict[str, str]
|
|
Middleware hooks for PenguiFlow.
Bases: Protocol
Base middleware signature receiving :class:FlowEvent objects.
log_flow_events(logger: Logger | None = None, *, start_level: int = logging.INFO, success_level: int = logging.INFO, error_level: int = logging.ERROR, latency_callback: LatencyCallback | None = None) -> Middleware
Return middleware that emits structured node lifecycle logs.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
logger
|
Logger | None
|
Optional :class: |
None
|
start_level
|
int
|
Logging level for |
INFO
|
success_level
|
int
|
Logging level for |
INFO
|
error_level
|
int
|
Logging level for |
ERROR
|
latency_callback
|
LatencyCallback | None
|
Optional callable invoked with |
None
|
Developer-facing debugging helpers for PenguiFlow.
format_flow_event(event: FlowEvent) -> dict[str, Any]
Return a structured payload ready for logging.
The returned dictionary mirrors :meth:FlowEvent.to_payload and flattens any
embedded FlowError payload so that log aggregators can index the error
metadata (flow_error_code, flow_error_message, ...).
Logging utilities for PenguiFlow with structured output support.
ExtraFormatter(fmt: str | None = None, datefmt: str | None = None)
Bases: Formatter
A formatter that appends extra fields to the standard log format.
Output example
INFO:penguiflow.planner:planner_action [step=1, thought="...", next_node="search"]
StructuredFormatter(include_timestamp: bool = True, indent: int | None = None)
Bases: Formatter
A formatter that outputs JSON lines with all extra fields included.
This is useful for debugging and log aggregation systems that expect structured JSON output.
class-attribute
instance-attribute
¶
STANDARD_FIELDS = frozenset({'name', 'msg', 'args', 'created', 'filename', 'funcName', 'levelname', 'levelno', 'lineno', 'module', 'msecs', 'pathname', 'process', 'processName', 'relativeCreated', 'stack_info', 'exc_info', 'exc_text', 'thread', 'threadName', 'taskName', 'message'})
configure_logging(level: int | str = logging.INFO, *, structured: bool = False, include_extras: bool = True, logger_name: str = 'penguiflow') -> None
Configure penguiflow logging with optional structured output.
Parameters¶
level : int | str Log level (e.g., logging.DEBUG, "DEBUG", logging.INFO) structured : bool If True, output JSON lines. If False, use human-readable format. include_extras : bool If True (and structured=False), append extra fields to log lines. logger_name : str Logger name to configure. Default: "penguiflow"
Examples¶
from penguiflow.logging import configure_logging configure_logging(level="DEBUG", structured=True)
For debugging arg-fill issues:¶
configure_logging(level="DEBUG", include_extras=True)