science_live.pipeline.pipeline

science_live.pipeline.pipeline#

Science Live Pipeline: Main Pipeline Orchestrator#

Main pipeline class that orchestrates all 7 processing steps to convert natural language questions into structured scientific knowledge exploration.

This module provides the high-level ScienceLivePipeline class that users interact with, coordinating all pipeline steps from question processing to natural language response generation.

Module Contents#

Classes#

ScienceLivePipeline

Main pipeline that orchestrates all 7 processing steps.

Functions#

create_custom_pipeline

Create a custom pipeline with specific steps.

quick_process

Quick processing function for simple use cases.

Data#

__version__

__author__

__description__

API#

class science_live.pipeline.pipeline.ScienceLivePipeline(endpoint_manager, config: Dict[str, Any] = None)[source]#

Main pipeline that orchestrates all 7 processing steps.

This class provides a high-level interface for processing natural language questions through the complete pipeline from question processing to natural language response generation.

Pipeline Steps:

  1. QuestionProcessor - Parse and classify questions

  2. EntityExtractorLinker - Extract and link entities

  3. RosettaStatementGenerator - Generate Rosetta statements

  4. SPARQLGenerator - Convert to SPARQL queries

  5. QueryExecutor - Execute queries against nanopub endpoints

  6. ResultProcessor - Structure and group results

  7. NaturalLanguageGenerator - Generate natural language responses

Example: from science_live.core import EndpointManager from science_live.pipeline import ScienceLivePipeline

endpoint_manager = EndpointManager()
pipeline = ScienceLivePipeline(endpoint_manager)

result = await pipeline.process("What papers cite AlexNet?")
print(result.summary)

Initialization

Initialize the Science Live pipeline.

Args: endpoint_manager: EndpointManager instance for nanopub connections config: Optional configuration dictionary for pipeline components

async process(question: str, user_id: Optional[str] = None, **kwargs) → science_live.pipeline.common.NaturalLanguageResult[source]#

Process a natural language question through the complete pipeline.

Args: question: Natural language question to process user_id: Optional user identifier for personalization **kwargs: Additional context parameters (session_id, preferences, etc.)

Returns: NaturalLanguageResult containing summary, detailed results, confidence explanation, and suggestions

Raises: ValueError: If question is invalid or context is malformed Exception: For pipeline processing errors (handled gracefully)

async process_batch(questions: List[str], user_id: Optional[str] = None, **kwargs) → List[science_live.pipeline.common.NaturalLanguageResult][source]#

Process multiple questions in batch mode.

Args: questions: List of natural language questions user_id: Optional user identifier **kwargs: Additional context parameters

Returns: List of NaturalLanguageResult objects, one per question

Note: Questions are processed concurrently for better performance. Failed questions return error responses rather than exceptions.

get_pipeline_info() → Dict[str, Any][source]#

Get information about the pipeline configuration and status.

Returns: Dictionary containing pipeline information

async health_check() → Dict[str, Any][source]#

Perform health check on pipeline components.

Returns: Dictionary containing health status of each component

science_live.pipeline.pipeline.create_custom_pipeline(steps: List[Any]) → science_live.pipeline.pipeline.create_custom_pipeline.CustomPipeline[source]#

Create a custom pipeline with specific steps.

This function allows creating specialized pipelines by selecting and ordering specific processing steps.

Args: steps: List of pipeline step instances in desired order

Returns: CustomPipeline instance that processes data through the given steps

Example: from science_live.pipeline import ( QuestionProcessor, EntityExtractorLinker, create_custom_pipeline )

custom_steps = [
    QuestionProcessor(),
    EntityExtractorLinker(endpoint_manager)
]
custom_pipeline = create_custom_pipeline(custom_steps)

result = await custom_pipeline.process("What is DNA?")
async science_live.pipeline.pipeline.quick_process(question: str, endpoint_manager=None) → science_live.pipeline.common.NaturalLanguageResult[source]#

Quick processing function for simple use cases.

Args: question: Natural language question to process endpoint_manager: Optional endpoint manager (creates test endpoint if None)

Returns: NaturalLanguageResult with processed response

Example: from science_live.pipeline.pipeline import quick_process

result = await quick_process("What papers cite AlexNet?")
print(result.summary)
science_live.pipeline.pipeline.__version__ = '1.0.0'#
science_live.pipeline.pipeline.__author__ = 'Science Live Team'#
science_live.pipeline.pipeline.__description__ = 'Main pipeline orchestrator for Science Live'#
Contents