Skip to content

verbatim_core API Reference

Transformer

verbatim_core.transformer.Transformer

Turn a question and chunks into a cited answer, built from spans of the chunks.

The first spans fill the template of the answer, and every span gets one citation that points at it by chunk id and offsets. One transformer can serve concurrent calls: the settings of a call come in its config, and nothing is set on the transformer between calls.

Source code in packages/core/verbatim_core/transformer.py 
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
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
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
class Transformer:
    """Turn a question and chunks into a cited answer, built from spans of the chunks.

    The first spans fill the template of the answer, and every span gets one citation that points
    at it by chunk id and offsets. One transformer can serve concurrent calls: the settings of a
    call come in its config, and nothing is set on the transformer between calls.
    """

    def __init__(
        self,
        extractor: SpanExtractor[Any],
        templates: TemplateManager,
        *,
        config: TransformConfig | None = None,
    ) -> None:
        """Build a transformer from an extractor and the templates of the answers.

        :param extractor: Extracts the spans of the chunks
        :param templates: The templates the answers are written with
        :param config: The settings of a call, unless the call passes its own. Without it, the
            templates' settings are the default.
        :raises ConfigurationError: If the config is not a TransformConfig, or asks for structured
            template mode
        """
        self.extractor = extractor
        self.templates = templates
        self._config = None if config is None else _default_config(config, TransformConfig)
        # Templates in structured mode raise at the call: VerbatimRAG answers structured
        # questions itself with the same templates (ADR 0004).
        if self._config is not None and self._config.template_mode == "structured":
            raise ConfigurationError(_STRUCTURED_MODE_UNSUPPORTED, code="invalid_value")
        if templates.strategy(self.config.template_mode) is None:
            warn_contextual_without_client()

    @property
    def config(self) -> TransformConfig:
        """The settings of a call that passes none: the config given, else the templates'."""
        return self._config if self._config is not None else self.templates.config

    def transform(self, question: str, chunks: Sequence[Chunk], *, config: TransformConfig | None = None) -> Answer:
        """Write a cited answer to the question from the chunks.

        :param question: The question
        :param chunks: Chunks with unique ids
        :param config: Replaces the transformer's default config for this call
        :return: The answer, with its citations, spans, usage and warnings
        :raises InvalidInputError: If two chunks share an id, or the config is of the wrong type
            or asks for structured template mode
        :raises UpstreamError: If a service the extractor calls fails
        :raises ConfigurationError: If a service rejects its key, model or quota
        """
        config, strategy = self._call_settings(config)
        *_, answer_event = self._events(question, chunks, config, strategy)
        # The last event is always the answer.
        return cast("AnswerEvent", answer_event).answer

    async def atransform(
        self, question: str, chunks: Sequence[Chunk], *, config: TransformConfig | None = None
    ) -> Answer:
        """Async version of `transform`."""
        config, strategy = self._call_settings(config)
        events = [event async for event in self._aevents(question, chunks, config, strategy)]
        return cast("AnswerEvent", events[-1]).answer

    def stream(
        self, question: str, chunks: Sequence[Chunk], *, config: TransformConfig | None = None
    ) -> Iterator[TransformEvent]:
        """Write the answer of `transform` as events: the spans, then the answer.

        :raises InvalidInputError: When called, if the config is of the wrong type or asks for
            structured template mode. The other errors of `transform` come from the iteration.
        """
        config, strategy = self._call_settings(config)
        return self._events(question, chunks, config, strategy)

    def astream(
        self, question: str, chunks: Sequence[Chunk], *, config: TransformConfig | None = None
    ) -> AsyncIterator[TransformEvent]:
        """Async version of `stream`."""
        config, strategy = self._call_settings(config)
        return self._aevents(question, chunks, config, strategy)

    def _call_settings(self, config: TransformConfig | None) -> tuple[TransformConfig, TemplateStrategy]:
        """Return the config and the template strategy of one call."""
        call_config = _call_config(config, self.config, TransformConfig)
        if call_config.template_mode == "structured":
            if config is None:
                # The templates' settings, switched to structured mode after construction
                raise ConfigurationError(_STRUCTURED_MODE_UNSUPPORTED, code="invalid_value")
            raise InvalidInputError(_STRUCTURED_MODE_UNSUPPORTED, code="invalid_value", param="config.template_mode")

        strategy = self.templates.strategy(call_config.template_mode)
        if strategy is None:
            # Only contextual mode needs a client. Until the deprecation window ends, it falls
            # back to static templates.
            warn_contextual_without_client()
            # Static mode needs no client, so it always has a strategy.
            strategy = cast("TemplateStrategy", self.templates.strategy("static"))
        return call_config, strategy

    def _events(
        self,
        question: str,
        chunks: Sequence[Chunk],
        config: TransformConfig,
        strategy: TemplateStrategy,
    ) -> Iterator[TransformEvent]:
        logger.debug("Transforming %d chunks in %s mode", len(chunks), config.template_mode)
        extraction = self.extractor.extract(question, chunks)
        yield SpansEvent(spans=extraction.spans)
        display, references = _split(extraction.spans, config)
        template = strategy.generate_result(question, _texts(display + references), len(references), seed=config.seed)
        yield AnswerEvent(answer=_answer(extraction, template, display, references, _fill(strategy, config)))

    async def _aevents(
        self,
        question: str,
        chunks: Sequence[Chunk],
        config: TransformConfig,
        strategy: TemplateStrategy,
    ) -> AsyncIterator[TransformEvent]:
        logger.debug("Transforming %d chunks in %s mode", len(chunks), config.template_mode)
        extraction = await self.extractor.aextract(question, chunks)
        yield SpansEvent(spans=extraction.spans)
        display, references = _split(extraction.spans, config)
        template = await strategy.generate_result_async(
            question, _texts(display + references), len(references), seed=config.seed
        )
        yield AnswerEvent(answer=_answer(extraction, template, display, references, _fill(strategy, config)))

config property

The settings of a call that passes none: the config given, else the templates'.

__init__(extractor, templates, *, config=None)

Build a transformer from an extractor and the templates of the answers.

:param extractor: Extracts the spans of the chunks :param templates: The templates the answers are written with :param config: The settings of a call, unless the call passes its own. Without it, the templates' settings are the default. :raises ConfigurationError: If the config is not a TransformConfig, or asks for structured template mode

Source code in packages/core/verbatim_core/transformer.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def __init__(
    self,
    extractor: SpanExtractor[Any],
    templates: TemplateManager,
    *,
    config: TransformConfig | None = None,
) -> None:
    """Build a transformer from an extractor and the templates of the answers.

    :param extractor: Extracts the spans of the chunks
    :param templates: The templates the answers are written with
    :param config: The settings of a call, unless the call passes its own. Without it, the
        templates' settings are the default.
    :raises ConfigurationError: If the config is not a TransformConfig, or asks for structured
        template mode
    """
    self.extractor = extractor
    self.templates = templates
    self._config = None if config is None else _default_config(config, TransformConfig)
    # Templates in structured mode raise at the call: VerbatimRAG answers structured
    # questions itself with the same templates (ADR 0004).
    if self._config is not None and self._config.template_mode == "structured":
        raise ConfigurationError(_STRUCTURED_MODE_UNSUPPORTED, code="invalid_value")
    if templates.strategy(self.config.template_mode) is None:
        warn_contextual_without_client()

transform(question, chunks, *, config=None)

Write a cited answer to the question from the chunks.

:param question: The question :param chunks: Chunks with unique ids :param config: Replaces the transformer's default config for this call :return: The answer, with its citations, spans, usage and warnings :raises InvalidInputError: If two chunks share an id, or the config is of the wrong type or asks for structured template mode :raises UpstreamError: If a service the extractor calls fails :raises ConfigurationError: If a service rejects its key, model or quota

Source code in packages/core/verbatim_core/transformer.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def transform(self, question: str, chunks: Sequence[Chunk], *, config: TransformConfig | None = None) -> Answer:
    """Write a cited answer to the question from the chunks.

    :param question: The question
    :param chunks: Chunks with unique ids
    :param config: Replaces the transformer's default config for this call
    :return: The answer, with its citations, spans, usage and warnings
    :raises InvalidInputError: If two chunks share an id, or the config is of the wrong type
        or asks for structured template mode
    :raises UpstreamError: If a service the extractor calls fails
    :raises ConfigurationError: If a service rejects its key, model or quota
    """
    config, strategy = self._call_settings(config)
    *_, answer_event = self._events(question, chunks, config, strategy)
    # The last event is always the answer.
    return cast("AnswerEvent", answer_event).answer

atransform(question, chunks, *, config=None) async

Async version of transform.

Source code in packages/core/verbatim_core/transformer.py 
111
112
113
114
115
116
117
async def atransform(
    self, question: str, chunks: Sequence[Chunk], *, config: TransformConfig | None = None
) -> Answer:
    """Async version of `transform`."""
    config, strategy = self._call_settings(config)
    events = [event async for event in self._aevents(question, chunks, config, strategy)]
    return cast("AnswerEvent", events[-1]).answer

stream(question, chunks, *, config=None)

Write the answer of transform as events: the spans, then the answer.

:raises InvalidInputError: When called, if the config is of the wrong type or asks for structured template mode. The other errors of transform come from the iteration.

Source code in packages/core/verbatim_core/transformer.py 
119
120
121
122
123
124
125
126
127
128
def stream(
    self, question: str, chunks: Sequence[Chunk], *, config: TransformConfig | None = None
) -> Iterator[TransformEvent]:
    """Write the answer of `transform` as events: the spans, then the answer.

    :raises InvalidInputError: When called, if the config is of the wrong type or asks for
        structured template mode. The other errors of `transform` come from the iteration.
    """
    config, strategy = self._call_settings(config)
    return self._events(question, chunks, config, strategy)

astream(question, chunks, *, config=None)

Async version of stream.

Source code in packages/core/verbatim_core/transformer.py 
130
131
132
133
134
135
def astream(
    self, question: str, chunks: Sequence[Chunk], *, config: TransformConfig | None = None
) -> AsyncIterator[TransformEvent]:
    """Async version of `stream`."""
    config, strategy = self._call_settings(config)
    return self._aevents(question, chunks, config, strategy)

TransformConfig

verbatim_core.transformer.TransformConfig

Bases: BaseModel

Settings of a Transformer call.

Source code in packages/core/verbatim_core/_transform_config.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
class TransformConfig(BaseModel):
    """Settings of a `Transformer` call."""

    model_config = ConfigDict(frozen=True, extra="forbid", hide_input_in_errors=True)

    # The first spans are shown in the answer, the rest are cited for reference only.
    max_display_spans: NonNegativeInt = 5
    # The Transformer rejects "structured": structured answers stay in VerbatimRAG (ADR 0004).
    template_mode: Literal["static", "contextual", "random", "question_specific", "structured"] = "static"
    citation_mode: Literal["inline", "hidden"] = "inline"  # citation markers in the text, or none
    citation_format: str = "[{number}]"  # the marker, with {number} and {span_id}
    seed: int | None = None  # random template mode chooses by the question and this seed

    @field_validator("citation_format")
    @classmethod
    def validate_citation_format(cls, citation_format: str) -> str:
        try:
            citation_format.format(number=1, span_id="1")
        except (KeyError, IndexError, ValueError) as e:
            msg = "citation_format may use only {number} and {span_id}"
            raise ValueError(msg) from e
        return citation_format

VerbatimTransform

verbatim_core.transform.VerbatimTransform

RAG-agnostic verbatim transform over a Transformer (sync/async).

Context items become chunks with positional ids (ctx_0, ctx_1, ...), and the Transformer's answer becomes a QueryResponse. The template manager's settings apply, with this transform's number of displayed spans.

Source code in packages/core/verbatim_core/transform.py 
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 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
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
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
@deprecated("VerbatimTransform", use="Transformer")
class VerbatimTransform:
    """RAG-agnostic verbatim transform over a `Transformer` (sync/async).

    Context items become chunks with positional ids (ctx_0, ctx_1, ...), and the `Transformer`'s
    answer becomes a QueryResponse. The template manager's settings apply, with this transform's
    number of displayed spans.
    """

    def __init__(  # noqa: PLR0913, PLR0917  # deprecated public signature, kept as it was
        self,
        llm_client: LLMClient | None = None,
        extractor: SpanExtractor[Any] | None = None,
        template_manager: TemplateManager | None = None,
        max_display_spans: int = _DEFAULT_MAX_DISPLAY_SPANS,
        extraction_mode: str = "auto",
        template_mode: str = "contextual",
        span_match_mode: str = "exact",
        fuzzy_threshold: float = 0.8,
        extraction_prompt: str | None = None,
        system_prompt: str | None = None,
    ) -> None:
        """Build the transform from its parts, with defaults for the parts not given.

        Args:
            llm_client: The client of the default extractor and template manager. Defaults to
                `LLMClient()`.
            extractor: The span extractor. Defaults to an `LLMSpanExtractor` with the extraction
                settings below.
            template_manager: The templates. Defaults to a `TemplateManager` in `template_mode`.
            max_display_spans: The number of spans shown in the answer. The rest are cited for
                reference only.
            extraction_mode: "batch", "individual" or "auto", for the default extractor.
            template_mode: The template mode of the default template manager.
            span_match_mode: "exact" or "fuzzy", for the default extractor.
            fuzzy_threshold: How closely, from 0 to 1, a quote must match the chunk text in fuzzy
                mode, for the default extractor.
            extraction_prompt: A Jinja2 prompt for the default extractor, instead of the bundled
                one.
            system_prompt: The system message of the default extractor, sent only with
                `extraction_prompt`.

        """
        self.llm_client = llm_client or LLMClient()
        self.extractor = extractor or LLMSpanExtractor(
            self.llm_client,
            config=LLMExtractConfig.model_validate(
                {
                    "extraction_mode": extraction_mode,
                    "span_match_mode": span_match_mode,
                    "fuzzy_threshold": fuzzy_threshold,
                    "extraction_prompt": extraction_prompt,
                    "system_prompt": system_prompt,
                }
            ),
        )
        self.template_manager = template_manager or TemplateManager(
            llm_client=self.llm_client, default_mode=template_mode
        )
        self.response_builder = ResponseBuilder()  # unused, kept for code that reads it
        self.max_display_spans = max_display_spans
        self._transformer = Transformer(_as_span_extractor(self.extractor), self.template_manager)

    def transform(
        self,
        question: str,
        context: Iterable[dict[str, Any]],
        answer: str | None = None,  # noqa: ARG002  # ignored, kept for compatibility
    ) -> QueryResponse:
        """Answer the question from the context items, with citations.

        Args:
            question: The question to answer.
            context: Dicts with "content" or "text", and optionally "title", "source" and
                "metadata", or objects with a text attribute.
            answer: Ignored.

        Returns:
            The cited answer.

        Raises:
            InvalidInputError: If a context item has another shape.
            ConfigurationError: If the template manager is in structured mode.

        """
        return _query_response(self._transformer, question, context, self._settings())

    async def transform_async(
        self,
        question: str,
        context: Iterable[dict[str, Any]],
        answer: str | None = None,  # noqa: ARG002  # ignored, kept for compatibility
    ) -> QueryResponse:
        """Async version of `transform`."""
        return await _aquery_response(self._transformer, question, context, self._settings())

    def _settings(self) -> TransformConfig:
        """Return the template manager's settings, with this transform's number of displayed spans."""
        if self.template_manager.config.template_mode == "structured":
            raise ConfigurationError(_STRUCTURED_MODE_UNSUPPORTED, code="invalid_value")
        return self.template_manager.config.model_copy(update={"max_display_spans": self.max_display_spans})

__init__(llm_client=None, extractor=None, template_manager=None, max_display_spans=_DEFAULT_MAX_DISPLAY_SPANS, extraction_mode='auto', template_mode='contextual', span_match_mode='exact', fuzzy_threshold=0.8, extraction_prompt=None, system_prompt=None)

Build the transform from its parts, with defaults for the parts not given.

Parameters:

Name Type Description Default
llm_client LLMClient | None

The client of the default extractor and template manager. Defaults to LLMClient().

 None
extractor SpanExtractor[Any] | None

The span extractor. Defaults to an LLMSpanExtractor with the extraction settings below.

 None
template_manager TemplateManager | None

The templates. Defaults to a TemplateManager in template_mode.

 None
max_display_spans int

The number of spans shown in the answer. The rest are cited for reference only.

 _DEFAULT_MAX_DISPLAY_SPANS
extraction_mode str

"batch", "individual" or "auto", for the default extractor.

 'auto'
template_mode str

The template mode of the default template manager.

 'contextual'
span_match_mode str

"exact" or "fuzzy", for the default extractor.

 'exact'
fuzzy_threshold float

How closely, from 0 to 1, a quote must match the chunk text in fuzzy mode, for the default extractor.

 0.8
extraction_prompt str | None

A Jinja2 prompt for the default extractor, instead of the bundled one.

 None
system_prompt str | None

The system message of the default extractor, sent only with extraction_prompt.

 None
Source code in packages/core/verbatim_core/transform.py 
39
40
41
42
43
44
45
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
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def __init__(  # noqa: PLR0913, PLR0917  # deprecated public signature, kept as it was
    self,
    llm_client: LLMClient | None = None,
    extractor: SpanExtractor[Any] | None = None,
    template_manager: TemplateManager | None = None,
    max_display_spans: int = _DEFAULT_MAX_DISPLAY_SPANS,
    extraction_mode: str = "auto",
    template_mode: str = "contextual",
    span_match_mode: str = "exact",
    fuzzy_threshold: float = 0.8,
    extraction_prompt: str | None = None,
    system_prompt: str | None = None,
) -> None:
    """Build the transform from its parts, with defaults for the parts not given.

    Args:
        llm_client: The client of the default extractor and template manager. Defaults to
            `LLMClient()`.
        extractor: The span extractor. Defaults to an `LLMSpanExtractor` with the extraction
            settings below.
        template_manager: The templates. Defaults to a `TemplateManager` in `template_mode`.
        max_display_spans: The number of spans shown in the answer. The rest are cited for
            reference only.
        extraction_mode: "batch", "individual" or "auto", for the default extractor.
        template_mode: The template mode of the default template manager.
        span_match_mode: "exact" or "fuzzy", for the default extractor.
        fuzzy_threshold: How closely, from 0 to 1, a quote must match the chunk text in fuzzy
            mode, for the default extractor.
        extraction_prompt: A Jinja2 prompt for the default extractor, instead of the bundled
            one.
        system_prompt: The system message of the default extractor, sent only with
            `extraction_prompt`.

    """
    self.llm_client = llm_client or LLMClient()
    self.extractor = extractor or LLMSpanExtractor(
        self.llm_client,
        config=LLMExtractConfig.model_validate(
            {
                "extraction_mode": extraction_mode,
                "span_match_mode": span_match_mode,
                "fuzzy_threshold": fuzzy_threshold,
                "extraction_prompt": extraction_prompt,
                "system_prompt": system_prompt,
            }
        ),
    )
    self.template_manager = template_manager or TemplateManager(
        llm_client=self.llm_client, default_mode=template_mode
    )
    self.response_builder = ResponseBuilder()  # unused, kept for code that reads it
    self.max_display_spans = max_display_spans
    self._transformer = Transformer(_as_span_extractor(self.extractor), self.template_manager)

transform(question, context, answer=None)

Answer the question from the context items, with citations.

Parameters:

Name Type Description Default
question str

The question to answer.

 required
context Iterable[dict[str, Any]]

Dicts with "content" or "text", and optionally "title", "source" and "metadata", or objects with a text attribute.

 required
answer str | None

Ignored.

 None

Returns:

Type Description
QueryResponse

The cited answer.

Raises:

Type Description
InvalidInputError

If a context item has another shape.
ConfigurationError

If the template manager is in structured mode.
Source code in packages/core/verbatim_core/transform.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def transform(
    self,
    question: str,
    context: Iterable[dict[str, Any]],
    answer: str | None = None,  # noqa: ARG002  # ignored, kept for compatibility
) -> QueryResponse:
    """Answer the question from the context items, with citations.

    Args:
        question: The question to answer.
        context: Dicts with "content" or "text", and optionally "title", "source" and
            "metadata", or objects with a text attribute.
        answer: Ignored.

    Returns:
        The cited answer.

    Raises:
        InvalidInputError: If a context item has another shape.
        ConfigurationError: If the template manager is in structured mode.

    """
    return _query_response(self._transformer, question, context, self._settings())

transform_async(question, context, answer=None) async

Async version of transform.

