Merge pull request #816 from Pipelex/release/v0.23.7

Release/v0.23.7

Author: thomashebrard

.pipelex/pipelex.toml

@@ -42,6 +42,7 @@ stuff_json_content = true
 stuff_text_content = true
 stuff_html_content = true
 error_stack_traces = true
+pipe_and_concept_registry = true
 
 [pipelex.pipeline_execution_config.graph_config.graphs_inclusion]
 # Control which graph outputs are generated

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Changelog
 
+## [v0.23.7] - 2026-04-06
+
+### Added
+
+- **Graph tracing for pipe run data**: Pipe run data and concept are now included inside the flowchart graph spec, enabling richer visualization of pipe execution results across all pipe types (LLM, extract, compose, search, image gen, sequence, condition, batch, parallel).
+- **Assignment pipe**: New `pipe_assignment` pattern for direct value assignment within pipe execution.
+
 ## [v0.23.6] - 2026-04-06
 
 ### Changed

pipelex/core/pipes/pipe_abstract.py

@@ -84,6 +84,37 @@ def concept_dependencies(self) -> list[Concept]:
 
         return unique_concepts
 
+    def _register_execution_data(self, job_metadata: JobMetadata, execution_data: dict[str, Any]) -> None:
+        """Register execution metadata with the graph tracer.
+
+        Called by pipe subclasses during execution to capture runtime-resolved data
+        (rendered prompts, resolved models, etc.) for the GraphSpec.
+        """
+        graph_context = job_metadata.graph_context
+        if graph_context is None:
+            return
+        tracer_manager = GraphTracerManager.get_instance()
+        if tracer_manager is None or graph_context.parent_node_id is None:
+            return
+        tracer_manager.register_execution_data(
+            graph_id=graph_context.graph_id,
+            node_id=graph_context.parent_node_id,
+            execution_data=execution_data,
+        )
+
+    def _make_single_concept_data_for_registry(self, concept: Concept) -> dict[str, Any]:
+        """Serialize a single concept for the graph registry, including its JSON Schema."""
+        concept_dict = concept.model_dump(mode="json")
+        try:
+            concept_dict["json_schema"] = concept.get_structure_class().model_json_schema()
+        except (TypeError, ValueError):
+            concept_dict["json_schema"] = None
+        return concept_dict
+
+    def _make_concept_data_for_registry(self) -> list[dict[str, Any]]:
+        """Serialize all unique concepts from this pipe for the graph registry."""
+        return [self._make_single_concept_data_for_registry(concept) for concept in self.concept_dependencies]
+
     @field_validator("code", mode="before")
     @classmethod
     def validate_pipe_code_syntax(cls, code: str) -> str:
@@ -411,13 +442,22 @@ async def run_pipe(
                         )
                         input_specs.append(input_spec)
 
+                # Serialize pipe and concept data for registries if enabled
+                pipe_data: dict[str, Any] | None = None
+                concept_data: list[dict[str, Any]] | None = None
+                if parent_graph_context.data_inclusion.pipe_and_concept_registry:
+                    pipe_data = self.model_dump(mode="json")
+                    concept_data = self._make_concept_data_for_registry()
+
                 graph_node_id, child_graph_context = tracer_manager.on_pipe_start(
                     graph_context=parent_graph_context,
                     pipe_code=self.code,
                     pipe_type=self.type,
                     node_kind=node_kind,
                     started_at=started_at,
                     input_specs=input_specs or None,
+                    pipe_data=pipe_data,
+                    concept_data=concept_data,
                 )
                 # Update job metadata with child graph context for nested pipes
                 if child_graph_context is not None:
@@ -480,11 +520,17 @@ async def run_pipe(
                     data_html=main_stuff.content.rendered_pretty_html() if parent_graph_context.data_inclusion.stuff_html_content else None,
                 )
 
+            # Serialize output concept for registry if enabled
+            output_concept_data: dict[str, Any] | None = None
+            if parent_graph_context.data_inclusion.pipe_and_concept_registry and main_stuff is not None:
+                output_concept_data = self._make_single_concept_data_for_registry(main_stuff.concept)
+
             tracer_manager.on_pipe_end_success(
                 graph_id=parent_graph_context.graph_id,
                 node_id=graph_node_id,
                 ended_at=datetime.now(timezone.utc),
                 output_spec=output_spec,
+                output_concept_data=output_concept_data,
             )
 
         pipe_run_params.pop_pipe_from_stack(pipe_code=self.code)

pipelex/core/pipes/pipe_assignment.py

