Merge pull request Pipelex#79 from Pipelex/release/v0.3.2

Release/v0.3.2

Author: thomashebrard

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## [v0.3.2] - 2025-06-13
+
+- Improved automatic insertion of class structure from BaseModel into prompts, based on the PipeLLM's `output_concept`. New unit test included.
+- The ReportingManager now reports costs for all pipeline IDs when no `pipeline_run_id` is specified.
+- The `make_from_str` method from the `StuffFactory` class now uses `Text` context by default.
+
 ## [v0.3.1] - 2025-06-10
 
 ### Added

README.md

@@ -1,8 +1,8 @@
 <div align="center">
   <a href="https://www.pipelex.com/"><img src="https://raw.githubusercontent.com/Pipelex/pipelex/main/.github/assets/logo.png" alt="Pipelex Logo" width="400" style="max-width: 100%; height: auto;"></a>
 
-  <h3 align="center">The simpler way to build reliable LLM Pipelines</h3>
-  <p align="center">Pipelex is an open‑source dev tool based on a simple declarative language<br/>that lets you define replicable, structured, composable LLM pipelines.</p>
+  <h2 align="center">Lean-code language for repeatable workflows</h2>
+  <p align="center">Pipelex is based on a simple declarative language that lets you define repeatable, structured, composable AI workflows.</p>
 
   <div>
     <a href="https://www.pipelex.com/demo"><strong>Demo</strong></a> -
@@ -19,14 +19,24 @@
     <br/>
     <br/>
     <a href="https://www.youtube.com/@PipelexAI"><img src="https://img.shields.io/badge/YouTube-FF0000?logo=youtube&logoColor=white" alt="YouTube"></a>
-    <a href="https://pipelex.com"><img src="https://img.shields.io/badge/Web-pipelex.com-03bb95?logo=google-chrome&logoColor=white&style=flat" alt="Website"></a>
+    <a href="https://pipelex.com"><img src="https://img.shields.io/badge/Homepage-03bb95?logo=google-chrome&logoColor=white&style=flat" alt="Website"></a>
+    <a href="https://github.com/Pipelex/pipelex-cookbook"><img src="https://img.shields.io/badge/Cookbook-03bb95?logo=github&logoColor=white&style=flat" alt="Cookbook"></a>
     <a href="https://discord.gg/SReshKQjWt"><img src="https://img.shields.io/badge/Discord-5865F2?logo=discord&logoColor=white" alt="Discord"></a>
     <br/> 
     <br/>
 </div>
 
 <div align="center">
-  <a href="https://www.pipelex.com/demo"><strong>Checkout our demo!</strong></a>
+  <h2 align="center">📜 The Knowledge Pipeline Manifesto</h2>
+  <p align="center">
+    <a href="https://www.pipelex.com/post/the-knowledge-pipeline-manifesto"><strong>Read why we built Pipelex to transform unreliable AI workflows into deterministic pipelines 🔗</strong></a>
+  </p>
+
+  <h2 align="center">🚀 See Pipelex in Action</h2>
+  <p align="center">
+    <a href="https://www.pipelex.com/demo"><strong>Checkout our Demo</strong></a>
+  </p>
+  
 </div>
 
 # 📑 Table of Contents
@@ -42,15 +52,32 @@
 
 # Introduction
 
-Pipelex™ is a developer tool designed to simplify building reliable AI applications. At its core is a clear, declarative pipeline language specifically crafted for knowledge-processing tasks.
+Pipelex makes it easy for developers to define and run repeatable AI workflows. At its core is a clear, declarative pipeline language specifically crafted for knowledge-processing tasks.
 
-**The Pipelex language uses pipelines,** or "pipes", each capable of integrating different language models (LLMs) or software to process knowledge. Pipes consistently deliver **structured, predictable outputs** at each stage.
+Build **pipelines** from modular pipes that snap together. Each pipe can use a different language model (LLM) or software to process knowledge. Pipes consistently deliver **structured, predictable outputs** at each stage.
 