Source code in packages/core/verbatim_core/transform.py 
117
118
119
120
121
122
123
124
async def transform_async(
    self,
    question: str,
    context: Iterable[dict[str, Any]],
    answer: str | None = None,  # noqa: ARG002  # ignored, kept for compatibility
) -> QueryResponse:
    """Async version of `transform`."""
    return await _aquery_response(self._transformer, question, context, self._settings())

LLMSpanExtractor

verbatim_core.extractors.LLMSpanExtractor

Bases: SpanExtractor[LLMExtractConfig]

Extract spans with an LLM that quotes the text of each chunk that answers the question.

Each quote is located in its chunk: the first exact occurrence, else in fuzzy match mode the best fuzzy match, the earliest on a tie. A span's text is the chunk's text at the located offsets, never the quote. A quote that can't be located is dropped and reported in the result's warnings with the code quote_not_found. Every span has the score 1.0, "extracted without a score".

A failed LLM call raises, and the whole extraction fails.

Source code in packages/core/verbatim_core/extractors.py 
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
class LLMSpanExtractor(SpanExtractor[LLMExtractConfig]):
    """Extract spans with an LLM that quotes the text of each chunk that answers the question.

    Each quote is located in its chunk: the first exact occurrence, else in fuzzy match mode the
    best fuzzy match, the earliest on a tie. A span's text is the chunk's text at the located
    offsets, never the quote. A quote that can't be located is dropped and reported in the
    result's warnings with the code `quote_not_found`. Every span has the score 1.0, "extracted
    without a score".

    A failed LLM call raises, and the whole extraction fails.
    """

    def __init__(  # noqa: PLR0913, PLR0917  # deprecated positional settings, kept for compatibility
        self,
        llm_client: LLMClient | None = None,
        model: str = _UNSET,
        extraction_mode: str = _UNSET,
        max_display_spans: int = _UNSET,
        batch_size: int = _UNSET,
        span_match_mode: str = _UNSET,
        fuzzy_threshold: float = _UNSET,
        extraction_prompt: str | None = _UNSET,
        system_prompt: str | None = _UNSET,
        *,
        config: LLMExtractConfig | None = None,
    ) -> None:
        """Build an extractor over an LLM client.

        :param llm_client: The client that makes the LLM calls
        :param config: The settings of a call, unless the call passes its own

        The other arguments are deprecated. Without ``llm_client`` the extractor builds an
        ``LLMClient(model)``. The settings from ``extraction_mode`` to ``system_prompt`` belong in
        ``config``, and ``max_display_spans`` is unused.
        :raises ConfigurationError: If ``config`` is not an ``LLMExtractConfig``, is passed
            together with the deprecated settings, or a deprecated setting has an invalid value
        """
        settings = {
            "extraction_mode": extraction_mode,
            "batch_size": batch_size,
            "span_match_mode": span_match_mode,
            "fuzzy_threshold": fuzzy_threshold,
            "extraction_prompt": extraction_prompt,
            "system_prompt": system_prompt,
        }
        self.config = _config_with_old_settings(type(self), LLMExtractConfig, config, settings)
        _warn_unused(type(self), max_display_spans=max_display_spans)
        if llm_client is None:
            warn_deprecated("LLMSpanExtractor without an llm_client", use="LLMSpanExtractor(LLMClient(model))")
            llm_client = LLMClient("gpt-4o-mini" if model is _UNSET else model)
        elif model is not _UNSET:
            warn_deprecated("Passing model to LLMSpanExtractor", use="the model of the llm_client")

        self.llm_client = llm_client

    @override
    def extract(self, question: str, chunks: Sequence[Chunk], *, config: LLMExtractConfig | None = None) -> Extraction:
        """Extract the spans of the chunks that are relevant to the question.

        :param question: The query or question
        :param chunks: Chunks with unique ids
        :param config: Replaces the extractor's default config for this call
        :return: The spans, with the usage and warnings of the call
        :raises InvalidInputError: If two chunks share an id, or the config is of the wrong type
        :raises UpstreamError: If an LLM call fails or its reply can't be read
        :raises ConfigurationError: If the LLM service rejects the key, the model or the quota
        """
        config = _call_config(config, self.config, LLMExtractConfig)
        _check_unique_ids(chunks)
        calls = _plan_llm_calls(question, chunks, config)
        logger.debug("Extracting spans from %d chunks in %d LLM calls", len(chunks), len(calls))
        replies = [_reply(self.llm_client, call) for call in calls]
        return _extraction_from_replies(chunks, calls, replies, config, self._model_id())

    @override
    async def aextract(
        self, question: str, chunks: Sequence[Chunk], *, config: LLMExtractConfig | None = None
    ) -> Extraction:
        """Async version of `extract`. It makes its LLM calls concurrently."""
        config = _call_config(config, self.config, LLMExtractConfig)
        _check_unique_ids(chunks)
        calls = _plan_llm_calls(question, chunks, config)
        logger.debug("Extracting spans from %d chunks in %d LLM calls", len(chunks), len(calls))
        replies = await asyncio.gather(*(_areply(self.llm_client, call) for call in calls))
        return _extraction_from_replies(chunks, calls, replies, config, self._model_id())

    @deprecated("LLMSpanExtractor.extract_spans", use="extract()")
    def extract_spans(self, question: str, search_results: list[Any]) -> dict[str, list[str]]:
        """Extract relevant spans from search results.

        Unlike `extract`, it tolerates failed LLM calls. When a batch call fails, it retries the
        chunks of the batch one by one, and a chunk whose call fails yields no spans.

        :param question: The query or question
        :param search_results: List of search results to extract from
        :return: Dictionary mapping result text to list of relevant spans
        """
        chunks = _chunks_from_results(search_results)
        batched, groups = _call_groups(len(chunks), self.config)
        spans = [span for group in groups for span in self._extract_group(question, chunks, group, batched=batched)]
        return _span_texts_by_chunk_text(chunks, Extraction(spans=spans))

    @deprecated("LLMSpanExtractor.extract_spans_async", use="aextract()")
    async def extract_spans_async(self, question: str, search_results: list[Any]) -> dict[str, list[str]]:
        """Async version of `extract_spans`."""
        if _below(type(self), "extract_spans", "extract_spans_async"):
            # A subclass's own extract_spans, written before 0.3.0
            return await asyncio.to_thread(self.extract_spans, question, search_results)
        chunks = _chunks_from_results(search_results)
        batched, groups = _call_groups(len(chunks), self.config)
        group_spans = await asyncio.gather(
            *(self._aextract_group(question, chunks, group, batched=batched) for group in groups)
        )
        spans = [span for spans in group_spans for span in spans]
        return _span_texts_by_chunk_text(chunks, Extraction(spans=spans))

    def _extract_group(self, question: str, chunks: list[Chunk], group: range, *, batched: bool) -> list[ExtractedSpan]:
        """Extract the spans of the chunks one LLM call covers, retrying them one by one if a batch call fails."""
        extract = _new_extract(self)
        if batched:
            try:
                batch = chunks[group.start : group.stop]
                return extract(question, batch, config=self._batch_config).spans
            except UpstreamError as error:
                _log_retry(group, error)
        spans: list[ExtractedSpan] = []
        for position in group:
            try:
                config = self._individual_config
                spans += extract(question, [chunks[position]], config=config).spans
            except UpstreamError as error:  # noqa: PERF203  # a failed chunk is logged, the others still run
                _log_failed_chunk(position, error)
        return spans

    async def _aextract_group(
        self, question: str, chunks: list[Chunk], group: range, *, batched: bool
    ) -> list[ExtractedSpan]:
        """Async version of `_extract_group`."""
        aextract = _new_aextract(self)
        if batched:
            try:
                extraction = await aextract(question, chunks[group.start : group.stop], config=self._batch_config)
            except UpstreamError as error:
                _log_retry(group, error)
            else:
                return extraction.spans
        spans: list[ExtractedSpan] = []
        for position in group:
            try:
                extraction = await aextract(question, [chunks[position]], config=self._individual_config)
                spans += extraction.spans
            except UpstreamError as error:  # noqa: PERF203  # a failed chunk is logged, the others still run
                _log_failed_chunk(position, error)
        return spans

    @property
    def _batch_config(self) -> LLMExtractConfig:
        return self.config.model_copy(update={"extraction_mode": "batch"})

    @property
    def _individual_config(self) -> LLMExtractConfig:
        return self.config.model_copy(update={"extraction_mode": "individual"})

    def _model_id(self) -> str | None:
        return cast("str | None", getattr(self.llm_client, "model", None))

__init__(llm_client=None, model=_UNSET, extraction_mode=_UNSET, max_display_spans=_UNSET, batch_size=_UNSET, span_match_mode=_UNSET, fuzzy_threshold=_UNSET, extraction_prompt=_UNSET, system_prompt=_UNSET, *, config=None)

Build an extractor over an LLM client.

:param llm_client: The client that makes the LLM calls :param config: The settings of a call, unless the call passes its own

The other arguments are deprecated. Without llm_client the extractor builds an LLMClient(model). The settings from extraction_mode to system_prompt belong in config, and max_display_spans is unused. :raises ConfigurationError: If config is not an LLMExtractConfig, is passed together with the deprecated settings, or a deprecated setting has an invalid value

Source code in packages/core/verbatim_core/extractors.py 
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
def __init__(  # noqa: PLR0913, PLR0917  # deprecated positional settings, kept for compatibility
    self,
    llm_client: LLMClient | None = None,
    model: str = _UNSET,
    extraction_mode: str = _UNSET,
    max_display_spans: int = _UNSET,
    batch_size: int = _UNSET,
    span_match_mode: str = _UNSET,
    fuzzy_threshold: float = _UNSET,
    extraction_prompt: str | None = _UNSET,
    system_prompt: str | None = _UNSET,
    *,
    config: LLMExtractConfig | None = None,
) -> None:
    """Build an extractor over an LLM client.

    :param llm_client: The client that makes the LLM calls
    :param config: The settings of a call, unless the call passes its own

    The other arguments are deprecated. Without ``llm_client`` the extractor builds an
    ``LLMClient(model)``. The settings from ``extraction_mode`` to ``system_prompt`` belong in
    ``config``, and ``max_display_spans`` is unused.
    :raises ConfigurationError: If ``config`` is not an ``LLMExtractConfig``, is passed
        together with the deprecated settings, or a deprecated setting has an invalid value
    """
    settings = {
        "extraction_mode": extraction_mode,
        "batch_size": batch_size,
        "span_match_mode": span_match_mode,
        "fuzzy_threshold": fuzzy_threshold,
        "extraction_prompt": extraction_prompt,
        "system_prompt": system_prompt,
    }
    self.config = _config_with_old_settings(type(self), LLMExtractConfig, config, settings)
    _warn_unused(type(self), max_display_spans=max_display_spans)
    if llm_client is None:
        warn_deprecated("LLMSpanExtractor without an llm_client", use="LLMSpanExtractor(LLMClient(model))")
        llm_client = LLMClient("gpt-4o-mini" if model is _UNSET else model)
    elif model is not _UNSET:
        warn_deprecated("Passing model to LLMSpanExtractor", use="the model of the llm_client")

    self.llm_client = llm_client

extract(question, chunks, *, config=None)

Extract the spans of the chunks that are relevant to the question.

:param question: The query or question :param chunks: Chunks with unique ids :param config: Replaces the extractor's default config for this call :return: The spans, with the usage and warnings of the call :raises InvalidInputError: If two chunks share an id, or the config is of the wrong type :raises UpstreamError: If an LLM call fails or its reply can't be read :raises ConfigurationError: If the LLM service rejects the key, the model or the quota

Source code in packages/core/verbatim_core/extractors.py 
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
@override
def extract(self, question: str, chunks: Sequence[Chunk], *, config: LLMExtractConfig | None = None) -> Extraction:
    """Extract the spans of the chunks that are relevant to the question.

    :param question: The query or question
    :param chunks: Chunks with unique ids
    :param config: Replaces the extractor's default config for this call
    :return: The spans, with the usage and warnings of the call
    :raises InvalidInputError: If two chunks share an id, or the config is of the wrong type
    :raises UpstreamError: If an LLM call fails or its reply can't be read
    :raises ConfigurationError: If the LLM service rejects the key, the model or the quota
    """
    config = _call_config(config, self.config, LLMExtractConfig)
    _check_unique_ids(chunks)
    calls = _plan_llm_calls(question, chunks, config)
    logger.debug("Extracting spans from %d chunks in %d LLM calls", len(chunks), len(calls))
    replies = [_reply(self.llm_client, call) for call in calls]
    return _extraction_from_replies(chunks, calls, replies, config, self._model_id())

aextract(question, chunks, *, config=None) async

Async version of extract. It makes its LLM calls concurrently.

Source code in packages/core/verbatim_core/extractors.py 
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
@override
async def aextract(
    self, question: str, chunks: Sequence[Chunk], *, config: LLMExtractConfig | None = None
) -> Extraction:
    """Async version of `extract`. It makes its LLM calls concurrently."""
    config = _call_config(config, self.config, LLMExtractConfig)
    _check_unique_ids(chunks)
    calls = _plan_llm_calls(question, chunks, config)
    logger.debug("Extracting spans from %d chunks in %d LLM calls", len(chunks), len(calls))
    replies = await asyncio.gather(*(_areply(self.llm_client, call) for call in calls))
    return _extraction_from_replies(chunks, calls, replies, config, self._model_id())

extract_spans(question, search_results)

Extract relevant spans from search results.

Unlike extract, it tolerates failed LLM calls. When a batch call fails, it retries the chunks of the batch one by one, and a chunk whose call fails yields no spans.

:param question: The query or question :param search_results: List of search results to extract from :return: Dictionary mapping result text to list of relevant spans

Source code in packages/core/verbatim_core/extractors.py 
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
@deprecated("LLMSpanExtractor.extract_spans", use="extract()")
def extract_spans(self, question: str, search_results: list[Any]) -> dict[str, list[str]]:
    """Extract relevant spans from search results.

    Unlike `extract`, it tolerates failed LLM calls. When a batch call fails, it retries the
    chunks of the batch one by one, and a chunk whose call fails yields no spans.

    :param question: The query or question
    :param search_results: List of search results to extract from
    :return: Dictionary mapping result text to list of relevant spans
    """
    chunks = _chunks_from_results(search_results)
    batched, groups = _call_groups(len(chunks), self.config)
    spans = [span for group in groups for span in self._extract_group(question, chunks, group, batched=batched)]
    return _span_texts_by_chunk_text(chunks, Extraction(spans=spans))

extract_spans_async(question, search_results) async

Async version of extract_spans.

Source code in packages/core/verbatim_core/extractors.py 
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
@deprecated("LLMSpanExtractor.extract_spans_async", use="aextract()")
async def extract_spans_async(self, question: str, search_results: list[Any]) -> dict[str, list[str]]:
    """Async version of `extract_spans`."""
    if _below(type(self), "extract_spans", "extract_spans_async"):
        # A subclass's own extract_spans, written before 0.3.0
        return await asyncio.to_thread(self.extract_spans, question, search_results)
    chunks = _chunks_from_results(search_results)
    batched, groups = _call_groups(len(chunks), self.config)
    group_spans = await asyncio.gather(
        *(self._aextract_group(question, chunks, group, batched=batched) for group in groups)
    )
    spans = [span for spans in group_spans for span in spans]
    return _span_texts_by_chunk_text(chunks, Extraction(spans=spans))

SpanExtractor

verbatim_core.extractors.SpanExtractor

Bases: ABC, Generic[_ConfigT]

Base class for span extractors.

A subclass implements extract, and aextract when it has a native async path. Otherwise aextract runs extract in a worker thread. The type argument is the subclass's config class, as in SpanExtractor[LLMExtractConfig].

A subclass written against the old interface implements only extract_spans. It gets extract and aextract from this class, which locate each span text in its chunk and warn that the subclass should implement extract. A span text that can't be located is dropped and reported in the result's warnings with the code span_text_not_found.

Of a subclass's methods, the one defined in the most specific class is the implementation. A subclass of an extractor that overrides extract_spans, as subclasses written before 0.3.0 did, is called through it like a subclass that implements only extract_spans. A subclass that overrides extract below a parent's aextract gets an aextract that runs its extract in a worker thread. A call's LLM calls then run one after another: to keep the parent's concurrent aextract, override aextract too, for example with await super().aextract(...) and the same changes to its result as in extract.

Source code in packages/core/verbatim_core/extractors.py 
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
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
class SpanExtractor(ABC, Generic[_ConfigT]):
    """Base class for span extractors.

    A subclass implements `extract`, and `aextract` when it has a native async path. Otherwise
    `aextract` runs `extract` in a worker thread. The type argument is the subclass's config
    class, as in `SpanExtractor[LLMExtractConfig]`.

    A subclass written against the old interface implements only `extract_spans`. It gets
    `extract` and `aextract` from this class, which locate each span text in its chunk and warn
    that the subclass should implement `extract`. A span text that can't be located is dropped
    and reported in the result's warnings with the code `span_text_not_found`.

    Of a subclass's methods, the one defined in the most specific class is the implementation. A
    subclass of an extractor that overrides `extract_spans`, as subclasses written before 0.3.0
    did, is called through it like a subclass that implements only `extract_spans`. A subclass
    that overrides `extract` below a parent's `aextract` gets an `aextract` that runs its
    `extract` in a worker thread. A call's LLM calls then run one after another: to keep the
    parent's concurrent `aextract`, override `aextract` too, for example with
    `await super().aextract(...)` and the same changes to its result as in `extract`.
    """

    # For a subclass that overrides extract_spans below its parent's extract: the parent, whose
    # extract the extract_spans adapters call. The subclass's own extract calls extract_spans.
    _extract_owner: ClassVar[type | None] = None

    def __init_subclass__(cls, **kwargs: object) -> None:
        """Check that the subclass implements `extract` or `extract_spans`, and route its calls."""
        super().__init_subclass__(**kwargs)
        if not (_overrides(cls, "extract") or _overrides(cls, "extract_spans")):
            msg = f"{cls.__qualname__} must implement extract() or extract_spans()"
            raise TypeError(msg)
        extract_owner = _owner(cls, "extract")
        if extract_owner is not SpanExtractor and (
            _below(cls, "extract_spans", "extract") or _below(cls, "extract_spans_async", "extract")
        ):
            cls._extract_owner = cls._extract_owner or extract_owner
            cls.extract = SpanExtractor.extract
            cls.aextract = SpanExtractor.aextract
        elif _owner(cls, "aextract") is not SpanExtractor and _below(cls, "extract", "aextract"):
            cls.aextract = SpanExtractor.aextract  # runs extract in a worker thread

    def extract(self, question: str, chunks: Sequence[Chunk], *, config: _ConfigT | None = None) -> Extraction:
        """Extract the spans of the chunks that are relevant to the question.

        :param question: The query or question
        :param chunks: Chunks with unique ids
        :param config: Replaces the extractor's default config for this call
        :return: The spans, with the usage and warnings of the call
        :raises InvalidInputError: If two chunks share an id
        """
        _check_old_interface_call(type(self), chunks, config)
        return _locate_span_texts(chunks, self.extract_spans(question, _as_search_results(chunks)))

    async def aextract(self, question: str, chunks: Sequence[Chunk], *, config: _ConfigT | None = None) -> Extraction:
        """Async version of `extract`."""
        if _overrides(type(self), "extract"):
            return await asyncio.to_thread(self.extract, question, chunks, config=config)

        _check_old_interface_call(type(self), chunks, config)
        if not _below(type(self), "extract_spans", "extract_spans_async"):
            span_texts = await self.extract_spans_async(question, _as_search_results(chunks))
        else:
            span_texts = await asyncio.to_thread(self.extract_spans, question, _as_search_results(chunks))
        return _locate_span_texts(chunks, span_texts)

    @deprecated("SpanExtractor.extract_spans", use="extract()")
    def extract_spans(self, question: str, search_results: list[Any]) -> dict[str, list[str]]:
        """Extract relevant spans from search results.

        :param question: The query or question
        :param search_results: List of search results to extract from
        :return: Dictionary mapping result text to list of relevant spans
        """
        chunks = _chunks_from_results(search_results)
        return _span_texts_by_chunk_text(chunks, _new_extract(self)(question, chunks))

    @deprecated("SpanExtractor.extract_spans_async", use="aextract()")
    async def extract_spans_async(self, question: str, search_results: list[Any]) -> dict[str, list[str]]:
        """Async version of `extract_spans`."""
        if not _overrides(type(self), "extract"):
            return await asyncio.to_thread(self.extract_spans, question, search_results)

        chunks = _chunks_from_results(search_results)
        return _span_texts_by_chunk_text(chunks, await _new_aextract(self)(question, chunks))

__init_subclass__(**kwargs)

