Skip to main content

GenericWorkflow

What it is​

A generic workflow wrapper that:

  • Renders a SPARQL query from a Jinja2 template and validated Pydantic parameters
  • Executes the query via a TripleStoreService
  • Converts results to a Python list using SPARQLUtils
  • Exposes the workflow as a LangChain StructuredTool

Public API​

  • class GenericWorkflow(Generic[T])
    • __init__(name, description, sparql_template, arguments_model, triple_store_service)
      • Stores workflow metadata, a SPARQL Jinja2 template, the Pydantic arguments model, and the triple store service.
    • run(parameters: T)
      • Renders sparql_template with parameters.model_dump(), executes the SPARQL query, and returns results as a list.
      • On any exception, returns [{ "error": "<message>" }].
    • as_tools() -> list[BaseTool]
      • Returns a list containing a single StructuredTool that:
        • Validates kwargs using arguments_model
        • Calls run() with the constructed model instance

Configuration/Dependencies​

  • Pydantic: arguments_model must be a subclass of pydantic.BaseModel.
  • Jinja2: Imported inside run() (from jinja2 import Template) to render the SPARQL template.
  • Triple store: Requires an instance of naas_abi_core.services.triple_store.TripleStoreService.TripleStoreService implementing query(sparql_query).
  • Result conversion: Uses naas_abi_core.utils.SPARQL.SPARQLUtils(...).results_to_list(results).
  • LangChain tools: Uses langchain_core.tools.StructuredTool / BaseTool.

Usage​

from pydantic import BaseModel
from naas_abi_core.modules.templatablesparqlquery.workflows.GenericWorkflow import GenericWorkflow

# Define the validated inputs for the template
class Args(BaseModel):
    limit: int = 10

sparql_template = """
SELECT ?s ?p ?o WHERE {
  ?s ?p ?o .
}
LIMIT {{ limit }}
"""

# triple_store_service must be provided by your environment
workflow = GenericWorkflow(
    name="example_sparql_query",
    description="Runs a templated SPARQL query",
    sparql_template=sparql_template,
    arguments_model=Args,
    triple_store_service=triple_store_service,
)

result = workflow.run(Args(limit=5))
tools = workflow.as_tools()

Caveats​

  • run() prints the rendered SPARQL query to stdout (print(sparql_query)).
  • Errors are swallowed and returned as a list with an "error" dict, which may mask exceptions.
  • The SPARQL template is rendered with Jinja2 using parameters.model_dump(); template variables must match model field names.