-Pipelex employs user-friendly TOML syntax, enabling developers to intuitively define workflows in a narrative-like manner. This approach facilitates collaboration between business professionals, developers, and language models (LLMs), ensuring clarity and ease of communication.
+Pipelex uses TOML syntax, making workflows readable and shareable. Business professionals, developers, and AI coding agents can all understand and modify the same pipeline definitions.
+
+Example:
+```toml
+[concept]
+Buyer = "The person who made the purchase"
+PurchaseDocumentText = "Transcript of a receipt, invoice, or order confirmation"
+
+[pipe.extract_buyer]
+PipeLLM = "Extract buyer from purchase document"
+inputs = { purchase_document_text = "PurchaseDocumentText" }
+output = "Buyer"
+llm = "llm_to_extract_info"
+prompt_template = """
+Extract the first and last name of the buyer from this purchase document:
+@purchase_document_text
+"""
+```
 
-Pipes function like modular building blocks, **assembled by connecting other pipes sequentially, in parallel, or by calling sub-pipes.** This assembly resembles function calls in traditional programming but emphasizes a more intuitive, plug-and-play structure, focused explicitly on clear knowledge input and output.
+Pipes are modular building blocks that **connect sequentially, run in parallel, or call sub-pipes.** Like function calls in traditional programming, but with a clear contract: knowledge-in, knowledge-out. This modularity makes pipelines perfect for sharing: fork someone's invoice processor, adapt it for receipts, share it back. 
 