Check that the subclass implements extract or extract_spans, and route its calls.

Source code in packages/core/verbatim_core/extractors.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def __init_subclass__(cls, **kwargs: object) -> None:
    """Check that the subclass implements `extract` or `extract_spans`, and route its calls."""
    super().__init_subclass__(**kwargs)
    if not (_overrides(cls, "extract") or _overrides(cls, "extract_spans")):
        msg = f"{cls.__qualname__} must implement extract() or extract_spans()"
        raise TypeError(msg)
    extract_owner = _owner(cls, "extract")
    if extract_owner is not SpanExtractor and (
        _below(cls, "extract_spans", "extract") or _below(cls, "extract_spans_async", "extract")
    ):
        cls._extract_owner = cls._extract_owner or extract_owner
        cls.extract = SpanExtractor.extract
        cls.aextract = SpanExtractor.aextract
    elif _owner(cls, "aextract") is not SpanExtractor and _below(cls, "extract", "aextract"):
        cls.aextract = SpanExtractor.aextract  # runs extract in a worker thread

extract(question, chunks, *, config=None)

Extract the spans of the chunks that are relevant to the question.

:param question: The query or question :param chunks: Chunks with unique ids :param config: Replaces the extractor's default config for this call :return: The spans, with the usage and warnings of the call :raises InvalidInputError: If two chunks share an id

Source code in packages/core/verbatim_core/extractors.py 
102
103
104
105
106
107
108
109
110
111
112
def extract(self, question: str, chunks: Sequence[Chunk], *, config: _ConfigT | None = None) -> Extraction:
    """Extract the spans of the chunks that are relevant to the question.

    :param question: The query or question
    :param chunks: Chunks with unique ids
    :param config: Replaces the extractor's default config for this call
    :return: The spans, with the usage and warnings of the call
    :raises InvalidInputError: If two chunks share an id
    """
    _check_old_interface_call(type(self), chunks, config)
    return _locate_span_texts(chunks, self.extract_spans(question, _as_search_results(chunks)))

aextract(question, chunks, *, config=None) async

Async version of extract.

Source code in packages/core/verbatim_core/extractors.py 
114
115
116
117
118
119
120
121
122
123
124
async def aextract(self, question: str, chunks: Sequence[Chunk], *, config: _ConfigT | None = None) -> Extraction:
    """Async version of `extract`."""
    if _overrides(type(self), "extract"):
        return await asyncio.to_thread(self.extract, question, chunks, config=config)

    _check_old_interface_call(type(self), chunks, config)
    if not _below(type(self), "extract_spans", "extract_spans_async"):
        span_texts = await self.extract_spans_async(question, _as_search_results(chunks))
    else:
        span_texts = await asyncio.to_thread(self.extract_spans, question, _as_search_results(chunks))
    return _locate_span_texts(chunks, span_texts)

extract_spans(question, search_results)

Extract relevant spans from search results.

:param question: The query or question :param search_results: List of search results to extract from :return: Dictionary mapping result text to list of relevant spans

Source code in packages/core/verbatim_core/extractors.py 
126
127
128
129
130
131
132
133
134
135
@deprecated("SpanExtractor.extract_spans", use="extract()")
def extract_spans(self, question: str, search_results: list[Any]) -> dict[str, list[str]]:
    """Extract relevant spans from search results.

    :param question: The query or question
    :param search_results: List of search results to extract from
    :return: Dictionary mapping result text to list of relevant spans
    """
    chunks = _chunks_from_results(search_results)
    return _span_texts_by_chunk_text(chunks, _new_extract(self)(question, chunks))

extract_spans_async(question, search_results) async

Async version of extract_spans.

Source code in packages/core/verbatim_core/extractors.py 
137
138
139
140
141
142
143
144
@deprecated("SpanExtractor.extract_spans_async", use="aextract()")
async def extract_spans_async(self, question: str, search_results: list[Any]) -> dict[str, list[str]]:
    """Async version of `extract_spans`."""
    if not _overrides(type(self), "extract"):
        return await asyncio.to_thread(self.extract_spans, question, search_results)

    chunks = _chunks_from_results(search_results)
    return _span_texts_by_chunk_text(chunks, await _new_aextract(self)(question, chunks))

LLMClient

verbatim_core.llm_client.LLMClient

Centralized LLM interaction handler with async support.

Provides a unified interface for all OpenAI API calls used throughout the Verbatim RAG system, including span extraction and template generation.

Source code in packages/core/verbatim_core/llm_client.py 
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
class LLMClient:
    """Centralized LLM interaction handler with async support.

    Provides a unified interface for all OpenAI API calls used throughout
    the Verbatim RAG system, including span extraction and template generation.
    """

    def __init__(
        self,
        model: str = "gpt-4o-mini",
        temperature: float = 0.7,
        api_base: str = "https://api.openai.com/v1",
        api_key: str | None = None,
    ) -> None:
        """Initialize the LLM client.

        :param model: The OpenAI model to use
        :param temperature: Default temperature for completions
        :param api_base: The base URL for the OpenAI API (can be used with custom models and with VLLM)
        :param api_key: Optional API key. Falls back to OPENAI_API_KEY when unset.
        """
        self.model = model
        self.temperature = temperature
        self.api_key = api_key or os.getenv("OPENAI_API_KEY") or "EMPTY"
        self.client = openai.OpenAI(base_url=api_base, api_key=self.api_key)

        self.async_client = openai.AsyncOpenAI(base_url=api_base, api_key=self.api_key)

    def complete(
        self,
        prompt: str,
        json_mode: bool = False,  # noqa: FBT001, FBT002  # public signature
        temperature: float | None = None,
        system_prompt: str | None = None,
    ) -> str:
        """Complete text synchronously.

        :param prompt: The prompt to send
        :param json_mode: Whether to request JSON output format
        :param temperature: Override default temperature
        :param system_prompt: Optional system message to prepend
        :return: The completion text
        :raises UpstreamError: If the call fails or the reply is empty
        :raises ConfigurationError: If the service rejects the key, the model or the quota
        """
        return self.completion(prompt, json_mode=json_mode, temperature=temperature, system_prompt=system_prompt).text

    async def complete_async(
        self,
        prompt: str,
        json_mode: bool = False,  # noqa: FBT001, FBT002  # public signature
        temperature: float | None = None,
        system_prompt: str | None = None,
    ) -> str:
        """Asynchronous text completion. See `complete`."""
        completion = await self.completion_async(
            prompt, json_mode=json_mode, temperature=temperature, system_prompt=system_prompt
        )
        return completion.text

    def completion(
        self,
        prompt: str,
        *,
        json_mode: bool = False,
        temperature: float | None = None,
        system_prompt: str | None = None,
    ) -> Completion:
        """Like `complete`, and also returns the model name and the token counts the provider sent."""
        request = self._request(prompt, json_mode, temperature, system_prompt)
        try:
            response = self.client.chat.completions.create(**request)
        except APIError as e:
            raise _translate_error(e) from e
        return _completion(response)

    async def completion_async(
        self,
        prompt: str,
        *,
        json_mode: bool = False,
        temperature: float | None = None,
        system_prompt: str | None = None,
    ) -> Completion:
        """Asynchronous version of `completion`."""
        request = self._request(prompt, json_mode, temperature, system_prompt)
        try:
            response = await self.async_client.chat.completions.create(**request)
        except APIError as e:
            raise _translate_error(e) from e
        return _completion(response)

    def _request(
        self,
        prompt: str,
        json_mode: bool,  # noqa: FBT001  # mirrors the keyword arguments of the public callers
        temperature: float | None,
        system_prompt: str | None,
    ) -> dict[str, Any]:
        messages = []
        if system_prompt:
            messages.append({"role": "system", "content": system_prompt})
        messages.append({"role": "user", "content": prompt})
        request = {
            "model": self.model,
            "messages": messages,
            "temperature": temperature if temperature is not None else self.temperature,
        }
        if json_mode:
            request["response_format"] = {"type": "json_object"}
        return request

    def extract_spans(self, question: str, documents: dict[str, str]) -> dict[str, list[str]]:
        """Specialized method for span extraction from documents.

        :param question: The user's question
        :param documents: Dictionary mapping doc IDs to document text
        :return: Dictionary mapping doc IDs to lists of extracted spans
        """
        prompt = self._build_extraction_prompt(question, documents)
        try:
            response = self.complete(prompt, json_mode=True)
            return json.loads(response)
        except (json.JSONDecodeError, KeyError) as e:
            logger.warning("Span extraction failed: %s", e)
            # Return empty results for all documents on failure
            return {doc_id: [] for doc_id in documents}

    async def extract_spans_async(self, question: str, documents: dict[str, str]) -> dict[str, list[str]]:
        """Async span extraction from documents.

        :param question: The user's question
        :param documents: Dictionary mapping doc IDs to document text
        :return: Dictionary mapping doc IDs to lists of extracted spans
        """
        prompt = self._build_extraction_prompt(question, documents)
        try:
            response = await self.complete_async(prompt, json_mode=True)
            return json.loads(response)
        except (json.JSONDecodeError, KeyError) as e:
            logger.warning("Async span extraction failed: %s", e)
            return {doc_id: [] for doc_id in documents}

    def extract_structured(
        self,
        question: str,
        template: str,
        placeholders: dict[str, str],
        documents: list[str],
    ) -> dict[str, list[dict[str, any]]]:
        """Extract spans organized by template placeholders with document attribution.

        :param question: The user's question
        :param template: Template with placeholders like [METHODOLOGY]
        :param placeholders: Dict mapping placeholder names to hints
        :param documents: List of document texts
        :return: Dict mapping placeholder names to lists of {text, doc} objects
        """
        prompt = self._build_structured_extraction_prompt(question, template, placeholders, documents)
        try:
            response = self.complete(prompt, json_mode=True)
            return self._normalize_structured_response(json.loads(response), placeholders)
        except (json.JSONDecodeError, KeyError) as e:
            logger.warning("Structured extraction failed: %s", e)
            return {name: [] for name in placeholders}

    async def extract_structured_async(
        self,
        question: str,
        template: str,
        placeholders: dict[str, str],
        documents: list[str],
    ) -> dict[str, list[dict[str, any]]]:
        """Async version of structured extraction with document attribution.

        :param question: The user's question
        :param template: Template with placeholders like [METHODOLOGY]
        :param placeholders: Dict mapping placeholder names to hints
        :param documents: List of document texts
        :return: Dict mapping placeholder names to lists of {text, doc} objects
        """
        prompt = self._build_structured_extraction_prompt(question, template, placeholders, documents)
        try:
            response = await self.complete_async(prompt, json_mode=True)
            return self._normalize_structured_response(json.loads(response), placeholders)
        except (json.JSONDecodeError, KeyError) as e:
            logger.warning("Structured extraction failed: %s", e)
            return {name: [] for name in placeholders}

    def _normalize_structured_response(
        self, response: dict[str, Any], placeholders: dict[str, str]
    ) -> dict[str, list[dict[str, any]]]:
        """Normalize LLM response to ensure consistent format.

        Handles both old format (list of strings) and new format (list of {text, doc}).
        """
        result = {}
        for name in placeholders:
            items = response.get(name, [])
            normalized = []
            for item in items:
                if isinstance(item, str):
                    # Old format - just text, no doc attribution
                    normalized.append({"text": item, "doc": 0})
                elif isinstance(item, dict) and "text" in item:
                    # New format with doc attribution
                    normalized.append({"text": item["text"], "doc": item.get("doc", 0)})
            result[name] = normalized
        return result

    def _build_structured_extraction_prompt(
        self,
        question: str,
        template: str,
        placeholders: dict[str, str],
        documents: list[str],
    ) -> str:
        """Build prompt for structured extraction with document attribution."""
        placeholder_spec = "\n".join(f"- {name}: {hint}" for name, hint in placeholders.items())
        docs_text = "\n\n---\n\n".join(f"[Document {i}]\n{doc}" for i, doc in enumerate(documents))

        return load_prompt(
            "extraction/structured",
            question=question,
            template=template,
            placeholder_spec=placeholder_spec,
            docs_text=docs_text,
        )

    def generate_template(  # noqa: PLR0913, PLR0917  # public signature
        self,
        question: str,
        spans: list[str],
        citation_count: int,
        use_per_fact: bool = True,  # noqa: FBT001, FBT002  # public signature
        template_preview_chars: int | None = 100,
        preserve_span_newlines: bool = False,  # noqa: FBT001, FBT002  # public signature
        template_prompt: str | None = None,
        system_prompt: str | None = None,
    ) -> str:
        """Generate a contextual template for the given question and spans.

        :param question: The user's question
        :param spans: List of spans that will fill the template
        :param citation_count: Number of citation-only spans
        :param use_per_fact: Whether to use per-span placeholders
        :param template_preview_chars: Max chars per span shown to the template LLM.
            None = no truncation. Default 100 reproduces pre-0.2.8 behaviour.
        :param preserve_span_newlines: Keep newlines in spans shown to the template
            LLM. Default False collapses them (pre-0.2.8 behaviour).
        :param template_prompt: Custom Jinja2 prompt string. Variables available:
            question, n_spans, citation_count, spans_block (per-fact) or
            span_preview (aggregate). None uses the bundled prompt.
        :param system_prompt: System message for the template-generation call.
        :return: Generated template string
        """
        prompt = self._template_prompt(
            question,
            spans,
            citation_count,
            use_per_fact,
            template_preview_chars,
            preserve_span_newlines,
            template_prompt,
        )
        try:
            return self.complete(prompt, temperature=self.temperature, system_prompt=system_prompt)
        except Exception as e:  # noqa: BLE001  # any LLM failure falls back to the default template
            logger.error("Template generation failed: %s", e)  # noqa: TRY400  # a traceback can carry customer text
            return self._fallback_template(citation_count > 0)

    async def generate_template_async(  # noqa: PLR0913, PLR0917  # public signature
        self,
        question: str,
        spans: list[str],
        citation_count: int,
        use_per_fact: bool = True,  # noqa: FBT001, FBT002  # public signature
        template_preview_chars: int | None = 100,
        preserve_span_newlines: bool = False,  # noqa: FBT001, FBT002  # public signature
        template_prompt: str | None = None,
        system_prompt: str | None = None,
    ) -> str:
        """Async template generation. See generate_template for parameter docs."""
        prompt = self._template_prompt(
            question,
            spans,
            citation_count,
            use_per_fact,
            template_preview_chars,
            preserve_span_newlines,
            template_prompt,
        )
        try:
            return await self.complete_async(prompt, temperature=self.temperature, system_prompt=system_prompt)
        except Exception as e:  # noqa: BLE001  # any LLM failure falls back to the default template
            logger.error("Async template generation failed: %s", e)  # noqa: TRY400  # a traceback can carry customer text
            return self._fallback_template(citation_count > 0)

    def template_completion(  # noqa: PLR0913, PLR0917  # public signature
        self,
        question: str,
        spans: list[str],
        citation_count: int,
        use_per_fact: bool = True,  # noqa: FBT001, FBT002  # public signature
        template_preview_chars: int | None = 100,
        preserve_span_newlines: bool = False,  # noqa: FBT001, FBT002  # public signature
        template_prompt: str | None = None,
        system_prompt: str | None = None,
    ) -> Completion:
        """Like `generate_template`, but returns the completion and raises when the call fails.

        :raises UpstreamError: If the call fails or the reply is empty
        :raises ConfigurationError: If the service rejects the key, the model or the quota
        """
        prompt = self._template_prompt(
            question,
            spans,
            citation_count,
            use_per_fact,
            template_preview_chars,
            preserve_span_newlines,
            template_prompt,
        )
        return client_completion(self, prompt, temperature=self.temperature, system_prompt=system_prompt)

    async def template_completion_async(  # noqa: PLR0913, PLR0917  # public signature
        self,
        question: str,
        spans: list[str],
        citation_count: int,
        use_per_fact: bool = True,  # noqa: FBT001, FBT002  # public signature
        template_preview_chars: int | None = 100,
        preserve_span_newlines: bool = False,  # noqa: FBT001, FBT002  # public signature
        template_prompt: str | None = None,
        system_prompt: str | None = None,
    ) -> Completion:
        """Async version of `template_completion`."""
        prompt = self._template_prompt(
            question,
            spans,
            citation_count,
            use_per_fact,
            template_preview_chars,
            preserve_span_newlines,
            template_prompt,
        )
        return await client_completion_async(self, prompt, temperature=self.temperature, system_prompt=system_prompt)

    def _template_prompt(  # noqa: PLR0913, PLR0917  # mirrors the public template methods
        self,
        question: str,
        spans: list[str],
        citation_count: int,
        use_per_fact: bool,  # noqa: FBT001  # mirrors the public template methods
        template_preview_chars: int | None,
        preserve_span_newlines: bool,  # noqa: FBT001  # mirrors the public template methods
        template_prompt: str | None,
    ) -> str:
        build = (
            self._build_per_fact_template_prompt
            if use_per_fact and len(spans) <= _MAX_PER_FACT_SPANS
            else self._build_aggregate_template_prompt
        )
        return build(
            question,
            spans,
            citation_count,
            template_preview_chars,
            preserve_span_newlines,
            template_prompt,
        )

    def _build_extraction_prompt(self, question: str, documents: dict[str, str]) -> str:
        """Build the prompt for batch span extraction."""
        return load_prompt(
            "extraction/default",
            question=question,
            documents=json.dumps(documents, indent=2),
        )

    def _build_per_fact_template_prompt(  # noqa: PLR0913, PLR0917  # mirrors the public template methods
        self,
        question: str,
        spans: list[str],
        citation_count: int,
        template_preview_chars: int | None = 100,
        preserve_span_newlines: bool = False,  # noqa: FBT001, FBT002  # mirrors the public template methods
        template_prompt: str | None = None,
    ) -> str:
        span_lines = []
        for i, span in enumerate(spans, start=1):
            text = span if preserve_span_newlines else span.replace("\n", " ")
            text = text.strip()
            if template_preview_chars is not None:
                text = text[:template_preview_chars]
            suffix = "..." if template_preview_chars is not None else ""
            span_lines.append(f"{i}. {text}{suffix}")
        spans_block = "\n".join(span_lines)

        ctx = {
            "question": question,
            "n_spans": len(spans),
            "spans_block": spans_block,
            "citation_count": citation_count,
        }
        if template_prompt is not None:
            return render_prompt(template_prompt, **ctx)
        return load_prompt("template/per_fact", **ctx)

    def _build_aggregate_template_prompt(  # noqa: PLR0913, PLR0917  # mirrors the public template methods
        self,
        question: str,
        spans: list[str],
        citation_count: int,
        template_preview_chars: int | None = 100,
        preserve_span_newlines: bool = False,  # noqa: FBT001, FBT002  # mirrors the public template methods
        template_prompt: str | None = None,
    ) -> str:
        preview_chars = 50 if template_preview_chars is None else min(50, template_preview_chars)
        preview_spans = []
        for span in spans[:3]:
            text = span if preserve_span_newlines else span.replace("\n", " ")
            preview_spans.append(text[:preview_chars] + "...")
        span_preview = " | ".join(preview_spans)

        ctx = {
            "question": question,
            "n_spans": len(spans),
            "span_preview": span_preview,
            "citation_count": citation_count,
        }
        if template_prompt is not None:
            return render_prompt(template_prompt, **ctx)
        return load_prompt("template/aggregate", **ctx)

    def _fallback_template(self, has_citations: bool = False) -> str:  # noqa: FBT001, FBT002  # callers pass the flag positionally
        """Return a simple fallback template when generation fails."""
        return load_prompt("template/fallback", has_citations=has_citations)

    # Batch span extraction API (for compatibility)
    def extract_relevant_spans_batch(self, question: str, documents: dict[str, str]) -> dict[str, list[str]]:
        """Extract spans for a batch of documents. Same as `extract_spans`."""
        return self.extract_spans(question, documents)

    async def extract_relevant_spans_batch_async(
        self, question: str, documents: dict[str, str]
    ) -> dict[str, list[str]]:
        """Extract spans for a batch of documents asynchronously. Same as `extract_spans_async`."""
        return await self.extract_spans_async(question, documents)

    # Single-doc convenience
    def extract_relevant_spans(self, question: str, document_text: str) -> list[str]:
        """Extract spans from a single document."""
        result = self.extract_relevant_spans_batch(question, {"doc": document_text})
        return result.get("doc", [])

    async def extract_relevant_spans_async(self, question: str, document_text: str) -> list[str]:
        """Extract spans from a single document asynchronously."""
        result = await self.extract_relevant_spans_batch_async(question, {"doc": document_text})
        return result.get("doc", [])

    # Template generation (simple compatibility methods)
    def simple_complete(self, prompt: str) -> str:
        """Complete a prompt with the default settings."""
        return self.complete(prompt)

    async def simple_complete_async(self, prompt: str) -> str:
        """Complete a prompt asynchronously with the default settings."""
        return await self.complete_async(prompt)

