verbatim_core API Reference

VerbatimTransform

verbatim_core.transform.VerbatimTransform

RAG-agnostic verbatim transform using existing components (sync/async).

Source code in packages/core/verbatim_core/transform.py

class VerbatimTransform:
    """RAG-agnostic verbatim transform using existing components (sync/async)."""

    def __init__(
        self,
        llm_client: LLMClient | None = None,
        extractor: SpanExtractor | None = None,
        template_manager: TemplateManager | None = None,
        max_display_spans: int = 5,
        extraction_mode: str = "auto",
        template_mode: str = "contextual",
    ):
        self.llm_client = llm_client or LLMClient()
        self.extractor = extractor or LLMSpanExtractor(
            llm_client=self.llm_client,
            extraction_mode=extraction_mode,
            max_display_spans=max_display_spans,
        )
        self.template_manager = template_manager or TemplateManager(
            llm_client=self.llm_client, default_mode=template_mode
        )
        self.response_builder = ResponseBuilder()
        self.max_display_spans = max_display_spans

    def transform(
        self,
        question: str,
        context: Iterable[Dict[str, Any]],
        answer: str | None = None,  # Ignored for now
    ):
        search_results = _coerce_context_to_results(list(context))

        relevant_spans = self.extractor.extract_spans(question, search_results)  # type: ignore[arg-type]

        all_spans = []
        for doc_text, spans in relevant_spans.items():
            for span in spans:
                all_spans.append({"text": span, "doc_text": doc_text})
        display_spans = all_spans[: self.max_display_spans]
        citation_spans = all_spans[self.max_display_spans :]

        answer_text = self.template_manager.process(question, display_spans, citation_spans)
        answer_text = self.response_builder.clean_answer(answer_text)

        return self.response_builder.build_response(
            question=question,
            answer=answer_text,
            search_results=search_results,  # compatible with response builder expectations
            relevant_spans=relevant_spans,
            display_span_count=len(display_spans),
        )

    async def transform_async(
        self,
        question: str,
        context: Iterable[Dict[str, Any]],
        answer: str | None = None,
    ):
        search_results = _coerce_context_to_results(list(context))

        relevant_spans = await self.extractor.extract_spans_async(  # type: ignore[attr-defined]
            question,
            search_results,  # type: ignore[arg-type]
        )

        all_spans = []
        for doc_text, spans in relevant_spans.items():
            for span in spans:
                all_spans.append({"text": span, "doc_text": doc_text})
        display_spans = all_spans[: self.max_display_spans]
        citation_spans = all_spans[self.max_display_spans :]

        answer_text = await self.template_manager.process_async(  # type: ignore[attr-defined]
            question, display_spans, citation_spans
        )
        answer_text = self.response_builder.clean_answer(answer_text)

        return self.response_builder.build_response(
            question=question,
            answer=answer_text,
            search_results=search_results,
            relevant_spans=relevant_spans,
            display_span_count=len(display_spans),
        )

LLMSpanExtractor

verbatim_core.extractors.LLMSpanExtractor

Bases: SpanExtractor

Extract spans using an LLM with centralized client and batch processing.

Source code in packages/core/verbatim_core/extractors.py

