Skip to content

generation

Composable generation APIs for Rago.

Modules:

  • base –

    Base classes for generation steps.

  • cohere –

    CohereGen class for text generation using Cohere's API.

  • deepseek –

    DeepSeek generation module.

  • fireworks –

    FireworksGen class for text generation using Fireworks API.

  • gemini –

    GeminiGen class for text generation using Google's Gemini model.

  • groq –

    Groq class for text generation.

  • hugging_face –

    Hugging Face classes for text generation.

  • hugging_face_inf –

    Hugging Face inferencing classes for text generation.

  • llama –

    Llama generation module.

  • openai –

    OpenAI Generation Model class for flexible GPT-based text generation.

  • phi –

    Phi generation module.

  • together –

    TogetherGen class for text generation using Together AI's API.

Classes:

Generation 

Generation(
    api_key: str = '',
    model_name: str = '',
    backend: str = '',
    engine: str = '',
    temperature: float = 0.0,
    prompt_template: str = '',
    output_max_length: int = 500,
    structured_output: Type[BaseModel] | None = None,
    api_params: dict[str, Any] | None = None,
    device: str = 'auto',
    system_message: str = '',
    cache: Any = None,
    logs: dict[str, Any] | None = None,
)

Bases: StepBase

Public generation wrapper that resolves a concrete backend lazily.

Methods:

  • apply –

    Apply declarative configuration to the generation wrapper.

  • generate –

    Resolve the concrete generator and run generation.

  • process –

    Process the current pipeline content with generation.

Source code in src/rago/generation/__init__.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def __init__(
    self,
    api_key: str = '',
    model_name: str = '',
    backend: str = '',
    engine: str = '',
    temperature: float = 0.0,
    prompt_template: str = '',
    output_max_length: int = 500,
    structured_output: Type[BaseModel] | None = None,
    api_params: dict[str, Any] | None = None,
    device: str = 'auto',
    system_message: str = '',
    cache: Any = None,
    logs: dict[str, Any] | None = None,
) -> None:
    super().__init__()
    self.backend = backend.lower() if backend else ''
    self.engine = engine.lower() if engine else ''
    self.params = GenerationParameters(
        api_key=api_key,
        model_name=model_name,
        temperature=temperature,
        prompt_template=prompt_template,
        output_max_length=output_max_length,
        structured_output=structured_output,
        api_params=api_params or {},
        device=device,
        system_message=system_message,
    )
    self.cache = cache
    self.logs = logs if logs is not None else {}

apply 

apply(parameters: Any) -> None

Apply declarative configuration to the generation wrapper.

Source code in src/rago/generation/__init__.py 
84
85
86
87
88
89
90
91
92
93
def apply(self, parameters: Any) -> None:
    """Apply declarative configuration to the generation wrapper."""
    super().apply(parameters)
    for key, value in config_to_dict(parameters).items():
        if key == 'backend' and isinstance(value, str):
            self.backend = value.lower()
        elif key == 'engine' and isinstance(value, str):
            self.engine = value.lower()
        else:
            self.params.params[key] = value

generate 

generate(query: str, data: list[str]) -> str | BaseModel

Resolve the concrete generator and run generation.

Source code in src/rago/generation/__init__.py 
165
166
167
168
169
def generate(self, query: str, data: list[str]) -> str | BaseModel:
    """Resolve the concrete generator and run generation."""
    generator_instance = self._resolve()
    normalized_data = [str(item) for item in ensure_list(data)]
    return generator_instance.generate(query, normalized_data)

process 

process(inp: Input) -> Output

Process the current pipeline content with generation.

Source code in src/rago/generation/__init__.py 
171
172
173
174
def process(self, inp: Input) -> Output:
    """Process the current pipeline content with generation."""
    generator_instance = self._resolve()
    return generator_instance.process(inp)

GenerationBase 

GenerationBase(
    model_name: Optional[str] = None,
    temperature: Optional[float] = None,
    prompt_template: str = '',
    output_max_length: int = 500,
    device: str = 'auto',
    structured_output: Optional[Type[BaseModel]] = None,
    system_message: str = '',
    api_params: dict[str, Any] = DEFAULT_API_PARAMS,
    api_key: str = '',
    cache: Cache | None = None,
    logs: dict[str, Any] | None = None,
)

Bases: StepBase

Generic generation step.