__init__(model='gpt-4o-mini', temperature=0.7, api_base='https://api.openai.com/v1', api_key=None)

Initialize the LLM client.

:param model: The OpenAI model to use :param temperature: Default temperature for completions :param api_base: The base URL for the OpenAI API (can be used with custom models and with VLLM) :param api_key: Optional API key. Falls back to OPENAI_API_KEY when unset.

Source code in packages/core/verbatim_core/llm_client.py 
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
def __init__(
    self,
    model: str = "gpt-4o-mini",
    temperature: float = 0.7,
    api_base: str = "https://api.openai.com/v1",
    api_key: str | None = None,
) -> None:
    """Initialize the LLM client.

    :param model: The OpenAI model to use
    :param temperature: Default temperature for completions
    :param api_base: The base URL for the OpenAI API (can be used with custom models and with VLLM)
    :param api_key: Optional API key. Falls back to OPENAI_API_KEY when unset.
    """
    self.model = model
    self.temperature = temperature
    self.api_key = api_key or os.getenv("OPENAI_API_KEY") or "EMPTY"
    self.client = openai.OpenAI(base_url=api_base, api_key=self.api_key)

    self.async_client = openai.AsyncOpenAI(base_url=api_base, api_key=self.api_key)

complete(prompt, json_mode=False, temperature=None, system_prompt=None)

Complete text synchronously.

:param prompt: The prompt to send :param json_mode: Whether to request JSON output format :param temperature: Override default temperature :param system_prompt: Optional system message to prepend :return: The completion text :raises UpstreamError: If the call fails or the reply is empty :raises ConfigurationError: If the service rejects the key, the model or the quota

Source code in packages/core/verbatim_core/llm_client.py 
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
def complete(
    self,
    prompt: str,
    json_mode: bool = False,  # noqa: FBT001, FBT002  # public signature
    temperature: float | None = None,
    system_prompt: str | None = None,
) -> str:
    """Complete text synchronously.

    :param prompt: The prompt to send
    :param json_mode: Whether to request JSON output format
    :param temperature: Override default temperature
    :param system_prompt: Optional system message to prepend
    :return: The completion text
    :raises UpstreamError: If the call fails or the reply is empty
    :raises ConfigurationError: If the service rejects the key, the model or the quota
    """
    return self.completion(prompt, json_mode=json_mode, temperature=temperature, system_prompt=system_prompt).text

complete_async(prompt, json_mode=False, temperature=None, system_prompt=None) async

Asynchronous text completion. See complete.

Source code in packages/core/verbatim_core/llm_client.py 
269
270
271
272
273
274
275
276
277
278
279
280
async def complete_async(
    self,
    prompt: str,
    json_mode: bool = False,  # noqa: FBT001, FBT002  # public signature
    temperature: float | None = None,
    system_prompt: str | None = None,
) -> str:
    """Asynchronous text completion. See `complete`."""
    completion = await self.completion_async(
        prompt, json_mode=json_mode, temperature=temperature, system_prompt=system_prompt
    )
    return completion.text

completion(prompt, *, json_mode=False, temperature=None, system_prompt=None)

Like complete, and also returns the model name and the token counts the provider sent.

Source code in packages/core/verbatim_core/llm_client.py 
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
def completion(
    self,
    prompt: str,
    *,
    json_mode: bool = False,
    temperature: float | None = None,
    system_prompt: str | None = None,
) -> Completion:
    """Like `complete`, and also returns the model name and the token counts the provider sent."""
    request = self._request(prompt, json_mode, temperature, system_prompt)
    try:
        response = self.client.chat.completions.create(**request)
    except APIError as e:
        raise _translate_error(e) from e
    return _completion(response)

completion_async(prompt, *, json_mode=False, temperature=None, system_prompt=None) async

Asynchronous version of completion.

Source code in packages/core/verbatim_core/llm_client.py 
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
async def completion_async(
    self,
    prompt: str,
    *,
    json_mode: bool = False,
    temperature: float | None = None,
    system_prompt: str | None = None,
) -> Completion:
    """Asynchronous version of `completion`."""
    request = self._request(prompt, json_mode, temperature, system_prompt)
    try:
        response = await self.async_client.chat.completions.create(**request)
    except APIError as e:
        raise _translate_error(e) from e
    return _completion(response)

extract_spans(question, documents)

Specialized method for span extraction from documents.

:param question: The user's question :param documents: Dictionary mapping doc IDs to document text :return: Dictionary mapping doc IDs to lists of extracted spans

Source code in packages/core/verbatim_core/llm_client.py 
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
def extract_spans(self, question: str, documents: dict[str, str]) -> dict[str, list[str]]:
    """Specialized method for span extraction from documents.

    :param question: The user's question
    :param documents: Dictionary mapping doc IDs to document text
    :return: Dictionary mapping doc IDs to lists of extracted spans
    """
    prompt = self._build_extraction_prompt(question, documents)
    try:
        response = self.complete(prompt, json_mode=True)
        return json.loads(response)
    except (json.JSONDecodeError, KeyError) as e:
        logger.warning("Span extraction failed: %s", e)
        # Return empty results for all documents on failure
        return {doc_id: [] for doc_id in documents}

extract_spans_async(question, documents) async

Async span extraction from documents.

:param question: The user's question :param documents: Dictionary mapping doc IDs to document text :return: Dictionary mapping doc IDs to lists of extracted spans

Source code in packages/core/verbatim_core/llm_client.py 
350
351
352
353
354
355
356
357
358
359
360
361
362
363
async def extract_spans_async(self, question: str, documents: dict[str, str]) -> dict[str, list[str]]:
    """Async span extraction from documents.

    :param question: The user's question
    :param documents: Dictionary mapping doc IDs to document text
    :return: Dictionary mapping doc IDs to lists of extracted spans
    """
    prompt = self._build_extraction_prompt(question, documents)
    try:
        response = await self.complete_async(prompt, json_mode=True)
        return json.loads(response)
    except (json.JSONDecodeError, KeyError) as e:
        logger.warning("Async span extraction failed: %s", e)
        return {doc_id: [] for doc_id in documents}

extract_structured(question, template, placeholders, documents)

Extract spans organized by template placeholders with document attribution.

:param question: The user's question :param template: Template with placeholders like [METHODOLOGY] :param placeholders: Dict mapping placeholder names to hints :param documents: List of document texts :return: Dict mapping placeholder names to lists of {text, doc} objects

Source code in packages/core/verbatim_core/llm_client.py 
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
def extract_structured(
    self,
    question: str,
    template: str,
    placeholders: dict[str, str],
    documents: list[str],
) -> dict[str, list[dict[str, any]]]:
    """Extract spans organized by template placeholders with document attribution.

    :param question: The user's question
    :param template: Template with placeholders like [METHODOLOGY]
    :param placeholders: Dict mapping placeholder names to hints
    :param documents: List of document texts
    :return: Dict mapping placeholder names to lists of {text, doc} objects
    """
    prompt = self._build_structured_extraction_prompt(question, template, placeholders, documents)
    try:
        response = self.complete(prompt, json_mode=True)
        return self._normalize_structured_response(json.loads(response), placeholders)
    except (json.JSONDecodeError, KeyError) as e:
        logger.warning("Structured extraction failed: %s", e)
        return {name: [] for name in placeholders}

extract_structured_async(question, template, placeholders, documents) async

Async version of structured extraction with document attribution.

:param question: The user's question :param template: Template with placeholders like [METHODOLOGY] :param placeholders: Dict mapping placeholder names to hints :param documents: List of document texts :return: Dict mapping placeholder names to lists of {text, doc} objects

Source code in packages/core/verbatim_core/llm_client.py 
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
async def extract_structured_async(
    self,
    question: str,
    template: str,
    placeholders: dict[str, str],
    documents: list[str],
) -> dict[str, list[dict[str, any]]]:
    """Async version of structured extraction with document attribution.

    :param question: The user's question
    :param template: Template with placeholders like [METHODOLOGY]
    :param placeholders: Dict mapping placeholder names to hints
    :param documents: List of document texts
    :return: Dict mapping placeholder names to lists of {text, doc} objects
    """
    prompt = self._build_structured_extraction_prompt(question, template, placeholders, documents)
    try:
        response = await self.complete_async(prompt, json_mode=True)
        return self._normalize_structured_response(json.loads(response), placeholders)
    except (json.JSONDecodeError, KeyError) as e:
        logger.warning("Structured extraction failed: %s", e)
        return {name: [] for name in placeholders}

generate_template(question, spans, citation_count, use_per_fact=True, template_preview_chars=100, preserve_span_newlines=False, template_prompt=None, system_prompt=None)

Generate a contextual template for the given question and spans.

:param question: The user's question :param spans: List of spans that will fill the template :param citation_count: Number of citation-only spans :param use_per_fact: Whether to use per-span placeholders :param template_preview_chars: Max chars per span shown to the template LLM. None = no truncation. Default 100 reproduces pre-0.2.8 behaviour. :param preserve_span_newlines: Keep newlines in spans shown to the template LLM. Default False collapses them (pre-0.2.8 behaviour). :param template_prompt: Custom Jinja2 prompt string. Variables available: question, n_spans, citation_count, spans_block (per-fact) or span_preview (aggregate). None uses the bundled prompt. :param system_prompt: System message for the template-generation call. :return: Generated template string

Source code in packages/core/verbatim_core/llm_client.py 
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
def generate_template(  # noqa: PLR0913, PLR0917  # public signature
    self,
    question: str,
    spans: list[str],
    citation_count: int,
    use_per_fact: bool = True,  # noqa: FBT001, FBT002  # public signature
    template_preview_chars: int | None = 100,
    preserve_span_newlines: bool = False,  # noqa: FBT001, FBT002  # public signature
    template_prompt: str | None = None,
    system_prompt: str | None = None,
) -> str:
    """Generate a contextual template for the given question and spans.

    :param question: The user's question
    :param spans: List of spans that will fill the template
    :param citation_count: Number of citation-only spans
    :param use_per_fact: Whether to use per-span placeholders
    :param template_preview_chars: Max chars per span shown to the template LLM.
        None = no truncation. Default 100 reproduces pre-0.2.8 behaviour.
    :param preserve_span_newlines: Keep newlines in spans shown to the template
        LLM. Default False collapses them (pre-0.2.8 behaviour).
    :param template_prompt: Custom Jinja2 prompt string. Variables available:
        question, n_spans, citation_count, spans_block (per-fact) or
        span_preview (aggregate). None uses the bundled prompt.
    :param system_prompt: System message for the template-generation call.
    :return: Generated template string
    """
    prompt = self._template_prompt(
        question,
        spans,
        citation_count,
        use_per_fact,
        template_preview_chars,
        preserve_span_newlines,
        template_prompt,
    )
    try:
        return self.complete(prompt, temperature=self.temperature, system_prompt=system_prompt)
    except Exception as e:  # noqa: BLE001  # any LLM failure falls back to the default template
        logger.error("Template generation failed: %s", e)  # noqa: TRY400  # a traceback can carry customer text
        return self._fallback_template(citation_count > 0)

generate_template_async(question, spans, citation_count, use_per_fact=True, template_preview_chars=100, preserve_span_newlines=False, template_prompt=None, system_prompt=None) async

Async template generation. See generate_template for parameter docs.

Source code in packages/core/verbatim_core/llm_client.py 
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
async def generate_template_async(  # noqa: PLR0913, PLR0917  # public signature
    self,
    question: str,
    spans: list[str],
    citation_count: int,
    use_per_fact: bool = True,  # noqa: FBT001, FBT002  # public signature
    template_preview_chars: int | None = 100,
    preserve_span_newlines: bool = False,  # noqa: FBT001, FBT002  # public signature
    template_prompt: str | None = None,
    system_prompt: str | None = None,
) -> str:
    """Async template generation. See generate_template for parameter docs."""
    prompt = self._template_prompt(
        question,
        spans,
        citation_count,
        use_per_fact,
        template_preview_chars,
        preserve_span_newlines,
        template_prompt,
    )
    try:
        return await self.complete_async(prompt, temperature=self.temperature, system_prompt=system_prompt)
    except Exception as e:  # noqa: BLE001  # any LLM failure falls back to the default template
        logger.error("Async template generation failed: %s", e)  # noqa: TRY400  # a traceback can carry customer text
        return self._fallback_template(citation_count > 0)

template_completion(question, spans, citation_count, use_per_fact=True, template_preview_chars=100, preserve_span_newlines=False, template_prompt=None, system_prompt=None)

Like generate_template, but returns the completion and raises when the call fails.

:raises UpstreamError: If the call fails or the reply is empty :raises ConfigurationError: If the service rejects the key, the model or the quota

Source code in packages/core/verbatim_core/llm_client.py 
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
def template_completion(  # noqa: PLR0913, PLR0917  # public signature
    self,
    question: str,
    spans: list[str],
    citation_count: int,
    use_per_fact: bool = True,  # noqa: FBT001, FBT002  # public signature
    template_preview_chars: int | None = 100,
    preserve_span_newlines: bool = False,  # noqa: FBT001, FBT002  # public signature
    template_prompt: str | None = None,
    system_prompt: str | None = None,
) -> Completion:
    """Like `generate_template`, but returns the completion and raises when the call fails.

    :raises UpstreamError: If the call fails or the reply is empty
    :raises ConfigurationError: If the service rejects the key, the model or the quota
    """
    prompt = self._template_prompt(
        question,
        spans,
        citation_count,
        use_per_fact,
        template_preview_chars,
        preserve_span_newlines,
        template_prompt,
    )
    return client_completion(self, prompt, temperature=self.temperature, system_prompt=system_prompt)

template_completion_async(question, spans, citation_count, use_per_fact=True, template_preview_chars=100, preserve_span_newlines=False, template_prompt=None, system_prompt=None) async

Async version of template_completion.

Source code in packages/core/verbatim_core/llm_client.py 
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
async def template_completion_async(  # noqa: PLR0913, PLR0917  # public signature
    self,
    question: str,
    spans: list[str],
    citation_count: int,
    use_per_fact: bool = True,  # noqa: FBT001, FBT002  # public signature
    template_preview_chars: int | None = 100,
    preserve_span_newlines: bool = False,  # noqa: FBT001, FBT002  # public signature
    template_prompt: str | None = None,
    system_prompt: str | None = None,
) -> Completion:
    """Async version of `template_completion`."""
    prompt = self._template_prompt(
        question,
        spans,
        citation_count,
        use_per_fact,
        template_preview_chars,
        preserve_span_newlines,
        template_prompt,
    )
    return await client_completion_async(self, prompt, temperature=self.temperature, system_prompt=system_prompt)

extract_relevant_spans_batch(question, documents)

Extract spans for a batch of documents. Same as extract_spans.

Source code in packages/core/verbatim_core/llm_client.py 
662
663
664
def extract_relevant_spans_batch(self, question: str, documents: dict[str, str]) -> dict[str, list[str]]:
    """Extract spans for a batch of documents. Same as `extract_spans`."""
    return self.extract_spans(question, documents)

extract_relevant_spans_batch_async(question, documents) async

Extract spans for a batch of documents asynchronously. Same as extract_spans_async.

Source code in packages/core/verbatim_core/llm_client.py 
666
667
668
669
670
async def extract_relevant_spans_batch_async(
    self, question: str, documents: dict[str, str]
) -> dict[str, list[str]]:
    """Extract spans for a batch of documents asynchronously. Same as `extract_spans_async`."""
    return await self.extract_spans_async(question, documents)

extract_relevant_spans(question, document_text)

Extract spans from a single document.

Source code in packages/core/verbatim_core/llm_client.py 
673
674
675
676
def extract_relevant_spans(self, question: str, document_text: str) -> list[str]:
    """Extract spans from a single document."""
    result = self.extract_relevant_spans_batch(question, {"doc": document_text})
    return result.get("doc", [])

extract_relevant_spans_async(question, document_text) async

Extract spans from a single document asynchronously.

Source code in packages/core/verbatim_core/llm_client.py 
678
679
680
681
async def extract_relevant_spans_async(self, question: str, document_text: str) -> list[str]:
    """Extract spans from a single document asynchronously."""
    result = await self.extract_relevant_spans_batch_async(question, {"doc": document_text})
    return result.get("doc", [])

simple_complete(prompt)

Complete a prompt with the default settings.

Source code in packages/core/verbatim_core/llm_client.py 
684
685
686
def simple_complete(self, prompt: str) -> str:
    """Complete a prompt with the default settings."""
    return self.complete(prompt)

simple_complete_async(prompt) async

Complete a prompt asynchronously with the default settings.

Source code in packages/core/verbatim_core/llm_client.py 
688
689
690
async def simple_complete_async(self, prompt: str) -> str:
    """Complete a prompt asynchronously with the default settings."""
    return await self.complete_async(prompt)

ResponseBuilder

verbatim_core.response_builder.ResponseBuilder

Builds structured query responses with highlights and citations.

Takes search results and extracted spans and creates a complete QueryResponse object with proper document highlighting and citation tracking.

Source code in packages/core/verbatim_core/response_builder.py 
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 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
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
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
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
class ResponseBuilder:
    """Builds structured query responses with highlights and citations.

    Takes search results and extracted spans and creates a complete QueryResponse
    object with proper document highlighting and citation tracking.
    """

    def __init__(self) -> None:
        """Initialize the response builder."""

    def build_response(
        self,
        question: str,
        answer: str,
        search_results: list[Any],
        relevant_spans: dict[str, list[str]],
        display_span_count: int | None = None,
    ) -> QueryResponse:
        """Build a complete QueryResponse from components.

        :param question: The original question
        :param answer: The generated answer text
        :param search_results: List of search results from the index
        :param relevant_spans: Dict mapping document text to extracted spans
        :param display_span_count: Number of spans to display vs cite-only
        :return: Complete QueryResponse object
        """
        documents_with_highlights = []
        all_citations = []

        current_citation_number = 1

        for result_index, result in enumerate(search_results):
            result_content = getattr(result, "text", "")
            highlights = []

            # Find spans for this document
            spans_for_doc = relevant_spans.get(result_content, [])

            if spans_for_doc:
                # Create highlights for this document
                highlights = self._create_highlights(result_content, spans_for_doc)

                # Create citations for each highlight
                for highlight_index, highlight in enumerate(highlights):
                    # Determine if this should be a display citation or reference-only
                    is_display = display_span_count is None or current_citation_number <= display_span_count

                    all_citations.append(
                        Citation(
                            text=highlight.text,
                            doc_index=result_index,
                            highlight_index=highlight_index,
                            number=current_citation_number,
                            type="display" if is_display else "reference",
                        )
                    )
                    current_citation_number += 1

            # Add document with highlights
            documents_with_highlights.append(
                DocumentWithHighlights(
                    content=result_content,
                    highlights=highlights,
                    title=getattr(result, "title", "") or result.metadata.get("title", ""),
                    source=getattr(result, "source", "") or result.metadata.get("source", ""),
                    metadata=getattr(result, "metadata", {}),
                )
            )

        # Create structured answer with citations
        structured_answer = StructuredAnswer(text=answer, citations=all_citations)

        return QueryResponse(
            question=question,
            answer=answer,
            structured_answer=structured_answer,
            documents=documents_with_highlights,
        )

    def build_from_answer(self, question: str, answer: Answer, chunks: Sequence[Chunk]) -> QueryResponse:
        """Build a QueryResponse from an answer and the chunks it was written from.

        Each citation's span is highlighted at its offsets, without searching for its text. When
        it overlaps a span cited before it in the same chunk, the earlier one keeps the highlight,
        and the later one gets neither a highlight nor a citation in the response.

        :param question: The original question
        :param answer: The answer, whose citations point at spans of the chunks
        :param chunks: The chunks the answer was written from, in the order of the documents
        :return: Complete QueryResponse object
        :raises InvalidInputError: If two chunks share an id, or a citation points outside them
        """
        documents, citations = _documents_and_citations(answer.citations, chunks)
        return QueryResponse(
            question=question,
            answer=answer.text,
            structured_answer=StructuredAnswer(text=answer.text, citations=citations),
            documents=documents,
            warnings=answer.warnings,
        )

    def _create_highlights(self, doc_content: str, spans: list[str]) -> list[Highlight]:
        """Create highlight objects for spans in document content.

        Uses sophisticated overlap detection to avoid conflicting highlights.

        :param doc_content: The full document text
        :param spans: List of text spans to highlight
        :return: List of Highlight objects
        """
        highlights: list[Highlight] = []
        highlighted_regions: set[tuple[int, int]] = set()

        for span in spans:
            # Find all occurrences of this span in the document
            start = 0
            while True:
                start = doc_content.find(span, start)
                if start == -1:
                    break

                end = start + len(span)

                # Check for overlap with existing highlights
                if not self._has_overlap(start, end, highlighted_regions):
                    highlights.append(Highlight(text=span, start=start, end=end))
                    highlighted_regions.add((start, end))

                # Continue searching from the end of current match
                start = end

        return highlights

    def _has_overlap(self, start: int, end: int, regions: set[tuple[int, int]]) -> bool:
        """Check if a text region overlaps with existing highlighted regions.

        :param start: Start position of new region
        :param end: End position of new region
        :param regions: Set of existing (start, end) tuples
        :return: True if there's overlap, False otherwise
        """
        return any(start < region_end and end > region_start for region_start, region_end in regions)

    def clean_answer(self, answer: str) -> str:
        """Clean up generated answer text.

        Removes common formatting issues and artifacts from LLM generation.

        :param answer: Raw answer text from generation
        :return: Cleaned answer text
        """
        if not answer:
            return ""

        # Remove surrounding quotes if present
        if (answer.startswith('"') and answer.endswith('"')) or (answer.startswith("'") and answer.endswith("'")):
            answer = answer[1:-1]

        # Convert literal newlines
        answer = answer.replace("\\n", "\n")

        # Clean up multiple spaces
        answer = re.sub(r" {2,}", " ", answer)

        # Clean up multiple newlines (but preserve paragraph breaks)
        answer = re.sub(r"\n{3,}", "\n\n", answer)

        return answer.strip()