@@ -0,0 +1,25 @@
+"""Base Assignment class for the pipe execution pre_run/run/post_run pattern.
+
+An Assignment carries resolved runtime data through the execution lifecycle:
+- Created in prepare_assignment() (pre_run phase): models resolved, prompts rendered
+- Consumed in execute() (run phase): pure execution using pre-resolved data
+- Finalized in finalize_assignment() (post_run phase): execution metadata captured
+
+Each pipe subclass defines its own Assignment type with operator-specific fields.
+"""
+
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class PipeAssignment(BaseModel):
+    """Base Assignment for pipe execution.
+
+    Subclasses add operator-specific fields (rendered prompts, resolved models, etc.).
+    The execution_data dict collects metadata for the GraphSpec's execution_data field.
+    """
+
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    execution_data: dict[str, Any] = Field(default_factory=dict)

pipelex/graph/graph_config.py

@@ -10,6 +10,7 @@ class DataInclusionConfig(ConfigModel):
     stuff_text_content: bool
     stuff_html_content: bool
     error_stack_traces: bool
+    pipe_and_concept_registry: bool
 
 
 class GraphsInclusionConfig(ConfigModel):

pipelex/graph/graph_tracer.py

@@ -1,6 +1,7 @@
 """GraphTracer implementation that builds GraphSpec during pipeline execution."""
 
 from datetime import datetime, timezone
+from typing import Any
 
 from typing_extensions import override
 
@@ -48,6 +49,7 @@ def __init__(
         self.error: ErrorSpec | None = None
         self.input_specs: list[IOSpec] = input_specs or []
         self.output_specs: list[IOSpec] = []
+        self.execution_data: dict[str, Any] = {}
 
     def to_node_spec(self) -> NodeSpec:
         """Convert to immutable NodeSpec."""
@@ -76,6 +78,7 @@ def to_node_spec(self) -> NodeSpec:
             node_io=node_io,
             error=self.error,
             metrics=self.metrics,
+            execution_data=self.execution_data,
         )
 
 
@@ -113,6 +116,9 @@ def __init__(self) -> None:
         # The branch_producer_node_id is snapshotted at registration time, before register_controller_output
         # overrides _stuff_producer_map to point branch stuff codes to the controller node
         self._parallel_combine_map: dict[str, tuple[str, list[tuple[str, str]]]] = {}
+        # Registries for pipe and concept data (keyed by pipe_ref and concept_ref)
+        self._pipe_registry: dict[str, dict[str, Any]] = {}
+        self._concept_registry: dict[str, dict[str, Any]] = {}
 
     @property
     def is_active(self) -> bool:
