Integration: TypeSafe

Classify documents, rank documents, and route text with TypeSafe's System One decision models, such as Jev, in Haystack applications.

Authors

deepset

GitHub Repo PyPI Package

Overview
Installation
Usage
License

Overview

TypeSafe builds System One models, such as Jev, which make fast, structured decisions for software. Instead of generating text, a System One model answers typed questions about a text in a single request and returns calibrated probabilities:

choice: picks one label from a set of options.
score: rates the text on ordered levels.
noul: returns the probability that a yes/no question is true.

The typesafe-haystack package brings these models to Haystack through three components:

TypeSafeDocumentClassifier answers typed questions about each document and stores the answers in its metadata.
TypeSafeRanker ranks documents by their relevance to a query, scoring each one with a score question.
TypeSafeTextRouter routes a text to the output of the label the model picks for it, with an optional low_confidence output for uncertain answers.

All three components also work with servers that implement the TypeSafe API, such as Ollaya, which runs open decision models on your own machine.

To use TypeSafe’s hosted models, you’ll need an API key from the TypeSafe console. Add it as an environment variable, TYPESAFE_API_KEY. For more information, see the TypeSafe docs.

Installation

pip install typesafe-haystack

Usage

Classify documents

TypeSafeDocumentClassifier sends each document to the model and stores the answers under meta["typesafe"], keyed by question ID. Documents that can’t be classified are returned in failed_documents.

from haystack import Document
from haystack_integrations.components.classifiers.typesafe import TypeSafeDocumentClassifier

classifier = TypeSafeDocumentClassifier(
    questions={
        "department": {
            "type": "choice",
            "instructions": "Which department should handle this ticket?",
            "criteria": {"billing": None, "technical": None, "sales": None},
        },
        "urgency": {
            "type": "score",
            "instructions": "How urgent is this ticket?",
            "criteria": ["can wait", "this week", "today"],
        },
        "refund": {"type": "noul", "instructions": "Is the customer asking for a refund?"},
    },
)

result = classifier.run(documents=[Document(content="I was charged twice, please refund me today.")])
print(result["documents"][0].meta["typesafe"])
# {'department': {'type': 'choice', 'choice': 'billing', 'confidence': 0.9516,
#                 'probabilities': {'billing': 0.9677, 'technical': 0.0221, 'sales': 0.0102}},
#  'urgency': {'type': 'score', 'score': 1.948, 'confidence': 0.9389,
#              'legend': {'0': 'can wait', '1': 'this week', '2': 'today'},
#              'probabilities': {'0': 0.0113, '1': 0.0295, '2': 0.9593}},
#  'refund': {'type': 'noul', 'noul': 0.9498}}

To send each document down a different branch of a pipeline based on its answers, connect the classifier to a MetadataRouter:

from haystack import Document, Pipeline
from haystack.components.routers import MetadataRouter
from haystack_integrations.components.classifiers.typesafe import TypeSafeDocumentClassifier

labels = ["billing", "technical"]

pipeline = Pipeline()
pipeline.add_component(
    "classifier",
    TypeSafeDocumentClassifier(
        questions={
            "department": {
                "type": "choice",
                "instructions": "Which department should handle this ticket?",
                "criteria": dict.fromkeys(labels),
            }
        },
    ),
)
pipeline.add_component(
    "router",
    MetadataRouter(
        rules={
            label: {"field": "meta.typesafe.department.choice", "operator": "==", "value": label}
            for label in labels
        }
    ),
)
pipeline.connect("classifier.documents", "router.documents")

result = pipeline.run(
    {
        "classifier": {
            "documents": [
                Document(content="I was charged twice for my subscription."),
                Document(content="The app crashes every time I open the settings page."),
            ]
        }
    }
)
print({output: [document.content for document in documents] for output, documents in result["router"].items()})
# {'billing': ['I was charged twice for my subscription.'],
#  'technical': ['The app crashes every time I open the settings page.'],
#  'unmatched': []}

Rank documents

TypeSafeRanker scores each document’s relevance to the query from 0 to 1 and returns the documents sorted from most to least relevant. The raw answer is stored under meta["typesafe"]["relevance"], next to any answers from a TypeSafeDocumentClassifier. Each document is a separate request, so place the ranker after a retriever with a small top_k.

from haystack import Document, Pipeline
from haystack.components.retrievers.in_memory import InMemoryBM25Retriever
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack_integrations.components.rankers.typesafe import TypeSafeRanker

document_store = InMemoryDocumentStore()
document_store.write_documents(
    [
        Document(content="Paris is the capital of France."),
        Document(content="France borders Spain, Italy, and Germany."),
        Document(content="Berlin is the capital of Germany."),
    ]
)

pipeline = Pipeline()
pipeline.add_component("retriever", InMemoryBM25Retriever(document_store=document_store))
pipeline.add_component("ranker", TypeSafeRanker())
pipeline.connect("retriever.documents", "ranker.documents")

query = "What is the capital of France?"
result = pipeline.run({"retriever": {"query": query, "top_k": 3}, "ranker": {"query": query, "top_k": 1}})
print(result["ranker"]["documents"][0].content)
# Paris is the capital of France.

To change how relevance is judged, pass your own instructions and criteria, the relevance levels from least to most relevant. Both can refer to the query and the passage. Use meta_fields_to_embed to show the model metadata fields, such as a title, next to the text.

Route text

TypeSafeTextRouter has one output per label. With min_confidence set, texts whose answer is less confident than the threshold go to the low_confidence output instead, so you can send them to a fallback such as an LLM or a human.

from haystack_integrations.components.routers.typesafe import TypeSafeTextRouter

router = TypeSafeTextRouter(
    labels={
        "billing": "Payments, invoices, refunds and charges",
        "technical": "Bugs, crashes and errors in the product",
    },
    instructions="Which team should handle this support request?",
    min_confidence=0.6,
)

print(router.run(text="I was charged twice for my subscription."))
# {'billing': 'I was charged twice for my subscription.'}

Run open models locally with Ollaya

Ollaya serves open decision models through the TypeSafe API. Start it and pull a model:

docker run -d --name ollaya -p 11435:11435 -v ollaya:/home/ollaya/.ollaya ghcr.io/ollaya-dev/ollaya
docker exec ollaya ollaya pull laya:en

Then point any of the components at the server with api_base_url and pick one of its models. Ollaya accepts any non-empty API key unless it is configured with one:

from haystack.utils import Secret
from haystack_integrations.components.routers.typesafe import TypeSafeTextRouter

router = TypeSafeTextRouter(
    labels=["billing", "technical"],
    model="laya:en",
    api_key=Secret.from_token("local"),
    api_base_url="http://localhost:11435",
)

print(router.run(text="The app crashes every time I open the settings page."))
# {'technical': 'The app crashes every time I open the settings page.'}

License

typesafe-haystack is distributed under the terms of the Apache-2.0 license.