__init__()

Initialize the response builder.

Source code in packages/core/verbatim_core/response_builder.py 
31
32
def __init__(self) -> None:
    """Initialize the response builder."""

build_response(question, answer, search_results, relevant_spans, display_span_count=None)

Build a complete QueryResponse from components.

:param question: The original question :param answer: The generated answer text :param search_results: List of search results from the index :param relevant_spans: Dict mapping document text to extracted spans :param display_span_count: Number of spans to display vs cite-only :return: Complete QueryResponse object

Source code in packages/core/verbatim_core/response_builder.py 
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 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
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def build_response(
    self,
    question: str,
    answer: str,
    search_results: list[Any],
    relevant_spans: dict[str, list[str]],
    display_span_count: int | None = None,
) -> QueryResponse:
    """Build a complete QueryResponse from components.

    :param question: The original question
    :param answer: The generated answer text
    :param search_results: List of search results from the index
    :param relevant_spans: Dict mapping document text to extracted spans
    :param display_span_count: Number of spans to display vs cite-only
    :return: Complete QueryResponse object
    """
    documents_with_highlights = []
    all_citations = []

    current_citation_number = 1

    for result_index, result in enumerate(search_results):
        result_content = getattr(result, "text", "")
        highlights = []

        # Find spans for this document
        spans_for_doc = relevant_spans.get(result_content, [])

        if spans_for_doc:
            # Create highlights for this document
            highlights = self._create_highlights(result_content, spans_for_doc)

            # Create citations for each highlight
            for highlight_index, highlight in enumerate(highlights):
                # Determine if this should be a display citation or reference-only
                is_display = display_span_count is None or current_citation_number <= display_span_count

                all_citations.append(
                    Citation(
                        text=highlight.text,
                        doc_index=result_index,
                        highlight_index=highlight_index,
                        number=current_citation_number,
                        type="display" if is_display else "reference",
                    )
                )
                current_citation_number += 1

        # Add document with highlights
        documents_with_highlights.append(
            DocumentWithHighlights(
                content=result_content,
                highlights=highlights,
                title=getattr(result, "title", "") or result.metadata.get("title", ""),
                source=getattr(result, "source", "") or result.metadata.get("source", ""),
                metadata=getattr(result, "metadata", {}),
            )
        )

    # Create structured answer with citations
    structured_answer = StructuredAnswer(text=answer, citations=all_citations)

    return QueryResponse(
        question=question,
        answer=answer,
        structured_answer=structured_answer,
        documents=documents_with_highlights,
    )

build_from_answer(question, answer, chunks)

Build a QueryResponse from an answer and the chunks it was written from.

Each citation's span is highlighted at its offsets, without searching for its text. When it overlaps a span cited before it in the same chunk, the earlier one keeps the highlight, and the later one gets neither a highlight nor a citation in the response.

:param question: The original question :param answer: The answer, whose citations point at spans of the chunks :param chunks: The chunks the answer was written from, in the order of the documents :return: Complete QueryResponse object :raises InvalidInputError: If two chunks share an id, or a citation points outside them

Source code in packages/core/verbatim_core/response_builder.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def build_from_answer(self, question: str, answer: Answer, chunks: Sequence[Chunk]) -> QueryResponse:
    """Build a QueryResponse from an answer and the chunks it was written from.

    Each citation's span is highlighted at its offsets, without searching for its text. When
    it overlaps a span cited before it in the same chunk, the earlier one keeps the highlight,
    and the later one gets neither a highlight nor a citation in the response.

    :param question: The original question
    :param answer: The answer, whose citations point at spans of the chunks
    :param chunks: The chunks the answer was written from, in the order of the documents
    :return: Complete QueryResponse object
    :raises InvalidInputError: If two chunks share an id, or a citation points outside them
    """
    documents, citations = _documents_and_citations(answer.citations, chunks)
    return QueryResponse(
        question=question,
        answer=answer.text,
        structured_answer=StructuredAnswer(text=answer.text, citations=citations),
        documents=documents,
        warnings=answer.warnings,
    )

clean_answer(answer)

Clean up generated answer text.

Removes common formatting issues and artifacts from LLM generation.

:param answer: Raw answer text from generation :return: Cleaned answer text

Source code in packages/core/verbatim_core/response_builder.py 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
def clean_answer(self, answer: str) -> str:
    """Clean up generated answer text.

    Removes common formatting issues and artifacts from LLM generation.

    :param answer: Raw answer text from generation
    :return: Cleaned answer text
    """
    if not answer:
        return ""

    # Remove surrounding quotes if present
    if (answer.startswith('"') and answer.endswith('"')) or (answer.startswith("'") and answer.endswith("'")):
        answer = answer[1:-1]

    # Convert literal newlines
    answer = answer.replace("\\n", "\n")

    # Clean up multiple spaces
    answer = re.sub(r" {2,}", " ", answer)

    # Clean up multiple newlines (but preserve paragraph breaks)
    answer = re.sub(r"\n{3,}", "\n\n", answer)

    return answer.strip()

TemplateManager

verbatim_core.templates.manager.TemplateManager

Template manager with strategy pattern and mode selection.

Manages different template strategies and provides a unified interface for template generation and filling. Supports persistence of configuration across sessions.

Its settings, config, are the default settings of a Transformer built with it.

Source code in packages/core/verbatim_core/templates/manager.py 
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 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
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
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
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
class TemplateManager:
    """Template manager with strategy pattern and mode selection.

    Manages different template strategies and provides a unified interface
    for template generation and filling. Supports persistence of configuration
    across sessions.

    Its settings, `config`, are the default settings of a `Transformer` built with it.
    """

    def __init__(
        self,
        llm_client: LLMClient | None = None,
        default_mode: str = "static",
        rag_system: Any = None,  # noqa: ANN401  # duck-typed RAG system; avoids circular import
    ) -> None:
        """Initialize template manager.

        :param llm_client: Optional LLM client for contextual and random modes
        :param default_mode: Default template mode ("static", "contextual", "random", "question_specific", "structured")
        :param rag_system: Optional RAG system for structured mode
        """
        self.llm_client = llm_client
        self.rag_system = rag_system

        # Initialize strategies
        self.strategies: dict[str, TemplateStrategy | None] = {
            "static": StaticTemplate(),
            "contextual": ContextualTemplate(llm_client) if llm_client else None,
            "random": RandomTemplate(llm_client=llm_client),
            "question_specific": QuestionSpecificTemplate(),
            "structured": StructuredTemplate(rag_system=rag_system),
        }

        # Validate initial mode
        mode = default_mode if default_mode in self.strategies else "static"
        if self.strategies[mode] is None:
            warn_contextual_without_client()
            mode = "static"

        self.config = TransformConfig.model_validate({"template_mode": mode})

    @property
    def current_mode(self) -> str:
        """The template mode of `config`."""
        return self.config.template_mode

    @current_mode.setter
    def current_mode(self, mode: str) -> None:
        warn_deprecated("Assigning TemplateManager.current_mode", use=_mode_replacement(mode))
        self._switch_mode(mode)

    @property
    def citation_mode(self) -> str:
        """The citation mode of `config`."""
        return self.config.citation_mode

    @citation_mode.setter
    def citation_mode(self, mode: str) -> None:
        warn_deprecated("Assigning TemplateManager.citation_mode", use="TransformConfig(citation_mode=...)")
        self._set_citations(citation_mode=mode)

    def set_mode(self, mode: str) -> bool:
        """Switch to a different template mode.

        Deprecated: pass `TransformConfig(template_mode=...)` to the `Transformer` or its call.

        :param mode: Template mode to switch to
        :return: True if mode was switched successfully
        """
        warn_deprecated("TemplateManager.set_mode", use=_mode_replacement(mode))
        return self._switch_mode(mode)

    def _switch_mode(self, mode: str) -> bool:
        if mode not in self.strategies:
            logger.warning(
                "Unknown template mode %s. The template mode stays %s.",
                mode,
                self.config.template_mode,
            )
            return False

        if self.strategies[mode] is None:
            warn_contextual_without_client()
            return False

        self._replace_config(template_mode=mode)
        return True

    def _replace_config(self, **settings: object) -> None:
        """Replace settings of `config`.

        :raises ConfigurationError: If a setting is invalid
        """
        # Only the settings set before stay set, so a call can tell them from the defaults.
        self.config = _config_from_settings(TransformConfig, {**self.config.model_dump(exclude_unset=True), **settings})

    def get_current_mode(self) -> str:
        """Get the current template mode.

        :return: Current mode name
        """
        return self.config.template_mode

    def strategy(self, mode: str) -> TemplateStrategy | None:
        """Return the strategy of a template mode, without switching to it.

        :param mode: Template mode
        :return: The strategy, or None if the mode needs an LLM client the manager doesn't have
        :raises KeyError: If there is no such mode
        """
        return self.strategies[mode]

    def get_available_modes(self) -> list[str]:
        """Get list of available template modes.

        :return: List of mode names that are available
        """
        return [mode for mode, strategy in self.strategies.items() if strategy is not None]

    def process(
        self,
        question: str,
        display_spans: list[dict[str, Any]],
        citation_spans: list[dict[str, Any]],
    ) -> str:
        """Generate and fill a template in one operation.

        :param question: The user's question
        :param display_spans: Spans to display with full text
        :param citation_spans: Spans for citation reference only
        :return: Completed response text
        """
        all_spans, citation_count = self._get_template_inputs(display_spans, citation_spans)

        # Generate template
        strategy = self._current_strategy()
        template = strategy.generate(question, all_spans, citation_count)

        # Fill template
        return strategy.fill(template, display_spans, citation_spans)

    async def process_async(
        self,
        question: str,
        display_spans: list[dict[str, Any]],
        citation_spans: list[dict[str, Any]],
    ) -> str:
        """Async version of process for contextual templates.

        :param question: The user's question
        :param display_spans: Spans to display with full text
        :param citation_spans: Spans for citation reference only
        :return: Completed response text
        """
        all_spans, citation_count = self._get_template_inputs(display_spans, citation_spans)

        strategy = self._current_strategy()

        # Use async generation if available
        if hasattr(strategy, "generate_async") and self.config.template_mode == "contextual":
            template = await cast("ContextualTemplate", strategy).generate_async(question, all_spans, citation_count)
        else:
            template = strategy.generate(question, all_spans, citation_count)

        return strategy.fill(template, display_spans, citation_spans)

    def _current_strategy(self) -> TemplateStrategy:
        """Return the strategy of the template mode of `config`.

        Switching modes skips a mode without a strategy, so the mode of `config` has one.
        """
        return cast("TemplateStrategy", self.strategies[self.config.template_mode])

    @staticmethod
    def _get_template_inputs(
        display_spans: list[dict[str, Any]],
        citation_spans: list[dict[str, Any]],
    ) -> tuple[list[str], int]:
        has_linked_citations = any(span.get("citation_ids") for span in display_spans)
        if has_linked_citations:
            return [span["text"] for span in display_spans], 0

        return [span["text"] for span in display_spans + citation_spans], len(citation_spans)

    def get_template(self, question: str = "", spans: list[str] | None = None, citation_count: int = 0) -> str:
        """Generate a template without filling it.

        :param question: The user's question
        :param spans: List of spans that will fill the template
        :param citation_count: Number of citation-only spans
        :return: Template string with placeholders
        """
        spans = spans or []
        strategy = self._current_strategy()
        return strategy.generate(question, spans, citation_count)

    def fill_template(
        self,
        template: str,
        display_spans: list[dict[str, Any]],
        citation_spans: list[dict[str, Any]],
    ) -> str:
        """Fill a template with content.

        :param template: Template string with placeholders
        :param display_spans: Spans to display with full text
        :param citation_spans: Spans for citation reference only
        :return: Filled template
        """
        strategy = self._current_strategy()
        return strategy.fill(template, display_spans, citation_spans)

    def save(self, filepath: str) -> None:
        """Save all template configurations to file.

        :param filepath: Path to save configuration
        """
        data = {
            "current_mode": self.config.template_mode,
            "citation_mode": self.config.citation_mode,
            "citation_format": self.config.citation_format,
            "strategies": {},
        }

        # Save state for each available strategy
        for mode, strategy in self.strategies.items():
            if strategy is not None:
                data["strategies"][mode] = strategy.save_state()

        # Create directory if it doesn't exist
        os.makedirs(os.path.dirname(filepath), exist_ok=True)  # noqa: PTH103, PTH120  # for a bare file name, Path.parent is "." where dirname is ""

        with Path(filepath).open("w") as f:
            json.dump(data, f, indent=2)

    def load(self, filepath: str) -> bool:
        """Load template configurations from file.

        :param filepath: Path to load configuration from
        :return: True if loaded successfully
        """
        if not os.path.exists(filepath):  # noqa: PTH110  # unlike Path.exists, an unreadable path is "not found"
            logger.warning("Template config file not found: %s", filepath)
            return False

        try:
            with Path(filepath).open() as f:
                data = json.load(f)

            # Load mode
            if "current_mode" in data:
                mode = data["current_mode"]
                if self.strategies.get(mode) is not None:
                    self._replace_config(template_mode=mode)

            # Load strategy states
            strategies_data = data.get("strategies", {})
            for mode, state in strategies_data.items():
                strategy = self.strategies.get(mode)
                if strategy is not None:
                    # A saved state of the wrong shape doesn't stop the other strategies.
                    try:
                        strategy.load_state(state)
                    except (KeyError, TypeError, ValueError):
                        logger.warning(
                            "Could not load the saved state of the %s templates",
                            mode,
                            exc_info=True,
                        )

            # Before 0.3.0 only the contextual templates saved the citation format.
            citations = {key: data[key] for key in ("citation_mode", "citation_format") if key in data}
            contextual_state = strategies_data.get("contextual") or {}
            if "citation_format" not in citations and "citation_format" in contextual_state:
                citations["citation_format"] = contextual_state["citation_format"]
            if citations:
                self._set_citations(**citations)

        # A file that can't be read, or isn't JSON
        except (OSError, ValueError):
            logger.warning("Could not load the template config %s", filepath, exc_info=True)
            return False
        else:
            return True

    def info(self) -> dict[str, Any]:
        """Get current template manager state information.

        :return: Dictionary with current state info
        """
        info_data = {
            "current_mode": self.config.template_mode,
            "available_modes": self.get_available_modes(),
            "has_llm_client": self.llm_client is not None,
        }

        # Add mode-specific info
        if self.config.template_mode == "random":
            random_strategy = self.strategies["random"]
            if hasattr(random_strategy, "get_template_count"):
                info_data["random_template_count"] = cast("RandomTemplate", random_strategy).get_template_count()

        return info_data

    # Convenience methods for specific modes

    def use_static_mode(self, template: str | None = None) -> None:
        """Switch to static mode with optional custom template.

        :param template: Optional custom template
        """
        if template:
            static_strategy = StaticTemplate(template)
            self.strategies["static"] = static_strategy

        self._switch_mode("static")

    def use_contextual_mode(
        self,
        use_per_fact: bool = True,
        template_preview_chars: int | None = 100,
        preserve_span_newlines: bool = False,
        template_prompt: str | None = None,
        system_prompt: str | None = None,
    ) -> bool:
        """Switch to contextual mode with configuration.

        :param use_per_fact: Whether to use per-span placeholders (≤8 spans).
        :param template_preview_chars: Max chars per span in the template-gen prompt.
            None = full span. Default 100 (pre-0.2.8 behaviour).
        :param preserve_span_newlines: Keep newlines in spans shown to the template LLM.
        :param template_prompt: Custom Jinja2 prompt string (see ContextualTemplate).
        :param system_prompt: System message for the template-generation call.
        :return: True if switched successfully
        """
        if not self.llm_client:
            warn_contextual_without_client()
            return False

        if self.strategies["contextual"] is None:
            self.strategies["contextual"] = ContextualTemplate(self.llm_client, citation_mode=self.config.citation_mode)

        contextual_strategy = cast("ContextualTemplate", self.strategies["contextual"])
        contextual_strategy.set_per_fact_mode(use_per_fact)
        contextual_strategy.template_preview_chars = template_preview_chars
        contextual_strategy.preserve_span_newlines = preserve_span_newlines
        contextual_strategy.template_prompt = template_prompt
        contextual_strategy.system_prompt = system_prompt

        return self._switch_mode("contextual")

    def use_random_mode(self, templates: list[str] | None = None) -> bool:
        """Switch to random mode with optional template pool.

        :param templates: Optional list of templates
        :return: True if switched successfully
        """
        if templates:
            random_strategy = RandomTemplate(templates, self.llm_client)
            self.strategies["random"] = random_strategy

        return self._switch_mode("random")

    def generate_random_templates(self, count: int = 10) -> bool:
        """Generate diverse random templates if in random mode.

        :param count: Number of templates to generate
        :return: True if generation was attempted
        """
        if self.config.template_mode != "random":
            logger.warning(
                "Generating templates needs random template mode, not %s",
                self.config.template_mode,
            )
            return False

        random_strategy = self.strategies["random"]
        if hasattr(random_strategy, "generate_diverse_templates"):
            try:
                cast("RandomTemplate", random_strategy).generate_diverse_templates(count)
            except ValueError as e:
                logger.warning("Could not generate templates: %s", e)
            else:
                return True

        return False

    def use_question_specific_mode(
        self,
        templates: list[dict[str, Any]] | None = None,
        *,
        embedder: Any = None,  # noqa: ANN401  # duck-typed embedder with encode(texts), such as a SentenceTransformer
    ) -> bool:
        """Switch to question-specific mode with optional template definitions.

        :param templates: Optional list of template configs, each with a "template" and its
                          "examples": [{
                              "template": "Template with [RELEVANT_SENTENCES]",
                              "examples": ["Example question 1", "Example question 2"]
                          }]
        :param embedder: A loaded embedding model, such as a SentenceTransformer, that embeds the
            examples and questions. See `QuestionSpecificTemplate`.
        :return: True if switched successfully
        """
        if templates or embedder is not None:
            self.strategies["question_specific"] = QuestionSpecificTemplate(templates, embedder=embedder)

        return self._switch_mode("question_specific")

    def use_structured_mode(
        self,
        template: str | None = None,
        placeholder_mappings: dict[str, str] | None = None,
    ) -> bool:
        r"""Switch to structured mode with semantic placeholders.

        :param template: Template with semantic placeholders like [METHODOLOGY], [RESULTS]
        :param placeholder_mappings: Custom placeholder → query mappings
        :return: True if switched successfully

        Example:
            manager.use_structured_mode(
                template="# Analysis\\n## Method\\n[METHODOLOGY]\\n## Results\\n[RESULTS]",
                placeholder_mappings={"THEIR_METHOD": "what method did the baseline use"}
            )

        """
        structured_strategy = cast("StructuredTemplate | None", self.strategies.get("structured"))

        if structured_strategy is None:
            structured_strategy = StructuredTemplate(
                rag_system=self.rag_system, citation_mode=self.config.citation_mode
            )
            self.strategies["structured"] = structured_strategy
        else:
            structured_strategy.set_citation_mode(self.config.citation_mode)

        # Update RAG system if not set
        if self.rag_system and not structured_strategy.rag_system:
            structured_strategy.rag_system = self.rag_system

        # Set template if provided
        if template:
            structured_strategy.set_template(template)

        # Add custom mappings if provided
        if placeholder_mappings:
            for placeholder, query in placeholder_mappings.items():
                structured_strategy.add_placeholder_mapping(placeholder, query)

        return self._switch_mode("structured")

    @deprecated("TemplateManager.set_rag_system", use="VerbatimRAG.query_async")
    def set_rag_system(self, rag_system: Any) -> None:  # noqa: ANN401  # duck-typed RAG system
        """Set the RAG system for modes that need it (structured).

        :param rag_system: RAG system instance
        """
        self._attach_rag_system(rag_system)

    def _attach_rag_system(self, rag_system: Any) -> None:  # noqa: ANN401  # duck-typed RAG system
        """Let structured mode call back into the RAG system, until #74 replaces it (ADR 0004)."""
        self.rag_system = rag_system

        # Update strategies that need RAG system
        structured_strategy = cast("StructuredTemplate | None", self.strategies.get("structured"))
        if structured_strategy:
            structured_strategy.rag_system = rag_system

    async def process_structured_async(
        self,
        question: str,
        template: str | None = None,
        placeholder_mappings: dict[str, str] | None = None,
    ) -> str:
        """Run structured extraction, as a convenience helper.

        Note: Prefer calling rag.query_async() directly after setting up
        structured mode. This method is kept for convenience.
        """
        if not self.use_structured_mode(template=template, placeholder_mappings=placeholder_mappings):
            msg = "Structured mode unavailable"
            raise ValueError(msg)

        if not self.rag_system:
            msg = "RAG system not set"
            raise ValueError(msg)

        # Delegate to RAG query which handles structured mode
        response = await self.rag_system.query_async(question)
        return response.answer

    def set_citation_mode(self, mode: str) -> None:
        """Configure how citations are rendered inside filled templates ("inline" or "hidden").

        Deprecated: pass `TransformConfig(citation_mode=...)` to the `Transformer` or its call.
        """
        warn_deprecated("TemplateManager.set_citation_mode", use="TransformConfig(citation_mode=...)")
        allowed = {"inline", "hidden"}
        if mode not in allowed:
            msg = f"Unsupported citation mode: {mode}"
            raise ValueError(msg)

        self._set_citations(citation_mode=mode)

    def set_citation_format(self, citation_format: str) -> None:
        """Configure citation marker format for all strategies.

        Deprecated: pass `TransformConfig(citation_format=...)` to the `Transformer` or its call.

        :param citation_format: str.format template with {number} and {span_id} variables.
            Default "[{number}]". Example: "[{span_id}]" renders "[cite1]" when
            span data includes {"span_id": "cite1"}.
        :raises ConfigurationError: If the format uses other variables
        """
        warn_deprecated("TemplateManager.set_citation_format", use="TransformConfig(citation_format=...)")
        self._set_citations(citation_format=citation_format)

    def _set_citations(self, **settings: str) -> None:
        """Replace the citation settings of `config` and of every strategy.

        `process` fills templates with the settings of the strategies.
        """
        self._replace_config(**settings)

        for strategy in self.strategies.values():
            # Each strategy takes the settings it has a setter for.
            if strategy and "citation_mode" in settings and hasattr(strategy, "set_citation_mode"):
                cast("Any", strategy).set_citation_mode(settings["citation_mode"])
            if strategy and "citation_format" in settings and hasattr(strategy, "set_citation_format"):
                cast("Any", strategy).set_citation_format(settings["citation_format"])