class LLMSpanExtractor(SpanExtractor):
    """Extract spans using an LLM with centralized client and batch processing."""

    def __init__(
        self,
        llm_client: LLMClient | None = None,
        model: str = "gpt-4o-mini",
        extraction_mode: str = "auto",
        max_display_spans: int = 5,
        batch_size: int = 5,
    ):
        """
        Initialize the LLM span extractor.

        :param llm_client: LLM client for extraction (creates one if None)
        :param model: The LLM model to use (if creating new client)
        :param extraction_mode: "batch", "individual", or "auto"
        :param max_display_spans: Maximum spans to prioritize for display
        :param batch_size: Maximum documents to process in batch mode
        """
        self.llm_client = llm_client or LLMClient(model)
        self.extraction_mode = extraction_mode
        self.max_display_spans = max_display_spans
        self.batch_size = batch_size

    def extract_spans(self, question: str, search_results: List[Any]) -> Dict[str, List[str]]:
        """
        Extract spans using LLM with mode selection.

        :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
        """
        if not search_results:
            return {}

        # Decide on processing mode
        should_batch = self.extraction_mode == "batch" or (
            self.extraction_mode == "auto" and len(search_results) <= self.batch_size
        )

        if should_batch:
            return self._extract_spans_batch(question, search_results)
        else:
            return self._extract_spans_individual(question, search_results)

    async def extract_spans_async(
        self, question: str, search_results: List[Any]
    ) -> Dict[str, List[str]]:
        """
        Async version of span extraction.

        :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
        """
        if not search_results:
            return {}

        should_batch = self.extraction_mode == "batch" or (
            self.extraction_mode == "auto" and len(search_results) <= self.batch_size
        )

        if should_batch:
            return await self._extract_spans_batch_async(question, search_results)
        else:
            return await self._extract_spans_individual_async(question, search_results)

    def _extract_spans_batch(
        self, question: str, search_results: List[Any]
    ) -> Dict[str, List[str]]:
        """
        Extract spans from multiple documents using batch processing.
        """
        print("Extracting spans (batch mode)...")

        # Limit to batch_size to avoid prompt size issues
        top_results = search_results[: self.batch_size]

        # Build document mapping for LLMClient
        documents_text = {}
        for i, result in enumerate(top_results):
            documents_text[f"doc_{i}"] = getattr(result, "text", "")

        try:
            # Use LLMClient for extraction
            extracted_data = self.llm_client.extract_spans(question, documents_text)

            # Map back to original search results and verify spans
            verified_spans = {}

            # Process documents that were included in batch
            for i, result in enumerate(top_results):
                doc_key = f"doc_{i}"
                result_text = getattr(result, "text", "")
                if doc_key in extracted_data:
                    verified = self._verify_spans(extracted_data[doc_key], result_text)
                    verified_spans[result_text] = verified
                else:
                    verified_spans[result_text] = []

            # Handle remaining documents (beyond batch_size) with empty spans
            for i in range(self.batch_size, len(search_results)):
                verified_spans[getattr(search_results[i], "text", "")] = []

            return verified_spans

        except Exception as e:
            print(f"Batch extraction failed, falling back to individual: {e}")
            return self._extract_spans_individual(question, search_results)

    async def _extract_spans_batch_async(
        self, question: str, search_results: List[Any]
    ) -> Dict[str, List[str]]:
        """
        Async batch extraction.
        """
        print("Extracting spans (async batch mode)...")

        top_results = search_results[: self.batch_size]

        documents_text = {}
        for i, result in enumerate(top_results):
            documents_text[f"doc_{i}"] = getattr(result, "text", "")

        try:
            extracted_data = await self.llm_client.extract_spans_async(question, documents_text)

            verified_spans = {}

            for i, result in enumerate(top_results):
                doc_key = f"doc_{i}"
                result_text = getattr(result, "text", "")
                if doc_key in extracted_data:
                    verified = self._verify_spans(extracted_data[doc_key], result_text)
                    verified_spans[result_text] = verified
                else:
                    verified_spans[result_text] = []

            for i in range(self.batch_size, len(search_results)):
                verified_spans[getattr(search_results[i], "text", "")] = []

            return verified_spans

        except Exception as e:
            print(f"Async batch extraction failed, falling back to individual: {e}")
            return await self._extract_spans_individual_async(question, search_results)

    def _extract_spans_individual(
        self, question: str, search_results: List[Any]
    ) -> Dict[str, List[str]]:
        """
        Extract spans from documents individually (one at a time).

        :param question: The query or question
        :param search_results: List of search results to process
        :return: Dictionary mapping result text to list of relevant spans
        """
        print("Extracting spans (individual mode)...")
        all_spans = {}

        for result in search_results:
            result_text = getattr(result, "text", "")
            try:
                extracted_spans = self.llm_client.extract_relevant_spans(question, result_text)
                verified = self._verify_spans(extracted_spans, result_text)
                all_spans[result_text] = verified
            except Exception as e:
                print(f"Individual extraction failed for document: {e}")
                all_spans[result_text] = []

        return all_spans

    async def _extract_spans_individual_async(
        self, question: str, search_results: List[Any]
    ) -> Dict[str, List[str]]:
        """
        Async individual extraction.
        """
        print("Extracting spans (async individual mode)...")
        all_spans = {}

        for result in search_results:
            result_text = getattr(result, "text", "")
            try:
                extracted_spans = await self.llm_client.extract_relevant_spans_async(
                    question, result_text
                )
                verified = self._verify_spans(extracted_spans, result_text)
                all_spans[result_text] = verified
            except Exception as e:
                print(f"Async individual extraction failed for document: {e}")
                all_spans[result_text] = []

        return all_spans

    def _verify_spans(self, spans: List[str], document_text: str) -> List[str]:
        """
        Verify that extracted spans actually exist in the document text.

        :param spans: List of spans to verify
        :param document_text: Original document text
        :return: List of verified spans that exist in the document
        """
        verified = []
        for span in spans:
            if span.strip() and span.strip() in document_text:
                verified.append(span.strip())
            else:
                print(f"Warning: Span not found verbatim in document: '{span[:100]}...'")
        return verified

__init__(llm_client=None, model='gpt-4o-mini', extraction_mode='auto', max_display_spans=5, batch_size=5)

Initialize the LLM span extractor.

:param llm_client: LLM client for extraction (creates one if None) :param model: The LLM model to use (if creating new client) :param extraction_mode: "batch", "individual", or "auto" :param max_display_spans: Maximum spans to prioritize for display :param batch_size: Maximum documents to process in batch mode

Source code in packages/core/verbatim_core/extractors.py