Methods:

  • apply –

    Apply attached configuration to the step.

  • generate –

    Generate text with optional language parameter.

  • process –

    Generate a result from the current pipeline content.

Source code in src/rago/generation/base.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def __init__(
    self,
    model_name: Optional[str] = None,
    temperature: Optional[float] = None,
    prompt_template: str = '',
    output_max_length: int = 500,
    device: str = 'auto',
    structured_output: Optional[Type[BaseModel]] = None,
    system_message: str = '',
    api_params: dict[str, Any] = DEFAULT_API_PARAMS,
    api_key: str = '',
    cache: Cache | None = None,
    logs: dict[str, Any] | None = None,
) -> None:
    super().__init__()
    self.api_key = api_key
    self.cache = cache
    self.logs = logs if logs is not None else {}

    self.model_name = (
        model_name if model_name is not None else self.default_model_name
    )
    self.output_max_length = (
        output_max_length or self.default_output_max_length
    )
    self.temperature = (
        temperature
        if temperature is not None
        else self.default_temperature
    )
    self.prompt_template = prompt_template or self.default_prompt_template
    self.structured_output = structured_output
    if api_params is DEFAULT_API_PARAMS:
        api_params = deepcopy(self.default_api_params or {})

    self.system_message = system_message
    self.api_params = api_params

    if device not in ['cpu', 'cuda', 'auto']:
        raise Exception(
            f'Device {device} not supported. Options: cpu, cuda, auto.'
        )

    cuda_available = torch.cuda.is_available()
    self.device_name = (
        'cpu' if device == 'cpu' or not cuda_available else 'cuda'
    )
    self.device = torch.device(self.device_name)

    self._validate()
    self._load_optional_modules()
    self._setup()

apply 

apply(parameters: Any) -> None

Apply attached configuration to the step.

Source code in src/rago/base.py 
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
def apply(self, parameters: Any) -> None:
    """Apply attached configuration to the step."""
    if parameters is None:
        return

    if _is_cache_backend(parameters):
        self.cache = parameters
        return

    if _is_vector_db(parameters):
        setattr(self, 'db', parameters)
        return

    if _is_text_splitter(parameters):
        setattr(self, 'splitter', parameters)
        return

    for key, value in config_to_dict(parameters).items():
        if key == 'cache':
            self.cache = value
        elif key == 'logs':
            self.logs = value if value is not None else {}
        else:
            setattr(self, key, value)

generate abstractmethod 

generate(query: str, data: list[str]) -> str | BaseModel

Generate text with optional language parameter.

Source code in src/rago/generation/base.py 
169
170
171
172
173
174
175
@abstractmethod
def generate(
    self,
    query: str,
    data: list[str],
) -> str | BaseModel:
    """Generate text with optional language parameter."""

process 

process(inp: Input) -> Output

Generate a result from the current pipeline content.

Source code in src/rago/generation/base.py 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
def process(self, inp: Input) -> Output:
    """Generate a result from the current pipeline content."""
    query = str(inp.query)
    data = [
        str(item)
        for item in ensure_list(
            inp.get('content', inp.get('data', inp.get('source')))
        )
    ]
    result = self.generate(query, data)
    output = Output.from_input(inp)
    output.result = result
    output.content = _serialize_generation_result(result)
    output.data = output.content
    return output

GenerationParameters 

GenerationParameters(**kwargs: Any)

Bases: ParametersBase

Parameters for configuring generation steps.

Methods:

  • apply –

    Merge additional configuration into this object.

  • process –

    Return the input unchanged for configuration-only objects.

Attributes:

  • params (dict[str, Any]) –

    Expose the underlying parameter mapping.

Source code in src/rago/base.py 
62
63
def __init__(self, **kwargs: Any) -> None:
    super().__init__(kwargs)

params property 

params: dict[str, Any]

Expose the underlying parameter mapping.

apply 

apply(parameters: Any) -> None

Merge additional configuration into this object.

Source code in src/rago/base.py 
86
87
88
def apply(self, parameters: Any) -> None:
    """Merge additional configuration into this object."""
    self.data.update(config_to_dict(parameters))

process 

process(inp: Input) -> Output

Return the input unchanged for configuration-only objects.

Source code in src/rago/base.py 
90
91
92
def process(self, inp: Input) -> Output:
    """Return the input unchanged for configuration-only objects."""
    return inp.to_output()