current_mode property writable

The template mode of config.

citation_mode property writable

The citation mode of config.

__init__(llm_client=None, default_mode='static', rag_system=None)

Initialize template manager.

:param llm_client: Optional LLM client for contextual and random modes :param default_mode: Default template mode ("static", "contextual", "random", "question_specific", "structured") :param rag_system: Optional RAG system for structured mode

Source code in packages/core/verbatim_core/templates/manager.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def __init__(
    self,
    llm_client: LLMClient | None = None,
    default_mode: str = "static",
    rag_system: Any = None,  # noqa: ANN401  # duck-typed RAG system; avoids circular import
) -> None:
    """Initialize template manager.

    :param llm_client: Optional LLM client for contextual and random modes
    :param default_mode: Default template mode ("static", "contextual", "random", "question_specific", "structured")
    :param rag_system: Optional RAG system for structured mode
    """
    self.llm_client = llm_client
    self.rag_system = rag_system

    # Initialize strategies
    self.strategies: dict[str, TemplateStrategy | None] = {
        "static": StaticTemplate(),
        "contextual": ContextualTemplate(llm_client) if llm_client else None,
        "random": RandomTemplate(llm_client=llm_client),
        "question_specific": QuestionSpecificTemplate(),
        "structured": StructuredTemplate(rag_system=rag_system),
    }

    # Validate initial mode
    mode = default_mode if default_mode in self.strategies else "static"
    if self.strategies[mode] is None:
        warn_contextual_without_client()
        mode = "static"

    self.config = TransformConfig.model_validate({"template_mode": mode})

set_mode(mode)

Switch to a different template mode.

Deprecated: pass TransformConfig(template_mode=...) to the Transformer or its call.

:param mode: Template mode to switch to :return: True if mode was switched successfully

Source code in packages/core/verbatim_core/templates/manager.py 
89
90
91
92
93
94
95
96
97
98
def set_mode(self, mode: str) -> bool:
    """Switch to a different template mode.

    Deprecated: pass `TransformConfig(template_mode=...)` to the `Transformer` or its call.

    :param mode: Template mode to switch to
    :return: True if mode was switched successfully
    """
    warn_deprecated("TemplateManager.set_mode", use=_mode_replacement(mode))
    return self._switch_mode(mode)

get_current_mode()

Get the current template mode.

:return: Current mode name

Source code in packages/core/verbatim_core/templates/manager.py 
124
125
126
127
128
129
def get_current_mode(self) -> str:
    """Get the current template mode.

    :return: Current mode name
    """
    return self.config.template_mode

strategy(mode)

Return the strategy of a template mode, without switching to it.

:param mode: Template mode :return: The strategy, or None if the mode needs an LLM client the manager doesn't have :raises KeyError: If there is no such mode

Source code in packages/core/verbatim_core/templates/manager.py 
131
132
133
134
135
136
137
138
def strategy(self, mode: str) -> TemplateStrategy | None:
    """Return the strategy of a template mode, without switching to it.

    :param mode: Template mode
    :return: The strategy, or None if the mode needs an LLM client the manager doesn't have
    :raises KeyError: If there is no such mode
    """
    return self.strategies[mode]

get_available_modes()

Get list of available template modes.

:return: List of mode names that are available

Source code in packages/core/verbatim_core/templates/manager.py 
140
141
142
143
144
145
def get_available_modes(self) -> list[str]:
    """Get list of available template modes.

    :return: List of mode names that are available
    """
    return [mode for mode, strategy in self.strategies.items() if strategy is not None]

process(question, display_spans, citation_spans)

Generate and fill a template in one operation.

:param question: The user's question :param display_spans: Spans to display with full text :param citation_spans: Spans for citation reference only :return: Completed response text

Source code in packages/core/verbatim_core/templates/manager.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
def process(
    self,
    question: str,
    display_spans: list[dict[str, Any]],
    citation_spans: list[dict[str, Any]],
) -> str:
    """Generate and fill a template in one operation.

    :param question: The user's question
    :param display_spans: Spans to display with full text
    :param citation_spans: Spans for citation reference only
    :return: Completed response text
    """
    all_spans, citation_count = self._get_template_inputs(display_spans, citation_spans)

    # Generate template
    strategy = self._current_strategy()
    template = strategy.generate(question, all_spans, citation_count)

    # Fill template
    return strategy.fill(template, display_spans, citation_spans)

process_async(question, display_spans, citation_spans) async

Async version of process for contextual templates.

:param question: The user's question :param display_spans: Spans to display with full text :param citation_spans: Spans for citation reference only :return: Completed response text

Source code in packages/core/verbatim_core/templates/manager.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
async def process_async(
    self,
    question: str,
    display_spans: list[dict[str, Any]],
    citation_spans: list[dict[str, Any]],
) -> str:
    """Async version of process for contextual templates.

    :param question: The user's question
    :param display_spans: Spans to display with full text
    :param citation_spans: Spans for citation reference only
    :return: Completed response text
    """
    all_spans, citation_count = self._get_template_inputs(display_spans, citation_spans)

    strategy = self._current_strategy()

    # Use async generation if available
    if hasattr(strategy, "generate_async") and self.config.template_mode == "contextual":
        template = await cast("ContextualTemplate", strategy).generate_async(question, all_spans, citation_count)
    else:
        template = strategy.generate(question, all_spans, citation_count)

    return strategy.fill(template, display_spans, citation_spans)

get_template(question='', spans=None, citation_count=0)

Generate a template without filling it.

:param question: The user's question :param spans: List of spans that will fill the template :param citation_count: Number of citation-only spans :return: Template string with placeholders

Source code in packages/core/verbatim_core/templates/manager.py 
212
213
214
215
216
217
218
219
220
221
222
def get_template(self, question: str = "", spans: list[str] | None = None, citation_count: int = 0) -> str:
    """Generate a template without filling it.

    :param question: The user's question
    :param spans: List of spans that will fill the template
    :param citation_count: Number of citation-only spans
    :return: Template string with placeholders
    """
    spans = spans or []
    strategy = self._current_strategy()
    return strategy.generate(question, spans, citation_count)

fill_template(template, display_spans, citation_spans)

Fill a template with content.

:param template: Template string with placeholders :param display_spans: Spans to display with full text :param citation_spans: Spans for citation reference only :return: Filled template

Source code in packages/core/verbatim_core/templates/manager.py 
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
def fill_template(
    self,
    template: str,
    display_spans: list[dict[str, Any]],
    citation_spans: list[dict[str, Any]],
) -> str:
    """Fill a template with content.

    :param template: Template string with placeholders
    :param display_spans: Spans to display with full text
    :param citation_spans: Spans for citation reference only
    :return: Filled template
    """
    strategy = self._current_strategy()
    return strategy.fill(template, display_spans, citation_spans)

save(filepath)

Save all template configurations to file.

:param filepath: Path to save configuration

Source code in packages/core/verbatim_core/templates/manager.py 
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
def save(self, filepath: str) -> None:
    """Save all template configurations to file.

    :param filepath: Path to save configuration
    """
    data = {
        "current_mode": self.config.template_mode,
        "citation_mode": self.config.citation_mode,
        "citation_format": self.config.citation_format,
        "strategies": {},
    }

    # Save state for each available strategy
    for mode, strategy in self.strategies.items():
        if strategy is not None:
            data["strategies"][mode] = strategy.save_state()

    # Create directory if it doesn't exist
    os.makedirs(os.path.dirname(filepath), exist_ok=True)  # noqa: PTH103, PTH120  # for a bare file name, Path.parent is "." where dirname is ""

    with Path(filepath).open("w") as f:
        json.dump(data, f, indent=2)

load(filepath)

Load template configurations from file.

:param filepath: Path to load configuration from :return: True if loaded successfully

Source code in packages/core/verbatim_core/templates/manager.py 
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
def load(self, filepath: str) -> bool:
    """Load template configurations from file.

    :param filepath: Path to load configuration from
    :return: True if loaded successfully
    """
    if not os.path.exists(filepath):  # noqa: PTH110  # unlike Path.exists, an unreadable path is "not found"
        logger.warning("Template config file not found: %s", filepath)
        return False

    try:
        with Path(filepath).open() as f:
            data = json.load(f)

        # Load mode
        if "current_mode" in data:
            mode = data["current_mode"]
            if self.strategies.get(mode) is not None:
                self._replace_config(template_mode=mode)

        # Load strategy states
        strategies_data = data.get("strategies", {})
        for mode, state in strategies_data.items():
            strategy = self.strategies.get(mode)
            if strategy is not None:
                # A saved state of the wrong shape doesn't stop the other strategies.
                try:
                    strategy.load_state(state)
                except (KeyError, TypeError, ValueError):
                    logger.warning(
                        "Could not load the saved state of the %s templates",
                        mode,
                        exc_info=True,
                    )

        # Before 0.3.0 only the contextual templates saved the citation format.
        citations = {key: data[key] for key in ("citation_mode", "citation_format") if key in data}
        contextual_state = strategies_data.get("contextual") or {}
        if "citation_format" not in citations and "citation_format" in contextual_state:
            citations["citation_format"] = contextual_state["citation_format"]
        if citations:
            self._set_citations(**citations)

    # A file that can't be read, or isn't JSON
    except (OSError, ValueError):
        logger.warning("Could not load the template config %s", filepath, exc_info=True)
        return False
    else:
        return True

info()

Get current template manager state information.

:return: Dictionary with current state info

Source code in packages/core/verbatim_core/templates/manager.py 
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
def info(self) -> dict[str, Any]:
    """Get current template manager state information.

    :return: Dictionary with current state info
    """
    info_data = {
        "current_mode": self.config.template_mode,
        "available_modes": self.get_available_modes(),
        "has_llm_client": self.llm_client is not None,
    }

    # Add mode-specific info
    if self.config.template_mode == "random":
        random_strategy = self.strategies["random"]
        if hasattr(random_strategy, "get_template_count"):
            info_data["random_template_count"] = cast("RandomTemplate", random_strategy).get_template_count()

    return info_data

use_static_mode(template=None)

Switch to static mode with optional custom template.

:param template: Optional custom template

Source code in packages/core/verbatim_core/templates/manager.py 
334
335
336
337
338
339
340
341
342
343
def use_static_mode(self, template: str | None = None) -> None:
    """Switch to static mode with optional custom template.

    :param template: Optional custom template
    """
    if template:
        static_strategy = StaticTemplate(template)
        self.strategies["static"] = static_strategy

    self._switch_mode("static")

use_contextual_mode(use_per_fact=True, template_preview_chars=100, preserve_span_newlines=False, template_prompt=None, system_prompt=None)

Switch to contextual mode with configuration.

:param use_per_fact: Whether to use per-span placeholders (≤8 spans). :param template_preview_chars: Max chars per span in the template-gen prompt. None = full span. Default 100 (pre-0.2.8 behaviour). :param preserve_span_newlines: Keep newlines in spans shown to the template LLM. :param template_prompt: Custom Jinja2 prompt string (see ContextualTemplate). :param system_prompt: System message for the template-generation call. :return: True if switched successfully

Source code in packages/core/verbatim_core/templates/manager.py 
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
def use_contextual_mode(
    self,
    use_per_fact: bool = True,
    template_preview_chars: int | None = 100,
    preserve_span_newlines: bool = False,
    template_prompt: str | None = None,
    system_prompt: str | None = None,
) -> bool:
    """Switch to contextual mode with configuration.

    :param use_per_fact: Whether to use per-span placeholders (≤8 spans).
    :param template_preview_chars: Max chars per span in the template-gen prompt.
        None = full span. Default 100 (pre-0.2.8 behaviour).
    :param preserve_span_newlines: Keep newlines in spans shown to the template LLM.
    :param template_prompt: Custom Jinja2 prompt string (see ContextualTemplate).
    :param system_prompt: System message for the template-generation call.
    :return: True if switched successfully
    """
    if not self.llm_client:
        warn_contextual_without_client()
        return False

    if self.strategies["contextual"] is None:
        self.strategies["contextual"] = ContextualTemplate(self.llm_client, citation_mode=self.config.citation_mode)

    contextual_strategy = cast("ContextualTemplate", self.strategies["contextual"])
    contextual_strategy.set_per_fact_mode(use_per_fact)
    contextual_strategy.template_preview_chars = template_preview_chars
    contextual_strategy.preserve_span_newlines = preserve_span_newlines
    contextual_strategy.template_prompt = template_prompt
    contextual_strategy.system_prompt = system_prompt

    return self._switch_mode("contextual")

use_random_mode(templates=None)

Switch to random mode with optional template pool.

:param templates: Optional list of templates :return: True if switched successfully

Source code in packages/core/verbatim_core/templates/manager.py 
379
380
381
382
383
384
385
386
387
388
389
def use_random_mode(self, templates: list[str] | None = None) -> bool:
    """Switch to random mode with optional template pool.

    :param templates: Optional list of templates
    :return: True if switched successfully
    """
    if templates:
        random_strategy = RandomTemplate(templates, self.llm_client)
        self.strategies["random"] = random_strategy

    return self._switch_mode("random")

generate_random_templates(count=10)

Generate diverse random templates if in random mode.

:param count: Number of templates to generate :return: True if generation was attempted

Source code in packages/core/verbatim_core/templates/manager.py 
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
def generate_random_templates(self, count: int = 10) -> bool:
    """Generate diverse random templates if in random mode.

    :param count: Number of templates to generate
    :return: True if generation was attempted
    """
    if self.config.template_mode != "random":
        logger.warning(
            "Generating templates needs random template mode, not %s",
            self.config.template_mode,
        )
        return False

    random_strategy = self.strategies["random"]
    if hasattr(random_strategy, "generate_diverse_templates"):
        try:
            cast("RandomTemplate", random_strategy).generate_diverse_templates(count)
        except ValueError as e:
            logger.warning("Could not generate templates: %s", e)
        else:
            return True

    return False

use_question_specific_mode(templates=None, *, embedder=None)

Switch to question-specific mode with optional template definitions.

:param templates: Optional list of template configs, each with a "template" and its "examples": [{ "template": "Template with [RELEVANT_SENTENCES]", "examples": ["Example question 1", "Example question 2"] }] :param embedder: A loaded embedding model, such as a SentenceTransformer, that embeds the examples and questions. See QuestionSpecificTemplate. :return: True if switched successfully

Source code in packages/core/verbatim_core/templates/manager.py 
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
def use_question_specific_mode(
    self,
    templates: list[dict[str, Any]] | None = None,
    *,
    embedder: Any = None,  # noqa: ANN401  # duck-typed embedder with encode(texts), such as a SentenceTransformer
) -> bool:
    """Switch to question-specific mode with optional template definitions.

    :param templates: Optional list of template configs, each with a "template" and its
                      "examples": [{
                          "template": "Template with [RELEVANT_SENTENCES]",
                          "examples": ["Example question 1", "Example question 2"]
                      }]
    :param embedder: A loaded embedding model, such as a SentenceTransformer, that embeds the
        examples and questions. See `QuestionSpecificTemplate`.
    :return: True if switched successfully
    """
    if templates or embedder is not None:
        self.strategies["question_specific"] = QuestionSpecificTemplate(templates, embedder=embedder)

    return self._switch_mode("question_specific")

use_structured_mode(template=None, placeholder_mappings=None)

Switch to structured mode with semantic placeholders.

:param template: Template with semantic placeholders like [METHODOLOGY], [RESULTS] :param placeholder_mappings: Custom placeholder → query mappings :return: True if switched successfully

Example

manager.use_structured_mode( template="# Analysis\n## Method\n[METHODOLOGY]\n## Results\n[RESULTS]", placeholder_mappings={"THEIR_METHOD": "what method did the baseline use"} )

Source code in packages/core/verbatim_core/templates/manager.py 
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
def use_structured_mode(
    self,
    template: str | None = None,
    placeholder_mappings: dict[str, str] | None = None,
) -> bool:
    r"""Switch to structured mode with semantic placeholders.

    :param template: Template with semantic placeholders like [METHODOLOGY], [RESULTS]
    :param placeholder_mappings: Custom placeholder → query mappings
    :return: True if switched successfully

    Example:
        manager.use_structured_mode(
            template="# Analysis\\n## Method\\n[METHODOLOGY]\\n## Results\\n[RESULTS]",
            placeholder_mappings={"THEIR_METHOD": "what method did the baseline use"}
        )

    """
    structured_strategy = cast("StructuredTemplate | None", self.strategies.get("structured"))

    if structured_strategy is None:
        structured_strategy = StructuredTemplate(
            rag_system=self.rag_system, citation_mode=self.config.citation_mode
        )
        self.strategies["structured"] = structured_strategy
    else:
        structured_strategy.set_citation_mode(self.config.citation_mode)

    # Update RAG system if not set
    if self.rag_system and not structured_strategy.rag_system:
        structured_strategy.rag_system = self.rag_system

    # Set template if provided
    if template:
        structured_strategy.set_template(template)

    # Add custom mappings if provided
    if placeholder_mappings:
        for placeholder, query in placeholder_mappings.items():
            structured_strategy.add_placeholder_mapping(placeholder, query)

    return self._switch_mode("structured")