def __init__(
    self,
    llm_client: LLMClient | None = None,
    model: str = "gpt-4o-mini",
    extraction_mode: str = "auto",
    max_display_spans: int = 5,
    batch_size: int = 5,
):
    """
    Initialize the LLM span extractor.

    :param llm_client: LLM client for extraction (creates one if None)
    :param model: The LLM model to use (if creating new client)
    :param extraction_mode: "batch", "individual", or "auto"
    :param max_display_spans: Maximum spans to prioritize for display
    :param batch_size: Maximum documents to process in batch mode
    """
    self.llm_client = llm_client or LLMClient(model)
    self.extraction_mode = extraction_mode
    self.max_display_spans = max_display_spans
    self.batch_size = batch_size

extract_spans(question, search_results)

Extract spans using LLM with mode selection.

: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

def extract_spans(self, question: str, search_results: List[Any]) -> Dict[str, List[str]]:
    """
    Extract spans using LLM with mode selection.

    :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
    """
    if not search_results:
        return {}

    # Decide on processing mode
    should_batch = self.extraction_mode == "batch" or (
        self.extraction_mode == "auto" and len(search_results) <= self.batch_size
    )

    if should_batch:
        return self._extract_spans_batch(question, search_results)
    else:
        return self._extract_spans_individual(question, search_results)

extract_spans_async(question, search_results) async

Async version of span extraction.

: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

async def extract_spans_async(
    self, question: str, search_results: List[Any]
) -> Dict[str, List[str]]:
    """
    Async version of span extraction.

    :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
    """
    if not search_results:
        return {}

    should_batch = self.extraction_mode == "batch" or (
        self.extraction_mode == "auto" and len(search_results) <= self.batch_size
    )

    if should_batch:
        return await self._extract_spans_batch_async(question, search_results)
    else:
        return await self._extract_spans_individual_async(question, search_results)

SpanExtractor

verbatim_core.extractors.SpanExtractor

Bases: ABC

Abstract base class for span extractors.

Source code in packages/core/verbatim_core/extractors.py

class SpanExtractor(ABC):
    """Abstract base class for span extractors."""

    @abstractmethod
    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
        """
        raise NotImplementedError

    async def extract_spans_async(
        self, question: str, search_results: List[Any]
    ) -> Dict[str, List[str]]:
        """Default async implementation that delegates to sync version."""
        import asyncio

        return await asyncio.to_thread(self.extract_spans, question, search_results)

extract_spans(question, search_results) abstractmethod

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

@abstractmethod
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
    """
    raise NotImplementedError

extract_spans_async(question, search_results) async

Default async implementation that delegates to sync version.

Source code in packages/core/verbatim_core/extractors.py