-Pipelex is distributed as an **open-source Python library,** with a hosted API launching soon, enabling effortless integration into existing software systems and automation frameworks. Additionally, Pipelex will provide an MCP server that will enable AI Agents to run pipelines like any other tool.
+Pipelex is an **open-source Python library** with a hosted API launching soon. It integrates seamlessly into existing systems and automation frameworks. Plus, it works as an [MCP server](https://github.com/Pipelex/pipelex-mcp) so AI agents can use pipelines as tools.
 
 # 🚀 Quick start
 

pipelex/cogt/llm/llm_worker_abstract.py

@@ -131,7 +131,9 @@ async def gen_object(
             result = await self._gen_object(llm_job=llm_job, schema=schema)
         except InstructorRetryException as exc:
             raise LLMCompletionError(
-                f"LLM Worker error: Instructor failed after retry with llm '{self.llm_engine.tag}': {exc}\nLLMPrompt: {llm_job.llm_prompt.desc}"
+                f"""Instructor failed to generate object: {schema} after retry with llm '{self.llm_engine.tag}'
+                Reason: {exc}
+                LLMPrompt: {llm_job.llm_prompt.desc}"""
             ) from exc
 
         # Cleanup result

pipelex/core/stuff.py

@@ -136,7 +136,7 @@ def as_list_of_fixed_content_type(self, item_type: Type[StuffContentType]) -> Li
         # Validate all items are of the expected type
         for i, item in enumerate(list_content.items):
             if not isinstance(item, item_type):
-                raise TypeError(f"Item {i} in list is of type {type(item)}, not {item_type}")
+                raise TypeError(f"Item {i} in list is of type {type(item)}, not {item_type}, in {self.stuff_name=} and {self.concept_code=}")
 
         return list_content
 

pipelex/core/stuff_factory.py

@@ -95,9 +95,9 @@ def make_from_blueprint(cls, blueprint: StuffBlueprint) -> "Stuff":
     @classmethod
     def make_from_str(
         cls,
-        concept_code: str,
         str_value: str,
         name: Optional[str] = None,
+        concept_code: str = NativeConcept.TEXT.code,
         pipelex_session_id: Optional[str] = None,
     ) -> Stuff:
         if not Concept.concept_str_contains_domain(concept_code):

pipelex/pipe_operators/pipe_llm.py

@@ -219,11 +219,14 @@ async def _run_operator_pipe(
             # TODO: This DYNAMIC_OUTPUT_CONCEPT should not be a field in the params attribute of PipeRunParams.
             # It should be an attribute of PipeRunParams.
             output_concept_code = pipe_run_params.dynamic_output_concept_code or pipe_run_params.params.get(PipeRunParamKey.DYNAMIC_OUTPUT_CONCEPT)
+
             if not output_concept_code:
                 raise RuntimeError(f"No output concept code provided for dynamic output pipe '{self.code}'")
         else:
             output_concept_code = self.output_concept_code
 
+        self.pipe_llm_prompt.output_concept_code = output_concept_code
+
         applied_output_multiplicity, is_multiple_output, fixed_nb_output = output_multiplicity_to_apply(
             output_multiplicity_base=self.output_multiplicity,
             output_multiplicity_override=pipe_run_params.output_multiplicity,
@@ -318,6 +321,7 @@ async def _run_operator_pipe(
                             domain=self.domain,
                             user_pipe_jinja2=user_pipe_jinja2,
                             system_prompt=system_prompt,
+                            output_concept_code=output_concept_code,
                         )
                         llm_prompt_2_factory = PipedLLMPromptFactory(
                             pipe_llm_prompt=pipe_llm_prompt_2,
@@ -340,6 +344,7 @@ async def _run_operator_pipe(
                     domain=self.domain,
                     user_pipe_jinja2=user_pipe_jinja2,
                     system_prompt=system_prompt,
+                    output_concept_code=output_concept_code,
                 )
                 llm_prompt_2_factory = PipedLLMPromptFactory(
                     pipe_llm_prompt=pipe_llm_prompt_2,

pipelex/pipe_operators/pipe_llm_prompt.py

@@ -174,6 +174,10 @@ async def _run_operator_pipe(
         # Append output structure prompt if needed
         if pipe_run_params.dynamic_output_concept_code:
             user_text += PipeLLMPrompt.get_output_structure_prompt(output_concept=pipe_run_params.dynamic_output_concept_code)
+        else:
+            user_text += PipeLLMPrompt.get_output_structure_prompt(output_concept=self.output_concept_code)
+
+        log.verbose(f"User text with {self.output_concept_code=}:\n {user_text}")
 
         ############################################################
         # System text
@@ -219,16 +223,18 @@ def get_output_structure_prompt(output_concept: str) -> str:
         if not output_class:
             return ""
 
-        fields = get_type_structure(output_class, base_class=StuffContent)
+        class_structure = get_type_structure(output_class, base_class=StuffContent)
 
-        if not fields:
+        if not class_structure:
             return ""
 
         output_structure_prompt = (
-            f"\n\n---\nRequested output format: The output should contain the following fields:\n"
-            f"{chr(10).join(fields)}\n"
+            f"\n\n---\nRequested output format: The output should be the following class: {class_name}\n"
+            f"{chr(10).join(class_structure)}\n"
             "You do NOT need to output a formatted JSON object, another LLM will take care of that. "
+            "If you cannot find a value that is Optional, output None for that field."
             "However, you MUST clearly output the values for each of these fields in your response.\n---\n"
+            "DO NOT create information. If the information is not present, output None."
         )
         return output_structure_prompt
 

pipelex/reporting/reporting_manager.py

@@ -94,7 +94,6 @@ def report_inference_job(self, inference_job: InferenceJobAbstract):
 
     @override
     def generate_report(self, pipeline_run_id: Optional[str] = None):
-        pipeline_run_id = pipeline_run_id or SpecialPipelineId.UNTITLED
         cost_report_file_path: Optional[str] = None
         if self._reporting_config.is_generate_cost_report_file_enabled:
             ensure_path(self._reporting_config.cost_report_dir_path)
@@ -104,13 +103,19 @@ def generate_report(self, pipeline_run_id: Optional[str] = None):
                 extension=self._reporting_config.cost_report_extension,
             )
 
-        registry = self._get_registry(pipeline_run_id)
-        CostRegistry.generate_report(
-            pipeline_run_id=pipeline_run_id,
-            llm_tokens_usages=registry.get_current_tokens_usage(),
-            unit_scale=self._reporting_config.cost_report_unit_scale,
-            cost_report_file_path=cost_report_file_path,
-        )
+        registries_to_process: Dict[str, UsageRegistry] = {}
+        if pipeline_run_id:
+            registries_to_process = {pipeline_run_id: self._get_registry(pipeline_run_id)}
+        else:
+            registries_to_process = self._usage_registries
+
+        for run_id, registry in registries_to_process.items():
+            CostRegistry.generate_report(
+                pipeline_run_id=run_id,
+                llm_tokens_usages=registry.get_current_tokens_usage(),
+                unit_scale=self._reporting_config.cost_report_unit_scale,
+                cost_report_file_path=cost_report_file_path,
+            )
 
     @override
     def close_registry(self, pipeline_run_id: str):