set_rag_system(rag_system)

Set the RAG system for modes that need it (structured).

:param rag_system: RAG system instance

Source code in packages/core/verbatim_core/templates/manager.py 
480
481
482
483
484
485
486
@deprecated("TemplateManager.set_rag_system", use="VerbatimRAG.query_async")
def set_rag_system(self, rag_system: Any) -> None:  # noqa: ANN401  # duck-typed RAG system
    """Set the RAG system for modes that need it (structured).

    :param rag_system: RAG system instance
    """
    self._attach_rag_system(rag_system)

process_structured_async(question, template=None, placeholder_mappings=None) async

Run structured extraction, as a convenience helper.

Note: Prefer calling rag.query_async() directly after setting up structured mode. This method is kept for convenience.

Source code in packages/core/verbatim_core/templates/manager.py 
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
async def process_structured_async(
    self,
    question: str,
    template: str | None = None,
    placeholder_mappings: dict[str, str] | None = None,
) -> str:
    """Run structured extraction, as a convenience helper.

    Note: Prefer calling rag.query_async() directly after setting up
    structured mode. This method is kept for convenience.
    """
    if not self.use_structured_mode(template=template, placeholder_mappings=placeholder_mappings):
        msg = "Structured mode unavailable"
        raise ValueError(msg)

    if not self.rag_system:
        msg = "RAG system not set"
        raise ValueError(msg)

    # Delegate to RAG query which handles structured mode
    response = await self.rag_system.query_async(question)
    return response.answer

set_citation_mode(mode)

Configure how citations are rendered inside filled templates ("inline" or "hidden").

Deprecated: pass TransformConfig(citation_mode=...) to the Transformer or its call.

Source code in packages/core/verbatim_core/templates/manager.py 
520
521
522
523
524
525
526
527
528
529
530
531
def set_citation_mode(self, mode: str) -> None:
    """Configure how citations are rendered inside filled templates ("inline" or "hidden").

    Deprecated: pass `TransformConfig(citation_mode=...)` to the `Transformer` or its call.
    """
    warn_deprecated("TemplateManager.set_citation_mode", use="TransformConfig(citation_mode=...)")
    allowed = {"inline", "hidden"}
    if mode not in allowed:
        msg = f"Unsupported citation mode: {mode}"
        raise ValueError(msg)

    self._set_citations(citation_mode=mode)

set_citation_format(citation_format)

Configure citation marker format for all strategies.

Deprecated: pass TransformConfig(citation_format=...) to the Transformer or its call.

:param citation_format: str.format template with {number} and {span_id} variables. Default "[{number}]". Example: "[{span_id}]" renders "[cite1]" when span data includes {"span_id": "cite1"}. :raises ConfigurationError: If the format uses other variables

Source code in packages/core/verbatim_core/templates/manager.py 
533
534
535
536
537
538
539
540
541
542
543
544
def set_citation_format(self, citation_format: str) -> None:
    """Configure citation marker format for all strategies.

    Deprecated: pass `TransformConfig(citation_format=...)` to the `Transformer` or its call.

    :param citation_format: str.format template with {number} and {span_id} variables.
        Default "[{number}]". Example: "[{span_id}]" renders "[cite1]" when
        span data includes {"span_id": "cite1"}.
    :raises ConfigurationError: If the format uses other variables
    """
    warn_deprecated("TemplateManager.set_citation_format", use="TransformConfig(citation_format=...)")
    self._set_citations(citation_format=citation_format)

QuestionSpecificTemplate

verbatim_core.templates.question_specific.QuestionSpecificTemplate

Bases: TemplateStrategy

Question-specific template strategy using semantic similarity matching.

Maintains a list of (template, example_questions) pairs. When a query arrives, computes similarity to all examples and selects the template from the best-matching pair.

Example usage

templates = [ { "template": "Methodology: [RELEVANT_SENTENCES]", "examples": ["What methods were used?", "How was this done?"] }, { "template": "Key Findings: [RELEVANT_SENTENCES]", "examples": ["What were the results?", "What did they find?"] } ] strategy = QuestionSpecificTemplate.from_pretrained("all-MiniLM-L6-v2", templates=templates)

Source code in packages/core/verbatim_core/templates/question_specific.py 
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 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
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
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
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
class QuestionSpecificTemplate(TemplateStrategy):
    """Question-specific template strategy using semantic similarity matching.

    Maintains a list of (template, example_questions) pairs. When a query arrives,
    computes similarity to all examples and selects the template from the
    best-matching pair.

    Example usage:
        templates = [
            {
                "template": "Methodology: [RELEVANT_SENTENCES]",
                "examples": ["What methods were used?", "How was this done?"]
            },
            {
                "template": "Key Findings: [RELEVANT_SENTENCES]",
                "examples": ["What were the results?", "What did they find?"]
            }
        ]
        strategy = QuestionSpecificTemplate.from_pretrained("all-MiniLM-L6-v2", templates=templates)
    """

    DEFAULT_TEMPLATE = """## Response

Based on the available documents:

[DISPLAY_SPANS]

---
*These excerpts are taken verbatim from the source documents to ensure accuracy.*"""

    def __init__(
        self,
        templates: list[dict[str, Any]] | None = None,
        model_name: str = _UNSET,
        device: str = _UNSET,
        citation_mode: str = "inline",
        *,
        embedder: Any = None,  # noqa: ANN401  # duck-typed embedder with encode(texts), such as a SentenceTransformer
    ) -> None:
        """Initialize question-specific template strategy.

        :param templates: List of template definitions, each with 'template' and 'examples' keys
        :param model_name: Deprecated. Load the embedder with `from_pretrained(model_name)`.
        :param device: Deprecated, with `model_name`.
        :param embedder: A loaded embedding model with an `encode(texts)` method, such as a
            SentenceTransformer. It embeds the template examples here and each question.
        """
        # Loading the embedder by name warns once: here if the caller named the model, else when
        # it is loaded.
        self._warn_on_load = model_name is _UNSET and device is _UNSET
        if not self._warn_on_load:
            warn_deprecated(
                "QuestionSpecificTemplate(model_name=..., device=...)",
                use="QuestionSpecificTemplate.from_pretrained(model_name, device)",
            )

        self.templates = templates or []
        # The name of the model that embeds, saved and loaded by. None for an embedder given
        # without a name.
        if model_name is not _UNSET:
            self.model_name = model_name
        else:
            self.model_name = None if embedder is not None else _DEFAULT_MODEL_NAME
        self.device = "cpu" if device is _UNSET else device
        self.citation_mode = citation_mode
        self.filler = TemplateFiller(citation_mode=citation_mode)
        self._embedder = embedder
        self._example_embeddings: dict[str, Any] | None = None  # Will store stacked embeddings for efficient matching

        if self.templates:
            self._validate_and_embed_templates()

    @classmethod
    def from_pretrained(
        cls,
        model_name: str,
        device: str | None = None,
        *,
        templates: list[dict[str, Any]] | None = None,
        citation_mode: str = "inline",
    ) -> "QuestionSpecificTemplate":
        """Load a sentence-transformers embedding model and build the templates with it.

        :param model_name: A sentence-transformers model id, or the path of a local model
        :param device: The device to run the model on. None lets sentence-transformers choose.
        :param templates: List of template definitions, each with 'template' and 'examples' keys
        :param citation_mode: "inline" or "hidden"
        :raises ConfigurationError: If sentence-transformers is not installed, or the model can't
            be loaded
        """
        embedder = _load_embedder(model_name, device)
        strategy = cls(templates, citation_mode=citation_mode, embedder=embedder)
        strategy.model_name = model_name
        strategy.device = str(embedder.device)
        return strategy

    def set_citation_mode(self, citation_mode: str) -> None:
        """Set how the filled templates render citations.

        Args:
            citation_mode: "inline" or "hidden".

        """
        self.citation_mode = citation_mode
        self.filler.set_citation_mode(citation_mode)

    def _get_embedder(self) -> Any:  # noqa: ANN401  # optional dependency type (SentenceTransformer)
        """Return the embedder. Without one, it is loaded by name, as before 0.3.0."""
        if self._embedder is None:
            if self.model_name is None:
                msg = (
                    "The question-specific templates don't name the model that embedded their "
                    "examples. Pass that model as embedder, for example with "
                    "TemplateManager.use_question_specific_mode(embedder=...) before load."
                )
                raise ConfigurationError(
                    msg,
                    code="missing_embedder",
                )
            if self._warn_on_load:
                warn_deprecated(
                    "QuestionSpecificTemplate without an embedder",
                    use="QuestionSpecificTemplate.from_pretrained(model_name)",
                    stacklevel=None,
                )
            self._embedder = _load_embedder(self.model_name, self.device)
        return self._embedder

    def _validate_and_embed_templates(self) -> None:
        """Validate template format and precompute example embeddings."""
        if not self.templates:
            return

        embedder = self._get_embedder()

        # Collect all examples and their template indices
        all_examples = []
        template_indices = []  # Which template each example belongs to

        for idx, config in enumerate(self.templates):
            if "template" not in config:
                msg = f"Template config {idx} missing 'template' key"
                raise ValueError(msg)
            if "examples" not in config or not config["examples"]:
                msg = f"Template config {idx} missing or empty 'examples'"
                raise ValueError(msg)

            # Validate template has required placeholders
            self.validate_template(config["template"])

            # Track which template these examples belong to
            examples = config["examples"]
            all_examples.extend(examples)
            template_indices.extend([idx] * len(examples))

        # Embed all examples at once (more efficient)
        embeddings = embedder.encode(all_examples)
        self._example_embeddings = {
            "embeddings": embeddings,  # numpy array of shape (N, embedding_dim)
            "template_indices": template_indices,  # which template each embedding belongs to
        }

        logger.debug("Validated and embedded %d example questions", len(all_examples))

    def set_question_templates(self, templates: list[dict[str, Any]]) -> None:
        """Set templates with example questions.

        :param templates: List of template configs, each with:
                          - "template": Template string with [RELEVANT_SENTENCES]
                          - "examples": List of example questions
        :raises ValueError: If template format is invalid
        """
        self.templates = templates
        self._example_embeddings = None
        self._validate_and_embed_templates()

    def _find_best_match(self, question: str) -> tuple[int, float]:
        """Find the best matching template for a question.

        :param question: User's question
        :return: Tuple of (template_index, best_similarity_score)
        """
        if not self.templates or self._example_embeddings is None:
            return -1, 0.0

        embedder = self._get_embedder()
        question_embedding = embedder.encode([question])[0]  # Shape: (embedding_dim,)

        similarities = [
            _cosine(example_embedding, question_embedding)
            for example_embedding in self._example_embeddings["embeddings"]
        ]

        # Find the best similarity, the first on a tie
        best_example_idx = max(range(len(similarities)), key=similarities.__getitem__)
        best_similarity = float(similarities[best_example_idx])
        best_template_idx = self._example_embeddings["template_indices"][best_example_idx]

        logger.debug("Question matched template %d (similarity: %.3f)", best_template_idx, best_similarity)

        return best_template_idx, best_similarity

    def generate(self, question: str, spans: list[str], citation_count: int = 0) -> str:
        """Generate template by finding best match for the question.

        :param question: User's question
        :param spans: List of spans (not used for matching)
        :param citation_count: Number of citation spans (not used for matching)
        :return: Best matching template or default template
        """
        template_idx, _similarity = self._find_best_match(question)

        if template_idx >= 0 and template_idx < len(self.templates):
            return self.templates[template_idx]["template"]

        logger.debug("Using default template (no templates configured)")
        return self.DEFAULT_TEMPLATE

    def fill(
        self,
        template: str,
        display_spans: list[dict[str, Any]],
        citation_spans: list[dict[str, Any]],
    ) -> str:
        """Fill the template with actual span content.

        :param template: Template string with placeholders
        :param display_spans: Spans to display with full text
        :param citation_spans: Spans for citation reference only
        :return: Filled template
        """
        return self.filler.fill(template, display_spans, citation_spans)

    def save_state(self) -> dict[str, Any]:
        """Save the current template configuration.

        :return: Dictionary with template state
        """
        return {
            "type": "question_specific",
            "templates": self.templates,
            "model_name": self.model_name,
            "device": self.device,
        }

    def load_state(self, state: dict[str, Any]) -> None:
        """Load template configuration from saved state.

        :param state: Dictionary with template configuration
        """
        if "templates" in state:
            self.templates = state["templates"]
        # An embedder given replaces the saved model.
        if self._embedder is None and "model_name" in state:
            self.model_name = state["model_name"]
        if self._embedder is None and "device" in state:
            self.device = state["device"]

        # Embed the loaded examples, with the embedder given, else the saved model
        self._example_embeddings = None
        if self.templates:
            self._validate_and_embed_templates()

    def add_template(self, template: str, examples: list[str]) -> None:
        """Add a new template to the list.

        :param template: Template string with [RELEVANT_SENTENCES]
        :param examples: List of example questions for this template
        :raises ValueError: If template is invalid
        """
        self.validate_template(template)
        if not examples:
            msg = "At least one example question is required"
            raise ValueError(msg)

        self.templates.append({"template": template, "examples": examples})

        # Revalidate and re-embed all templates
        self._example_embeddings = None
        self._validate_and_embed_templates()
        logger.debug("Added template (total: %d)", len(self.templates))

    def remove_template(self, index: int) -> None:
        """Remove a template by index.

        :param index: Index of template to remove
        :raises IndexError: If index is out of range
        """
        if index < 0 or index >= len(self.templates):
            msg = f"Template index {index} out of range [0, {len(self.templates) - 1}]"
            raise IndexError(msg)

        self.templates.pop(index)
        self._example_embeddings = None
        if self.templates:
            self._validate_and_embed_templates()
        logger.debug("Removed template at index %d (total: %d)", index, len(self.templates))

__init__(templates=None, model_name=_UNSET, device=_UNSET, citation_mode='inline', *, embedder=None)

Initialize question-specific template strategy.

:param templates: List of template definitions, each with 'template' and 'examples' keys :param model_name: Deprecated. Load the embedder with from_pretrained(model_name). :param device: Deprecated, with model_name. :param embedder: A loaded embedding model with an encode(texts) method, such as a SentenceTransformer. It embeds the template examples here and each question.

Source code in packages/core/verbatim_core/templates/question_specific.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def __init__(
    self,
    templates: list[dict[str, Any]] | None = None,
    model_name: str = _UNSET,
    device: str = _UNSET,
    citation_mode: str = "inline",
    *,
    embedder: Any = None,  # noqa: ANN401  # duck-typed embedder with encode(texts), such as a SentenceTransformer
) -> None:
    """Initialize question-specific template strategy.

    :param templates: List of template definitions, each with 'template' and 'examples' keys
    :param model_name: Deprecated. Load the embedder with `from_pretrained(model_name)`.
    :param device: Deprecated, with `model_name`.
    :param embedder: A loaded embedding model with an `encode(texts)` method, such as a
        SentenceTransformer. It embeds the template examples here and each question.
    """
    # Loading the embedder by name warns once: here if the caller named the model, else when
    # it is loaded.
    self._warn_on_load = model_name is _UNSET and device is _UNSET
    if not self._warn_on_load:
        warn_deprecated(
            "QuestionSpecificTemplate(model_name=..., device=...)",
            use="QuestionSpecificTemplate.from_pretrained(model_name, device)",
        )

    self.templates = templates or []
    # The name of the model that embeds, saved and loaded by. None for an embedder given
    # without a name.
    if model_name is not _UNSET:
        self.model_name = model_name
    else:
        self.model_name = None if embedder is not None else _DEFAULT_MODEL_NAME
    self.device = "cpu" if device is _UNSET else device
    self.citation_mode = citation_mode
    self.filler = TemplateFiller(citation_mode=citation_mode)
    self._embedder = embedder
    self._example_embeddings: dict[str, Any] | None = None  # Will store stacked embeddings for efficient matching

    if self.templates:
        self._validate_and_embed_templates()

from_pretrained(model_name, device=None, *, templates=None, citation_mode='inline') classmethod

Load a sentence-transformers embedding model and build the templates with it.

:param model_name: A sentence-transformers model id, or the path of a local model :param device: The device to run the model on. None lets sentence-transformers choose. :param templates: List of template definitions, each with 'template' and 'examples' keys :param citation_mode: "inline" or "hidden" :raises ConfigurationError: If sentence-transformers is not installed, or the model can't be loaded

Source code in packages/core/verbatim_core/templates/question_specific.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
@classmethod
def from_pretrained(
    cls,
    model_name: str,
    device: str | None = None,
    *,
    templates: list[dict[str, Any]] | None = None,
    citation_mode: str = "inline",
) -> "QuestionSpecificTemplate":
    """Load a sentence-transformers embedding model and build the templates with it.

    :param model_name: A sentence-transformers model id, or the path of a local model
    :param device: The device to run the model on. None lets sentence-transformers choose.
    :param templates: List of template definitions, each with 'template' and 'examples' keys
    :param citation_mode: "inline" or "hidden"
    :raises ConfigurationError: If sentence-transformers is not installed, or the model can't
        be loaded
    """
    embedder = _load_embedder(model_name, device)
    strategy = cls(templates, citation_mode=citation_mode, embedder=embedder)
    strategy.model_name = model_name
    strategy.device = str(embedder.device)
    return strategy

set_citation_mode(citation_mode)

Set how the filled templates render citations.

Parameters:

Name Type Description Default
citation_mode str

"inline" or "hidden".

 required
Source code in packages/core/verbatim_core/templates/question_specific.py 
122
123
124
125
126
127
128
129
130
def set_citation_mode(self, citation_mode: str) -> None:
    """Set how the filled templates render citations.

    Args:
        citation_mode: "inline" or "hidden".

    """
    self.citation_mode = citation_mode
    self.filler.set_citation_mode(citation_mode)

set_question_templates(templates)

Set templates with example questions.

:param templates: List of template configs, each with: - "template": Template string with [RELEVANT_SENTENCES] - "examples": List of example questions :raises ValueError: If template format is invalid

Source code in packages/core/verbatim_core/templates/question_specific.py 
190
191
192
193
194
195
196
197
198
199
200
def set_question_templates(self, templates: list[dict[str, Any]]) -> None:
    """Set templates with example questions.

    :param templates: List of template configs, each with:
                      - "template": Template string with [RELEVANT_SENTENCES]
                      - "examples": List of example questions
    :raises ValueError: If template format is invalid
    """
    self.templates = templates
    self._example_embeddings = None
    self._validate_and_embed_templates()

generate(question, spans, citation_count=0)

Generate template by finding best match for the question.

:param question: User's question :param spans: List of spans (not used for matching) :param citation_count: Number of citation spans (not used for matching) :return: Best matching template or default template

Source code in packages/core/verbatim_core/templates/question_specific.py 
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
def generate(self, question: str, spans: list[str], citation_count: int = 0) -> str:
    """Generate template by finding best match for the question.

    :param question: User's question
    :param spans: List of spans (not used for matching)
    :param citation_count: Number of citation spans (not used for matching)
    :return: Best matching template or default template
    """
    template_idx, _similarity = self._find_best_match(question)

    if template_idx >= 0 and template_idx < len(self.templates):
        return self.templates[template_idx]["template"]

    logger.debug("Using default template (no templates configured)")
    return self.DEFAULT_TEMPLATE

fill(template, display_spans, citation_spans)

Fill the template with actual span content.

:param template: Template string with placeholders :param display_spans: Spans to display with full text :param citation_spans: Spans for citation reference only :return: Filled template

Source code in packages/core/verbatim_core/templates/question_specific.py 
244
245
246
247
248
249
250
251
252
253
254
255
256
257
def fill(
    self,
    template: str,
    display_spans: list[dict[str, Any]],
    citation_spans: list[dict[str, Any]],
) -> str:
    """Fill the template with actual span content.

    :param template: Template string with placeholders
    :param display_spans: Spans to display with full text
    :param citation_spans: Spans for citation reference only
    :return: Filled template
    """
    return self.filler.fill(template, display_spans, citation_spans)

save_state()

Save the current template configuration.

:return: Dictionary with template state

Source code in packages/core/verbatim_core/templates/question_specific.py 
259
260
261
262
263
264
265
266
267
268
269
def save_state(self) -> dict[str, Any]:
    """Save the current template configuration.

    :return: Dictionary with template state
    """
    return {
        "type": "question_specific",
        "templates": self.templates,
        "model_name": self.model_name,
        "device": self.device,
    }

load_state(state)

Load template configuration from saved state.

:param state: Dictionary with template configuration