@@ -143,6 +149,8 @@ def setup(
         self._batch_item_map = {}
         self._batch_aggregate_map = {}
         self._parallel_combine_map = {}
+        self._pipe_registry = {}
+        self._concept_registry = {}
 
         return GraphContext(
             graph_id=graph_id,
@@ -181,6 +189,8 @@ def teardown(self) -> GraphSpec | None:
             pipeline_ref=self._pipeline_ref or PipelineRef(),
             nodes=nodes,
             edges=self._edges,
+            pipe_registry=dict(self._pipe_registry),
+            concept_registry=dict(self._concept_registry),
         )
 
         # Reset internal state
@@ -193,6 +203,8 @@ def teardown(self) -> GraphSpec | None:
         self._batch_item_map = {}
         self._batch_aggregate_map = {}
         self._parallel_combine_map = {}
+        self._pipe_registry = {}
+        self._concept_registry = {}
 
         return graph
 
@@ -409,6 +421,8 @@ def on_pipe_start(
         node_kind: NodeKind,
         started_at: datetime,
         input_specs: list[IOSpec] | None = None,
+        pipe_data: dict[str, Any] | None = None,
+        concept_data: list[dict[str, Any]] | None = None,
     ) -> tuple[str, GraphContext]:
         """Record the start of a pipe execution."""
         if not self._is_active:
@@ -433,6 +447,18 @@ def on_pipe_start(
         )
         self._nodes[node_id] = node_data
 
+        # Accumulate pipe and concept registry data (deduplicated)
+        if graph_context.data_inclusion.pipe_and_concept_registry:
+            if pipe_data is not None:
+                pipe_ref = f"{pipe_data.get('domain_code', '')}.{pipe_data.get('code', '')}"
+                if pipe_ref not in self._pipe_registry:
+                    self._pipe_registry[pipe_ref] = pipe_data
+            if concept_data is not None:
+                for concept_item in concept_data:
+                    concept_ref = f"{concept_item.get('domain_code', '')}.{concept_item.get('code', '')}"
+                    if concept_ref not in self._concept_registry:
+                        self._concept_registry[concept_ref] = concept_item
+
         # Add containment edge from parent if this is a child pipe
         if graph_context.parent_node_id is not None:
             self.add_edge(
@@ -449,6 +475,20 @@ def on_pipe_start(
 
         return node_id, child_context
 
+    @override
+    def register_execution_data(
+        self,
+        node_id: str,
+        execution_data: dict[str, Any],
+    ) -> None:
+        """Register execution metadata for a node."""
+        if not self._is_active:
+            return
+        node_data = self._nodes.get(node_id)
+        if node_data is None:
+            return
+        node_data.execution_data.update(execution_data)
+
     @override
     def on_pipe_end_success(
         self,
@@ -457,6 +497,7 @@ def on_pipe_end_success(
         output_preview: str | None = None,
         metrics: dict[str, float] | None = None,
         output_spec: IOSpec | None = None,
+        output_concept_data: dict[str, Any] | None = None,
     ) -> None:
         """Record successful completion of a pipe execution."""
         if not self._is_active:
@@ -472,6 +513,12 @@ def on_pipe_end_success(
         if metrics:
             node_data.metrics = metrics
 
+        # Accumulate output concept data (deduplicated)
+        if output_concept_data is not None:
+            concept_ref = f"{output_concept_data.get('domain_code', '')}.{output_concept_data.get('code', '')}"
+            if concept_ref not in self._concept_registry:
+                self._concept_registry[concept_ref] = output_concept_data
+
         # Store output spec and register in producer map for data flow tracking
         if output_spec is not None:
             # Skip pass-through outputs: if the output digest matches one of the node's

pipelex/graph/graph_tracer_manager.py

@@ -1,6 +1,7 @@
 """Graph tracer manager with singleton pattern for access from PipeAbstract without hub imports."""
 
 from datetime import datetime
+from typing import Any
 
 from pipelex.graph.graph_config import DataInclusionConfig
 from pipelex.graph.graph_context import GraphContext
@@ -175,6 +176,8 @@ def on_pipe_start(
         node_kind: NodeKind,
         started_at: datetime,
         input_specs: list[IOSpec] | None = None,
+        pipe_data: dict[str, Any] | None = None,
+        concept_data: list[dict[str, Any]] | None = None,
     ) -> tuple[str | None, GraphContext | None]:
         """Record the start of a pipe execution.
 
@@ -185,6 +188,8 @@ def on_pipe_start(
             node_kind: The kind of node (controller, operator, etc.).
             started_at: When the pipe started executing.
             input_specs: Optional list of IOSpec describing the inputs consumed.
+            pipe_data: Optional serialized pipe instance for the pipe registry.
+            concept_data: Optional list of serialized concept dicts for the concept registry.
 
         Returns:
             Tuple of (node_id, child_graph_context) if tracing is active, (None, None) otherwise.
@@ -200,6 +205,8 @@ def on_pipe_start(
             node_kind=node_kind,
             started_at=started_at,
             input_specs=input_specs,
+            pipe_data=pipe_data,
+            concept_data=concept_data,
         )
 
     def on_pipe_end_success(
@@ -210,6 +217,7 @@ def on_pipe_end_success(
         output_preview: str | None = None,
         metrics: dict[str, float] | None = None,
         output_spec: IOSpec | None = None,
+        output_concept_data: dict[str, Any] | None = None,
     ) -> None:
         """Record successful completion of a pipe execution.
 
@@ -220,6 +228,7 @@ def on_pipe_end_success(
             output_preview: Optional truncated preview of the output.
             metrics: Optional metrics (e.g., token counts).
             output_spec: Optional IOSpec describing the output produced.
+            output_concept_data: Optional serialized concept dict for the actual output concept.
         """
         if node_id is None:
             return
@@ -234,8 +243,29 @@ def on_pipe_end_success(
             output_preview=output_preview,
             metrics=metrics,
             output_spec=output_spec,
+            output_concept_data=output_concept_data,
         )
 
+    def register_execution_data(
+        self,
+        graph_id: str,
+        node_id: str | None,
+        execution_data: dict[str, Any],
+    ) -> None:
+        """Register execution metadata for a node.
+
+        Args:
+            graph_id: The graph identifier.
+            node_id: The node ID to attach execution data to.
+            execution_data: Dictionary of execution metadata.
+        """
+        if node_id is None:
+            return
+        tracer = self._get_tracer(graph_id)
+        if tracer is None:
+            return
+        tracer.register_execution_data(node_id=node_id, execution_data=execution_data)
+
     def on_pipe_end_error(
         self,
         graph_id: str,