Skip to main content

{{pipeline_name_pascal}}Pipeline

What it is​

  • A template pipeline module defining:
    • A configuration dataclass ({{pipeline_name_pascal}}PipelineConfiguration)
    • A parameters model ({{pipeline_name_pascal}}PipelineParameters)
    • A pipeline implementation ({{pipeline_name_pascal}}Pipeline) intended to produce an rdflib.Graph
  • The core run() logic is not implemented (pass).

Public API​

  • @dataclass {{pipeline_name_pascal}}PipelineConfiguration(PipelineConfiguration)

    • Purpose: Pipeline configuration container (currently empty; extend as needed).

  • class {{pipeline_name_pascal}}PipelineParameters(PipelineParameters)

    • Purpose: Input schema for the pipeline (currently empty; example field commented out).

  • class {{pipeline_name_pascal}}Pipeline(Pipeline[{{pipeline_name_pascal}}PipelineParameters])

    • __init__(configuration: {{pipeline_name_pascal}}PipelineConfiguration)
      • Purpose: Initialize pipeline with configuration.
    • run(parameters: {{pipeline_name_pascal}}PipelineParameters) -> rdflib.Graph
      • Purpose: Execute pipeline and return an RDFLib Graph.
      • Status: Not implemented.
    • as_tools() -> list[langchain_core.tools.BaseTool]
      • Purpose: Expose the pipeline as a LangChain StructuredTool named {{pipeline_name_pascal}}.
    • as_api(router: naas_abi_core.utils.Expose.APIRouter, ...) -> None
      • Purpose: API exposure hook.
      • Status: No routes are registered; returns None.

Configuration/Dependencies​

  • Depends on:
    • naas_abi_core.pipeline (Pipeline, PipelineConfiguration, PipelineParameters)
    • langchain_core.tools (BaseTool, StructuredTool)
    • rdflib (Graph, plus various RDF/OWL namespace utilities imported but unused in template)
    • pydantic.Field (imported; example usage commented out)
  • Notes:
    • Several imports (uuid, Optional, logger, RDF namespace symbols) are present but not used in this template.

Usage​

from rdflib import Graph

# Instantiate configuration and pipeline
cfg = {{pipeline_name_pascal}}PipelineConfiguration()
pipeline = {{pipeline_name_pascal}}Pipeline(cfg)

# Run (will do nothing until implemented)
params = {{pipeline_name_pascal}}PipelineParameters()
result: Graph = pipeline.run(params)  # currently not implemented

# Expose as a LangChain tool
tools = pipeline.as_tools()

Caveats​

  • run() is a stub and will not produce a Graph until implemented.
  • as_api() does not register any endpoints; API exposure must be added manually.