Source code in packages/core/verbatim_core/templates/question_specific.py 
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
def load_state(self, state: dict[str, Any]) -> None:
    """Load template configuration from saved state.

    :param state: Dictionary with template configuration
    """
    if "templates" in state:
        self.templates = state["templates"]
    # An embedder given replaces the saved model.
    if self._embedder is None and "model_name" in state:
        self.model_name = state["model_name"]
    if self._embedder is None and "device" in state:
        self.device = state["device"]

    # Embed the loaded examples, with the embedder given, else the saved model
    self._example_embeddings = None
    if self.templates:
        self._validate_and_embed_templates()

add_template(template, examples)

Add a new template to the list.

:param template: Template string with [RELEVANT_SENTENCES] :param examples: List of example questions for this template :raises ValueError: If template is invalid

Source code in packages/core/verbatim_core/templates/question_specific.py 
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
def add_template(self, template: str, examples: list[str]) -> None:
    """Add a new template to the list.

    :param template: Template string with [RELEVANT_SENTENCES]
    :param examples: List of example questions for this template
    :raises ValueError: If template is invalid
    """
    self.validate_template(template)
    if not examples:
        msg = "At least one example question is required"
        raise ValueError(msg)

    self.templates.append({"template": template, "examples": examples})

    # Revalidate and re-embed all templates
    self._example_embeddings = None
    self._validate_and_embed_templates()
    logger.debug("Added template (total: %d)", len(self.templates))

remove_template(index)

Remove a template by index.

:param index: Index of template to remove :raises IndexError: If index is out of range

Source code in packages/core/verbatim_core/templates/question_specific.py 
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
def remove_template(self, index: int) -> None:
    """Remove a template by index.

    :param index: Index of template to remove
    :raises IndexError: If index is out of range
    """
    if index < 0 or index >= len(self.templates):
        msg = f"Template index {index} out of range [0, {len(self.templates) - 1}]"
        raise IndexError(msg)

    self.templates.pop(index)
    self._example_embeddings = None
    if self.templates:
        self._validate_and_embed_templates()
    logger.debug("Removed template at index %d (total: %d)", index, len(self.templates))

Models

verbatim_core.models

Pydantic models for verbatim_core (RAG-agnostic).

Highlight

Bases: BaseModel

A highlighted character span within a document.

Source code in packages/core/verbatim_core/models.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
class Highlight(BaseModel):
    """A highlighted character span within a document."""

    text: str = Field(..., min_length=1)
    start: int = Field(..., ge=0)
    end: int = Field(..., ge=0)

    @model_validator(mode="after")
    def validate_end_after_start(self) -> Highlight:
        """Ensure the highlight ends after it starts."""
        if self.end <= self.start:
            msg = "end must be greater than start"
            raise ValueError(msg)
        return self
validate_end_after_start()

Ensure the highlight ends after it starts.

Source code in packages/core/verbatim_core/models.py 
20
21
22
23
24
25
26
@model_validator(mode="after")
def validate_end_after_start(self) -> Highlight:
    """Ensure the highlight ends after it starts."""
    if self.end <= self.start:
        msg = "end must be greater than start"
        raise ValueError(msg)
    return self

DocumentWithHighlights

Bases: BaseModel

A document together with its highlighted spans.

Source code in packages/core/verbatim_core/models.py 
29
30
31
32
33
34
35
36
class DocumentWithHighlights(BaseModel):
    """A document together with its highlighted spans."""

    content: str = Field(..., min_length=1)
    highlights: list[Highlight] = Field(default_factory=list)
    title: str = Field(default="")
    source: str = Field(default="")
    metadata: dict[str, Any] = Field(default_factory=dict)

Citation

Bases: BaseModel

A citation pointing to a highlight in a document.

Source code in packages/core/verbatim_core/models.py 
39
40
41
42
43
44
45
46
class Citation(BaseModel):
    """A citation pointing to a highlight in a document."""

    text: str = Field(..., min_length=1)
    doc_index: int = Field(..., ge=0)
    highlight_index: int = Field(..., ge=0)
    number: int | None = Field(default=None, ge=1)
    type: str | None = Field(default=None)

StructuredAnswer

Bases: BaseModel

Answer text with its citations.

Source code in packages/core/verbatim_core/models.py 
49
50
51
52
53
class StructuredAnswer(BaseModel):
    """Answer text with its citations."""

    text: str = Field(..., min_length=1)
    citations: list[Citation] = Field(default_factory=list)

QueryResponse

Bases: BaseModel

Full response to a query.

Source code in packages/core/verbatim_core/models.py 
56
57
58
59
60
61
62
63
64
65
66
class QueryResponse(BaseModel):
    """Full response to a query."""

    model_config = {"arbitrary_types_allowed": True}

    question: str = Field(..., min_length=1)
    answer: str = Field(..., min_length=1)
    structured_answer: StructuredAnswer
    documents: list[DocumentWithHighlights] = Field(default_factory=list)
    # How the answer is degraded but still correct, such as quotes that couldn't be located
    warnings: list[ResultWarning] = Field(default_factory=list)

StreamingResponseType

Bases: Enum

Kinds of streaming response chunks.

Source code in packages/core/verbatim_core/models.py 
69
70
71
72
73
74
class StreamingResponseType(Enum):
    """Kinds of streaming response chunks."""

    DOCUMENTS = "documents"
    HIGHLIGHTS = "highlights"
    ANSWER = "answer"

StreamingResponse

Bases: BaseModel

A single chunk of a streaming response.

Source code in packages/core/verbatim_core/models.py 
77
78
79
80
81
82
class StreamingResponse(BaseModel):
    """A single chunk of a streaming response."""

    type: StreamingResponseType
    data: Any
    done: bool = False

Shared types

verbatim_core.types

Types shared by the stateless interfaces of verbatim_core.

Where they live is provisional: they may move to a package shared with other libraries.

HasText

Bases: Protocol

An object with a text attribute.

Source code in packages/core/verbatim_core/types.py 
13
14
15
16
class HasText(Protocol):
    """An object with a `text` attribute."""

    text: str

Chunk

Bases: BaseModel

A piece of source text with an id, the unit spans are extracted from.

Source code in packages/core/verbatim_core/types.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
class Chunk(BaseModel):
    """A piece of source text with an id, the unit spans are extracted from."""

    model_config = ConfigDict(frozen=True, hide_input_in_errors=True)

    id: str = Field(min_length=1)
    text: str
    metadata: dict[str, Any] = Field(default_factory=dict)
    source_start: int | None = Field(default=None, ge=0)  # position in the source document
    source_end: int | None = Field(default=None, ge=0)

    @model_validator(mode="after")
    def validate_source_offsets_in_order(self) -> Chunk:
        """Check that `source_end` doesn't precede `source_start`."""
        if self.source_start is not None and self.source_end is not None and self.source_end < self.source_start:
            msg = "source_end must not precede source_start"
            raise ValueError(msg)
        return self
validate_source_offsets_in_order()

Check that source_end doesn't precede source_start.

Source code in packages/core/verbatim_core/types.py 
30
31
32
33
34
35
36
@model_validator(mode="after")
def validate_source_offsets_in_order(self) -> Chunk:
    """Check that `source_end` doesn't precede `source_start`."""
    if self.source_start is not None and self.source_end is not None and self.source_end < self.source_start:
        msg = "source_end must not precede source_start"
        raise ValueError(msg)
    return self

Span

Bases: BaseModel

A part of a text, given by offsets in Unicode code points.

The type is shared with other libraries, whose spans point into other texts. In this library the text is always a chunk, and extractors return an ExtractedSpan.

Source code in packages/core/verbatim_core/types.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
class Span(BaseModel):
    """A part of a text, given by offsets in Unicode code points.

    The type is shared with other libraries, whose spans point into other texts. In this library
    the text is always a chunk, and extractors return an `ExtractedSpan`.
    """

    model_config = ConfigDict(frozen=True, hide_input_in_errors=True)

    start: int = Field(ge=0)  # inclusive
    end: int  # exclusive
    text: str  # always equal to the slice the offsets point into
    score: float = Field(ge=0.0, le=1.0)

    @model_validator(mode="after")
    def validate_text_fits_offsets(self) -> Span:
        """Check that `end` is after `start` and that `text` is exactly as long as the offsets say."""
        if self.end <= self.start:
            msg = "end must be greater than start"
            raise ValueError(msg)
        if len(self.text) != self.end - self.start:
            msg = "text must be exactly as long as the span from start to end"
            raise ValueError(msg)
        return self
validate_text_fits_offsets()

Check that end is after start and that text is exactly as long as the offsets say.

Source code in packages/core/verbatim_core/types.py 
53
54
55
56
57
58
59
60
61
62
@model_validator(mode="after")
def validate_text_fits_offsets(self) -> Span:
    """Check that `end` is after `start` and that `text` is exactly as long as the offsets say."""
    if self.end <= self.start:
        msg = "end must be greater than start"
        raise ValueError(msg)
    if len(self.text) != self.end - self.start:
        msg = "text must be exactly as long as the span from start to end"
        raise ValueError(msg)
    return self

ExtractedSpan

Bases: Span

A span of a chunk, as an extractor returns it. Its offsets are relative to the chunk.

Source code in packages/core/verbatim_core/types.py 
65
66
67
68
class ExtractedSpan(Span):
    """A span of a chunk, as an extractor returns it. Its offsets are relative to the chunk."""

    chunk_id: str = Field(min_length=1)

Usage

Bases: BaseModel

What one model or service did during a call.

Source code in packages/core/verbatim_core/types.py 
71
72
73
74
75
76
77
78
79
class Usage(BaseModel):
    """What one model or service did during a call."""

    # Older pydantic 2 releases reserve the "model_" prefix for their own methods.
    model_config = ConfigDict(frozen=True, hide_input_in_errors=True, protected_namespaces=())

    model_id: str | None  # None for work without a model, or with a model the code doesn't name
    model_version: str | None
    units: dict[str, NonNegativeInt]  # for example {"chunks": 12, "input_tokens": 3410}

ResultWarning

Bases: BaseModel

Something that made a result less complete without making it wrong.

Source code in packages/core/verbatim_core/types.py 
82
83
84
85
86
87
88
class ResultWarning(BaseModel):
    """Something that made a result less complete without making it wrong."""

    model_config = ConfigDict(frozen=True, hide_input_in_errors=True)

    code: str = Field(min_length=1)  # stable snake_case identifier, for example "quote_not_found"
    message: str

Extraction

Bases: BaseModel

The spans an extractor found in a set of chunks.

Source code in packages/core/verbatim_core/types.py 
91
92
93
94
95
96
97
98
class Extraction(BaseModel):
    """The spans an extractor found in a set of chunks."""

    model_config = ConfigDict(frozen=True, hide_input_in_errors=True)

    spans: list[ExtractedSpan] = Field(default_factory=list)
    usage: list[Usage] = Field(default_factory=list)
    warnings: list[ResultWarning] = Field(default_factory=list)

AnswerCitation

Bases: BaseModel

A numbered citation in an answer. It points at its span by chunk id and offsets.

Source code in packages/core/verbatim_core/types.py 
101
102
103
104
105
106
107
108
109
110
class AnswerCitation(BaseModel):
    """A numbered citation in an answer. It points at its span by chunk id and offsets."""

    model_config = ConfigDict(frozen=True, hide_input_in_errors=True)

    number: int = Field(ge=1)
    type: Literal["display", "reference"]  # shown in the answer, or cited for reference only
    chunk_id: str = Field(min_length=1)
    start: int = Field(ge=0)
    end: int

Answer

Bases: BaseModel

A cited answer, written from the spans extracted from a set of chunks.

Source code in packages/core/verbatim_core/types.py 
113
114
115
116
117
118
119
120
121
122
class Answer(BaseModel):
    """A cited answer, written from the spans extracted from a set of chunks."""

    model_config = ConfigDict(frozen=True, hide_input_in_errors=True)

    text: str
    citations: list[AnswerCitation] = Field(default_factory=list)
    spans: list[ExtractedSpan] = Field(default_factory=list)
    usage: list[Usage] = Field(default_factory=list)
    warnings: list[ResultWarning] = Field(default_factory=list)

Errors

verbatim_core.errors

Errors raised by verbatim_core and verbatim_rag.

VerbatimError

Bases: Exception

Base class of every error the library raises on purpose.

Source code in packages/core/verbatim_core/errors.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
class VerbatimError(Exception):
    """Base class of every error the library raises on purpose."""

    code = "verbatim_error"

    def __init__(self, message: str, *, code: str | None = None) -> None:
        """Create the error.

        Args:
            message: What the problem is and how to fix it.
            code: A stable snake_case code, in place of the default code of the class.

        """
        super().__init__(message)
        if code is not None:
            self.code = code

    @override
    def __reduce__(self) -> tuple[type[VerbatimError], tuple[Any, ...], dict[str, Any]]:
        """Keep the keyword attributes when the error crosses a process boundary."""
        return type(self), self.args, self.__dict__
__init__(message, *, code=None)

Create the error.

Parameters:

Name Type Description Default
message str

What the problem is and how to fix it.

 required
code str | None

A stable snake_case code, in place of the default code of the class.

 None
Source code in packages/core/verbatim_core/errors.py 
15
16
17
18
19
20
21
22
23
24
25
def __init__(self, message: str, *, code: str | None = None) -> None:
    """Create the error.

    Args:
        message: What the problem is and how to fix it.
        code: A stable snake_case code, in place of the default code of the class.

    """
    super().__init__(message)
    if code is not None:
        self.code = code
__reduce__()

Keep the keyword attributes when the error crosses a process boundary.

Source code in packages/core/verbatim_core/errors.py 
27
28
29
30
@override
def __reduce__(self) -> tuple[type[VerbatimError], tuple[Any, ...], dict[str, Any]]:
    """Keep the keyword attributes when the error crosses a process boundary."""
    return type(self), self.args, self.__dict__

InvalidInputError

Bases: VerbatimError, ValueError

The caller sent something wrong. Only a different input fixes it.

Source code in packages/core/verbatim_core/errors.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
class InvalidInputError(VerbatimError, ValueError):
    """The caller sent something wrong. Only a different input fixes it."""

    code = "invalid_input"

    def __init__(self, message: str, *, code: str | None = None, param: str | None = None) -> None:
        """Create the error.

        Args:
            message: What the problem is and how to fix it.
            code: A stable snake_case code, in place of the default code of the class.
            param: The argument that holds the bad value, for example `chunks[3].id`.

        """
        super().__init__(message, code=code)
        self.param = param
__init__(message, *, code=None, param=None)

Create the error.

Parameters:

Name Type Description Default
message str

What the problem is and how to fix it.

 required
code str | None

A stable snake_case code, in place of the default code of the class.

 None
param str | None

The argument that holds the bad value, for example chunks[3].id.

 None
Source code in packages/core/verbatim_core/errors.py 
38
39
40
41
42
43
44
45
46
47
48
def __init__(self, message: str, *, code: str | None = None, param: str | None = None) -> None:
    """Create the error.

    Args:
        message: What the problem is and how to fix it.
        code: A stable snake_case code, in place of the default code of the class.
        param: The argument that holds the bad value, for example `chunks[3].id`.

    """
    super().__init__(message, code=code)
    self.param = param

ConfigurationError

Bases: VerbatimError

The deployment is wrong. Only the operator can fix it.

Source code in packages/core/verbatim_core/errors.py 
51
52
53
54
class ConfigurationError(VerbatimError):
    """The deployment is wrong. Only the operator can fix it."""

    code = "configuration_error"

UpstreamError

Bases: VerbatimError

A service the library calls failed or did not answer in time.

Source code in packages/core/verbatim_core/errors.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
class UpstreamError(VerbatimError):
    """A service the library calls failed or did not answer in time."""

    code = "upstream_error"

    def __init__(
        self,
        message: str,
        *,
        code: str | None = None,
        service: str = "unknown",
        retryable: bool = False,
        retry_after: float | None = None,
    ) -> None:
        """Create the error.

        Args:
            message: What the problem is and how to fix it.
            code: A stable snake_case code, in place of the default code of the class.
            service: The service that failed, for example `llm`.
            retryable: Whether the same call may succeed when retried.
            retry_after: Seconds to wait before retrying, when the service said so.

        """
        super().__init__(message, code=code)
        self.service = service
        self.retryable = retryable
        self.retry_after = retry_after
__init__(message, *, code=None, service='unknown', retryable=False, retry_after=None)

Create the error.

Parameters:

Name Type Description Default
message str

What the problem is and how to fix it.

 required
code str | None

A stable snake_case code, in place of the default code of the class.

 None
service str

The service that failed, for example llm.

 'unknown'
retryable bool

Whether the same call may succeed when retried.

 False
retry_after float | None

Seconds to wait before retrying, when the service said so.

 None
Source code in packages/core/verbatim_core/errors.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def __init__(
    self,
    message: str,
    *,
    code: str | None = None,
    service: str = "unknown",
    retryable: bool = False,
    retry_after: float | None = None,
) -> None:
    """Create the error.

    Args:
        message: What the problem is and how to fix it.
        code: A stable snake_case code, in place of the default code of the class.
        service: The service that failed, for example `llm`.
        retryable: Whether the same call may succeed when retried.
        retry_after: Seconds to wait before retrying, when the service said so.

    """
    super().__init__(message, code=code)
    self.service = service
    self.retryable = retryable
    self.retry_after = retry_after

Deprecation warning

verbatim_core._deprecation.VerbatimDeprecationWarning

Bases: DeprecationWarning

Emitted when code uses a part of the API that will be removed.

Source code in packages/core/verbatim_core/_deprecation.py 
28
29
class VerbatimDeprecationWarning(DeprecationWarning):
    """Emitted when code uses a part of the API that will be removed."""

UniversalDocument

verbatim_core.universal_document.UniversalDocument dataclass

Minimal document with content, title, source and metadata.

Source code in packages/core/verbatim_core/universal_document.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
@dataclass
class UniversalDocument:
    """Minimal document with content, title, source and metadata."""

    content: str
    title: str = ""
    source: str = ""
    metadata: dict[str, Any] = field(default_factory=dict)

    @classmethod
    def from_text(
        cls,
        text: str,
        title: str = "",
        source: str = "",
        metadata: dict[str, Any] | None = None,
    ) -> UniversalDocument:
        """Create a document from raw text."""
        return cls(content=text, title=title, source=source, metadata=metadata or {})

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> UniversalDocument:
        """Create a document from a dict with ``content`` (or ``text``)."""
        if not isinstance(data, dict):
            msg = "UniversalDocument.from_dict expects a dict"
            raise TypeError(msg)
        content = data.get("content") or data.get("text")
        if not isinstance(content, str) or not content:
            msg = "UniversalDocument requires 'content' (or 'text') as non-empty string"
            raise ValueError(msg)
        return cls(
            content=content,
            title=data.get("title", ""),
            source=data.get("source", ""),
            metadata=data.get("metadata") or {},
        )

    def to_context(self) -> dict[str, Any]:
        """Convert the document to a context dict."""
        return {
            "content": self.content,
            "title": self.title,
            "source": self.source,
            "metadata": self.metadata,
        }

from_text(text, title='', source='', metadata=None) classmethod

Create a document from raw text.

Source code in packages/core/verbatim_core/universal_document.py 
18
19
20
21
22
23
24
25
26
27
@classmethod
def from_text(
    cls,
    text: str,
    title: str = "",
    source: str = "",
    metadata: dict[str, Any] | None = None,
) -> UniversalDocument:
    """Create a document from raw text."""
    return cls(content=text, title=title, source=source, metadata=metadata or {})

from_dict(data) classmethod

Create a document from a dict with content (or text).

Source code in packages/core/verbatim_core/universal_document.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
@classmethod
def from_dict(cls, data: dict[str, Any]) -> UniversalDocument:
    """Create a document from a dict with ``content`` (or ``text``)."""
    if not isinstance(data, dict):
        msg = "UniversalDocument.from_dict expects a dict"
        raise TypeError(msg)
    content = data.get("content") or data.get("text")
    if not isinstance(content, str) or not content:
        msg = "UniversalDocument requires 'content' (or 'text') as non-empty string"
        raise ValueError(msg)
    return cls(
        content=content,
        title=data.get("title", ""),
        source=data.get("source", ""),
        metadata=data.get("metadata") or {},
    )

to_context()

Convert the document to a context dict.

Source code in packages/core/verbatim_core/universal_document.py 
46
47
48
49
50
51
52
53
def to_context(self) -> dict[str, Any]:
    """Convert the document to a context dict."""
    return {
        "content": self.content,
        "title": self.title,
        "source": self.source,
        "metadata": self.metadata,
    }