async def extract_spans_async(
    self, question: str, search_results: List[Any]
) -> Dict[str, List[str]]:
    """Default async implementation that delegates to sync version."""
    import asyncio

    return await asyncio.to_thread(self.extract_spans, question, search_results)

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

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",
    ):
        """
        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)
        """
        self.model = model
        self.temperature = temperature
        self.api_key = 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, temperature: Optional[float] = None
    ) -> str:
        """
        Synchronous text completion.

        :param prompt: The prompt to send
        :param json_mode: Whether to request JSON output format
        :param temperature: Override default temperature
        :return: The completion text
        """
        messages = [{"role": "user", "content": prompt}]
        kwargs = {
            "model": self.model,
            "messages": messages,
            "temperature": temperature if temperature is not None else self.temperature,
        }

        if json_mode:
            kwargs["response_format"] = {"type": "json_object"}

        response = self.client.chat.completions.create(**kwargs)
        return response.choices[0].message.content

    async def complete_async(
        self, prompt: str, json_mode: bool = False, temperature: Optional[float] = None
    ) -> str:
        """
        Asynchronous text completion.

        :param prompt: The prompt to send
        :param json_mode: Whether to request JSON output format
        :param temperature: Override default temperature
        :return: The completion text
        """
        messages = [{"role": "user", "content": prompt}]
        kwargs = {
            "model": self.model,
            "messages": messages,
            "temperature": temperature if temperature is not None else self.temperature,
        }

        if json_mode:
            kwargs["response_format"] = {"type": "json_object"}

        response = await self.async_client.chat.completions.create(**kwargs)
        return response.choices[0].message.content

    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:
            print(f"Span extraction failed: {e}")
            # Return empty results for all documents on failure
            return {doc_id: [] for doc_id in documents.keys()}

    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:
            print(f"Async span extraction failed: {e}")
            return {doc_id: [] for doc_id in documents.keys()}

    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:
            print(f"Structured extraction failed: {e}")
            return {name: [] for name in placeholders.keys()}

    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:
            print(f"Structured extraction failed: {e}")
            return {name: [] for name in placeholders.keys()}

    def _normalize_structured_response(
        self, response: Dict, 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.keys():
            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 f"""Extract verbatim spans from the documents for each placeholder in the template.

Question: {question}

Template to fill:
{template}

Placeholders to extract for:
{placeholder_spec}

Documents:
{docs_text}

Instructions:
1. For each placeholder, find EXACT verbatim quotes from the documents
2. Copy text exactly - no paraphrasing or modification
3. For each span, include which document it came from (0-indexed)
4. Return a JSON object mapping placeholder names to arrays of objects with "text" and "doc" fields
5. If no relevant information for a placeholder, use an empty array

Return ONLY valid JSON like:
{{
  "METHODOLOGY": [{{"text": "exact quote about methods...", "doc": 0}}],
  "RESULTS": [{{"text": "exact quote about results...", "doc": 1}}]
}}"""

    def generate_template(
        self,
        question: str,
        spans: List[str],
        citation_count: int,
        use_per_fact: bool = True,
    ) -> 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-fact placeholders
        :return: Generated template string
        """
        if use_per_fact and len(spans) <= 8:
            prompt = self._build_per_fact_template_prompt(question, spans, citation_count)
        else:
            prompt = self._build_aggregate_template_prompt(question, spans, citation_count)

        try:
            return self.complete(prompt, temperature=self.temperature)
        except Exception as e:
            print(f"Template generation failed: {e}")
            return self._fallback_template(citation_count > 0)

    async def generate_template_async(
        self,
        question: str,
        spans: List[str],
        citation_count: int,
        use_per_fact: bool = True,
    ) -> str:
        """
        Async template generation.

        :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-fact placeholders
        :return: Generated template string
        """
        if use_per_fact and len(spans) <= 8:
            prompt = self._build_per_fact_template_prompt(question, spans, citation_count)
        else:
            prompt = self._build_aggregate_template_prompt(question, spans, citation_count)

        try:
            return await self.complete_async(prompt, temperature=self.temperature)
        except Exception as e:
            print(f"Async template generation failed: {e}")
            return self._fallback_template(citation_count > 0)

    def _build_extraction_prompt(self, question: str, documents: Dict[str, str]) -> str:
        """Build the prompt for batch span extraction."""
        return f"""Extract EXACT verbatim text spans from multiple documents that answer the question.

# Rules
1. Extract **only** text that explicitly addresses the question
2. Never paraphrase, modify, or add to the original text
3. Preserve original wording, capitalization, and punctuation
4. Order spans within each document by relevance - MOST RELEVANT FIRST
5. Include complete sentences or paragraphs for context

# Output Format
Return a JSON object mapping document IDs to span arrays ordered by relevance:
{{
  "doc_0": ["most relevant span", "next most relevant span"],
  "doc_1": ["most relevant from doc 1"],
  "doc_2": []
}}

If no relevant information in a document, use empty array.

# Your Task
Question: {question}

Documents:
{json.dumps(documents, indent=2)}

Extract verbatim spans from each document:"""

    def _build_per_fact_template_prompt(
        self, question: str, spans: List[str], citation_count: int
    ) -> str:
        """Build prompt for per-fact template generation."""
        span_lines = []
        for i, span in enumerate(spans, start=1):
            clean = span.replace("\n", " ").strip()[:100]  # Truncate for prompt size
            span_lines.append(f"{i}. {clean}...")
        spans_block = "\n".join(span_lines)

        return f"""Generate a response template for this Q&A scenario:

Question: {question}

Content that will be inserted into the template:
- Total verbatim facts to show (display facts): {len(spans)}
- Full list of verbatim facts:
{spans_block}
- Additional citation-only facts (only numbers, no text shown): {citation_count}

Template strategy rules:
- Use per-fact placeholders [FACT_1]..[FACT_{len(spans)}] each exactly once.
- If citation-only facts exist, you MAY place [CITATION_REFS] exactly once where their numbers should appear, otherwise omit it.

Instructions:
- Intro: 1 concise sentence tying question to facts.
- Then present each fact in a structured way (bulleted list or numbered list). Each list item should contain exactly one placeholder at the start after a bold label you infer or a generic label (e.g. Fact 3) if unsure.
- DO NOT invent content beyond connective phrases; never summarize or paraphrase inside placeholders.
- No duplicate placeholders; no placeholder inside a heading alone.
- Avoid leading a bullet list with another nested bullet list.

Template requirements:
- Use only placeholders plus minimal connective prose (no actual span text).
- {"Include [CITATION_REFS] once" if citation_count > 0 else "Do NOT include [CITATION_REFS]"}.
- End without extra commentary like "Hope this helps".

Return ONLY the template text (no explanation)."""

    def _build_aggregate_template_prompt(
        self, question: str, spans: List[str], citation_count: int
    ) -> str:
        """Build prompt for aggregate template generation."""
        span_preview = " | ".join(span[:50] + "..." for span in spans[:3])

        return f"""Generate a response template for this Q&A scenario (OUTPUT MUST BE VALID GITHUB-FLAVORED MARKDOWN):

Question: {question}

Content that will be inserted into the template:
- Total verbatim facts to show (display facts): {len(spans)}
- Preview of content: {span_preview}
- Additional citation-only facts (only numbers, no text shown): {citation_count}

Template strategy rules (Markdown correctness is critical):
- Use [DISPLAY_SPANS] exactly once for the aggregate of all verbatim spans.
- If citation-only facts exist, you MAY place [CITATION_REFS] exactly once where their numbers should appear, otherwise omit it.

Markdown formatting requirements:
- Use only GitHub-Flavored Markdown (GFM): headings (##, ###), paragraphs, bullet/numbered lists, bold/italic, blockquotes, and tables.
- Do NOT wrap the entire template in code fences.
- Every heading must be followed by a blank line unless immediately followed by a list.
- Placeholders must not be inside backticks, code blocks, or HTML tags.

Instructions:
- Intro: 1 concise sentence tying question to spans.
- Provide a section header then include the aggregate placeholder.
- Do NOT invent or paraphrase span content; placeholders stand in for verbatim content only.
- Avoid nested lists; keep structure shallow and clean.

Template requirements:
- Must contain [DISPLAY_SPANS].
- {"Include [CITATION_REFS] once" if citation_count > 0 else "Do NOT include [CITATION_REFS]"}.
- End without extra commentary like "Hope this helps".

Return ONLY the template text (no explanation)."""

    def _fallback_template(self, has_citations: bool = False) -> str:
        """Return a simple fallback template when generation fails."""
        template = """## Response

Based on the available documents:

[DISPLAY_SPANS]"""

        if has_citations:
            template += "\n\n**Additional References:** [CITATION_REFS]"

        template += "\n\n---\n*These excerpts are taken verbatim from the source documents to ensure accuracy.*"

        return template

    # Batch span extraction API (for compatibility)
    def extract_relevant_spans_batch(
        self, question: str, documents: Dict[str, str]
    ) -> Dict[str, List[str]]:
        return self.extract_spans(question, documents)

    async def extract_relevant_spans_batch_async(
        self, question: str, documents: Dict[str, str]
    ) -> Dict[str, List[str]]:
        return await self.extract_spans_async(question, documents)

    # Single-doc convenience
    def extract_relevant_spans(self, question: str, document_text: str) -> List[str]:
        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]:
        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:
        return self.complete(prompt)

    async def simple_complete_async(self, prompt: str) -> str:
        return await self.complete_async(prompt)

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

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)

Source code in packages/core/verbatim_core/llm_client.py

def __init__(
    self,
    model: str = "gpt-4o-mini",
    temperature: float = 0.7,
    api_base: str = "https://api.openai.com/v1",
):
    """
    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)
    """
    self.model = model
    self.temperature = temperature
    self.api_key = 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)

Synchronous text completion.

:param prompt: The prompt to send :param json_mode: Whether to request JSON output format :param temperature: Override default temperature :return: The completion text

Source code in packages/core/verbatim_core/llm_client.py

def complete(
    self, prompt: str, json_mode: bool = False, temperature: Optional[float] = None
) -> str:
    """
    Synchronous text completion.

    :param prompt: The prompt to send
    :param json_mode: Whether to request JSON output format
    :param temperature: Override default temperature
    :return: The completion text
    """
    messages = [{"role": "user", "content": prompt}]
    kwargs = {
        "model": self.model,
        "messages": messages,
        "temperature": temperature if temperature is not None else self.temperature,
    }

    if json_mode:
        kwargs["response_format"] = {"type": "json_object"}

    response = self.client.chat.completions.create(**kwargs)
    return response.choices[0].message.content

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

Asynchronous text completion.

:param prompt: The prompt to send :param json_mode: Whether to request JSON output format :param temperature: Override default temperature :return: The completion text

Source code in packages/core/verbatim_core/llm_client.py

async def complete_async(
    self, prompt: str, json_mode: bool = False, temperature: Optional[float] = None
) -> str:
    """
    Asynchronous text completion.

    :param prompt: The prompt to send
    :param json_mode: Whether to request JSON output format
    :param temperature: Override default temperature
    :return: The completion text
    """
    messages = [{"role": "user", "content": prompt}]
    kwargs = {
        "model": self.model,
        "messages": messages,
        "temperature": temperature if temperature is not None else self.temperature,
    }

    if json_mode:
        kwargs["response_format"] = {"type": "json_object"}

    response = await self.async_client.chat.completions.create(**kwargs)
    return response.choices[0].message.content

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

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:
        print(f"Span extraction failed: {e}")
        # Return empty results for all documents on failure
        return {doc_id: [] for doc_id in documents.keys()}

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

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:
        print(f"Async span extraction failed: {e}")
        return {doc_id: [] for doc_id in documents.keys()}

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

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:
        print(f"Structured extraction failed: {e}")
        return {name: [] for name in placeholders.keys()}

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

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:
        print(f"Structured extraction failed: {e}")
        return {name: [] for name in placeholders.keys()}

generate_template(question, spans, citation_count, use_per_fact=True)

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-fact placeholders :return: Generated template string

Source code in packages/core/verbatim_core/llm_client.py

def generate_template(
    self,
    question: str,
    spans: List[str],
    citation_count: int,
    use_per_fact: bool = True,
) -> 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-fact placeholders
    :return: Generated template string
    """
    if use_per_fact and len(spans) <= 8:
        prompt = self._build_per_fact_template_prompt(question, spans, citation_count)
    else:
        prompt = self._build_aggregate_template_prompt(question, spans, citation_count)

    try:
        return self.complete(prompt, temperature=self.temperature)
    except Exception as e:
        print(f"Template generation failed: {e}")
        return self._fallback_template(citation_count > 0)

generate_template_async(question, spans, citation_count, use_per_fact=True) async

Async template generation.

: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-fact placeholders :return: Generated template string

Source code in packages/core/verbatim_core/llm_client.py

async def generate_template_async(
    self,
    question: str,
    spans: List[str],
    citation_count: int,
    use_per_fact: bool = True,
) -> str:
    """
    Async template generation.

    :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-fact placeholders
    :return: Generated template string
    """
    if use_per_fact and len(spans) <= 8:
        prompt = self._build_per_fact_template_prompt(question, spans, citation_count)
    else:
        prompt = self._build_aggregate_template_prompt(question, spans, citation_count)

    try:
        return await self.complete_async(prompt, temperature=self.temperature)
    except Exception as e:
        print(f"Async template generation failed: {e}")
        return self._fallback_template(citation_count > 0)

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

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):
        """Initialize the response builder."""
        pass

    def build_response(
        self,
        question: str,
        answer: str,
        search_results: List[Any],
        relevant_spans: Dict[str, List[str]],
        display_span_count: int = 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 _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
        """
        for region_start, region_end in regions:
            # Check for overlap: new region starts before old ends and ends after old starts
            if start < region_end and end > region_start:
                return True
        return False

    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('"'):
            answer = answer[1:-1]
        elif answer.startswith("'") and answer.endswith("'"):
            answer = answer[1:-1]

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

        # Clean up multiple spaces
        import re

        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

def __init__(self):
    """Initialize the response builder."""
    pass

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

def build_response(
    self,
    question: str,
    answer: str,
    search_results: List[Any],
    relevant_spans: Dict[str, List[str]],
    display_span_count: int = 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,
    )

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

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('"'):
        answer = answer[1:-1]
    elif answer.startswith("'") and answer.endswith("'"):
        answer = answer[1:-1]

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

    # Clean up multiple spaces
    import re

    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.

Source code in packages/core/verbatim_core/templates/manager.py

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.
    """

    def __init__(
        self,
        llm_client: Optional[LLMClient] = None,
        default_mode: str = "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
        """
        self.llm_client = llm_client
        self.rag_system = rag_system
        self.current_mode = default_mode
        self.citation_mode = "inline"

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

        # Validate initial mode
        if self.current_mode not in self.strategies:
            self.current_mode = "static"

        if self.strategies[self.current_mode] is None:
            print(f"Warning: {self.current_mode} mode requires LLM client, falling back to static")
            self.current_mode = "static"

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

        :param mode: Template mode to switch to
        :return: True if mode was switched successfully
        """
        if mode not in self.strategies:
            print(f"Unknown template mode: {mode}")
            return False

        if self.strategies[mode] is None:
            print(f"Mode {mode} is not available (requires LLM client)")
            return False

        self.current_mode = mode
        return True

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

        :return: Current mode name
        """
        return self.current_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
        """
        # Extract span texts for template generation
        all_spans = [span["text"] for span in display_spans + citation_spans]
        citation_count = len(citation_spans)

        # Generate template
        strategy = self.strategies[self.current_mode]
        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 = [span["text"] for span in display_spans + citation_spans]
        citation_count = len(citation_spans)

        strategy = self.strategies[self.current_mode]

        # Use async generation if available
        if hasattr(strategy, "generate_async") and self.current_mode == "contextual":
            template = await 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 get_template(
        self, question: str = "", spans: List[str] = 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.strategies[self.current_mode]
        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.strategies[self.current_mode]
        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.current_mode, "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)

        with open(filepath, "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):
            print(f"Template config file not found: {filepath}")
            return False

        try:
            with open(filepath, "r") 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.current_mode = mode

            # Load strategy states
            strategies_data = data.get("strategies", {})
            for mode, state in strategies_data.items():
                if mode in self.strategies and self.strategies[mode] is not None:
                    try:
                        self.strategies[mode].load_state(state)
                    except Exception as e:
                        print(f"Warning: Failed to load state for {mode} strategy: {e}")

            return True

        except Exception as e:
            print(f"Failed to load template config: {e}")
            return False

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

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

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

        return info_data

    # Convenience methods for specific modes

    def use_static_mode(self, template: str = 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.set_mode("static")

    def use_contextual_mode(self, use_per_fact: bool = True) -> bool:
        """
        Switch to contextual mode with configuration.

        :param use_per_fact: Whether to use per-fact placeholders
        :return: True if switched successfully
        """
        if not self.llm_client:
            print("Contextual mode requires LLM client")
            return False

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

        contextual_strategy = self.strategies["contextual"]
        contextual_strategy.set_per_fact_mode(use_per_fact)

        return self.set_mode("contextual")

    def use_random_mode(self, templates: List[str] = 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.set_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.current_mode != "random":
            print("Must be in random mode to generate templates")
            return False

        random_strategy = self.strategies["random"]
        if hasattr(random_strategy, "generate_diverse_templates"):
            try:
                random_strategy.generate_diverse_templates(count)
                return True
            except Exception as e:
                print(f"Template generation failed: {e}")

        return False

    def use_question_specific_mode(
        self, templates: Optional[Dict[str, Dict[str, Any]]] = None
    ) -> bool:
        """
        Switch to question-specific mode with optional template definitions.

        :param templates: Optional dictionary mapping category names to template configs
                          Format: {
                              "category_name": {
                                  "template": "Template with [RELEVANT_SENTENCES]",
                                  "examples": ["Example question 1", "Example question 2"]
                              }
                          }
        :return: True if switched successfully
        """
        if templates:
            question_specific_strategy = QuestionSpecificTemplate()
            question_specific_strategy.set_question_templates(templates)
            self.strategies["question_specific"] = question_specific_strategy

        return self.set_mode("question_specific")

    def use_structured_mode(
        self,
        template: str = None,
        placeholder_mappings: Optional[Dict[str, str]] = None,
    ) -> bool:
        """
        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 = self.strategies.get("structured")

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

        # Update RAG system if not set
        if self.rag_system and not structured_strategy.rag_system:
            structured_strategy.set_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.set_mode("structured")

    def set_rag_system(self, rag_system) -> None:
        """
        Set the RAG system for modes that need it (structured).

        :param rag_system: RAG system instance
        """
        self.rag_system = rag_system

        # Update strategies that need RAG system
        if "structured" in self.strategies and self.strategies["structured"]:
            self.strategies["structured"].set_rag_system(rag_system)

    async def process_structured_async(
        self,
        question: str,
        template: Optional[str] = None,
        placeholder_mappings: Optional[Dict[str, str]] = None,
    ) -> str:
        """
        Convenience helper to run structured extraction.

        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
        ):
            raise ValueError("Structured mode unavailable")

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

        # 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.

        :param mode: Citation rendering mode ("inline" or "hidden")
        """
        allowed = {"inline", "hidden"}
        if mode not in allowed:
            raise ValueError(f"Unsupported citation mode: {mode}")

        self.citation_mode = mode

        for strategy in self.strategies.values():
            if strategy and hasattr(strategy, "set_citation_mode"):
                strategy.set_citation_mode(mode)

__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

def __init__(
    self,
    llm_client: Optional[LLMClient] = None,
    default_mode: str = "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
    """
    self.llm_client = llm_client
    self.rag_system = rag_system
    self.current_mode = default_mode
    self.citation_mode = "inline"

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

    # Validate initial mode
    if self.current_mode not in self.strategies:
        self.current_mode = "static"

    if self.strategies[self.current_mode] is None:
        print(f"Warning: {self.current_mode} mode requires LLM client, falling back to static")
        self.current_mode = "static"

set_mode(mode)

Switch to a different template mode.

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

Source code in packages/core/verbatim_core/templates/manager.py

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

    :param mode: Template mode to switch to
    :return: True if mode was switched successfully
    """
    if mode not in self.strategies:
        print(f"Unknown template mode: {mode}")
        return False

    if self.strategies[mode] is None:
        print(f"Mode {mode} is not available (requires LLM client)")
        return False

    self.current_mode = mode
    return True

get_current_mode()

Get the current template mode.

:return: Current mode name

Source code in packages/core/verbatim_core/templates/manager.py

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

    :return: Current mode name
    """
    return self.current_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

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

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
    """
    # Extract span texts for template generation
    all_spans = [span["text"] for span in display_spans + citation_spans]
    citation_count = len(citation_spans)

    # Generate template
    strategy = self.strategies[self.current_mode]
    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

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 = [span["text"] for span in display_spans + citation_spans]
    citation_count = len(citation_spans)

    strategy = self.strategies[self.current_mode]

    # Use async generation if available
    if hasattr(strategy, "generate_async") and self.current_mode == "contextual":
        template = await 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

def get_template(
    self, question: str = "", spans: List[str] = 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.strategies[self.current_mode]
    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

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.strategies[self.current_mode]
    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

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

    :param filepath: Path to save configuration
    """
    data = {"current_mode": self.current_mode, "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)

    with open(filepath, "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

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):
        print(f"Template config file not found: {filepath}")
        return False

    try:
        with open(filepath, "r") 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.current_mode = mode

        # Load strategy states
        strategies_data = data.get("strategies", {})
        for mode, state in strategies_data.items():
            if mode in self.strategies and self.strategies[mode] is not None:
                try:
                    self.strategies[mode].load_state(state)
                except Exception as e:
                    print(f"Warning: Failed to load state for {mode} strategy: {e}")

        return True

    except Exception as e:
        print(f"Failed to load template config: {e}")
        return False

info()

Get current template manager state information.

:return: Dictionary with current state info

Source code in packages/core/verbatim_core/templates/manager.py

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

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

    # Add mode-specific info
    if self.current_mode == "random":
        random_strategy = self.strategies["random"]
        if hasattr(random_strategy, "get_template_count"):
            info_data["random_template_count"] = 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

def use_static_mode(self, template: str = 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.set_mode("static")

use_contextual_mode(use_per_fact=True)

Switch to contextual mode with configuration.

:param use_per_fact: Whether to use per-fact placeholders :return: True if switched successfully

Source code in packages/core/verbatim_core/templates/manager.py

def use_contextual_mode(self, use_per_fact: bool = True) -> bool:
    """
    Switch to contextual mode with configuration.

    :param use_per_fact: Whether to use per-fact placeholders
    :return: True if switched successfully
    """
    if not self.llm_client:
        print("Contextual mode requires LLM client")
        return False

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

    contextual_strategy = self.strategies["contextual"]
    contextual_strategy.set_per_fact_mode(use_per_fact)

    return self.set_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

def use_random_mode(self, templates: List[str] = 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.set_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

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.current_mode != "random":
        print("Must be in random mode to generate templates")
        return False

    random_strategy = self.strategies["random"]
    if hasattr(random_strategy, "generate_diverse_templates"):
        try:
            random_strategy.generate_diverse_templates(count)
            return True
        except Exception as e:
            print(f"Template generation failed: {e}")

    return False

use_question_specific_mode(templates=None)

Switch to question-specific mode with optional template definitions.

:param templates: Optional dictionary mapping category names to template configs Format: { "category_name": { "template": "Template with [RELEVANT_SENTENCES]", "examples": ["Example question 1", "Example question 2"] } } :return: True if switched successfully

Source code in packages/core/verbatim_core/templates/manager.py

def use_question_specific_mode(
    self, templates: Optional[Dict[str, Dict[str, Any]]] = None
) -> bool:
    """
    Switch to question-specific mode with optional template definitions.

    :param templates: Optional dictionary mapping category names to template configs
                      Format: {
                          "category_name": {
                              "template": "Template with [RELEVANT_SENTENCES]",
                              "examples": ["Example question 1", "Example question 2"]
                          }
                      }
    :return: True if switched successfully
    """
    if templates:
        question_specific_strategy = QuestionSpecificTemplate()
        question_specific_strategy.set_question_templates(templates)
        self.strategies["question_specific"] = question_specific_strategy

    return self.set_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

def use_structured_mode(
    self,
    template: str = None,
    placeholder_mappings: Optional[Dict[str, str]] = None,
) -> bool:
    """
    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 = self.strategies.get("structured")

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

    # Update RAG system if not set
    if self.rag_system and not structured_strategy.rag_system:
        structured_strategy.set_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.set_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

def set_rag_system(self, rag_system) -> None:
    """
    Set the RAG system for modes that need it (structured).

    :param rag_system: RAG system instance
    """
    self.rag_system = rag_system

    # Update strategies that need RAG system
    if "structured" in self.strategies and self.strategies["structured"]:
        self.strategies["structured"].set_rag_system(rag_system)

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

Convenience helper to run structured extraction.

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

async def process_structured_async(
    self,
    question: str,
    template: Optional[str] = None,
    placeholder_mappings: Optional[Dict[str, str]] = None,
) -> str:
    """
    Convenience helper to run structured extraction.

    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
    ):
        raise ValueError("Structured mode unavailable")

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

    # 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.

:param mode: Citation rendering mode ("inline" or "hidden")

Source code in packages/core/verbatim_core/templates/manager.py

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

    :param mode: Citation rendering mode ("inline" or "hidden")
    """
    allowed = {"inline", "hidden"}
    if mode not in allowed:
        raise ValueError(f"Unsupported citation mode: {mode}")

    self.citation_mode = mode

    for strategy in self.strategies.values():
        if strategy and hasattr(strategy, "set_citation_mode"):
            strategy.set_citation_mode(mode)

Models

verbatim_core.models

Pydantic models for verbatim_core (RAG-agnostic).

UniversalDocument

verbatim_core.universal_document.UniversalDocument dataclass

Source code in packages/core/verbatim_core/universal_document.py

@dataclass
class UniversalDocument:
    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":
        return cls(content=text, title=title, source=source, metadata=metadata or {})

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

    def to_context(self) -> Dict[str, Any]:
        return {
            "content": self.content,
            "title": self.title,
            "source": self.source,
            "metadata": self.metadata,
        }