From 8975ebd8ba5240c88b469dc5ee74fee18fc10d92 Mon Sep 17 00:00:00 2001 From: Cyber MacGeddon Date: Fri, 16 Oct 2026 15:15:06 +0100 Subject: [PATCH] feat: add decision-scoring service with Laya backend Add a new decision-scoring service that wraps the Laya model for fast, structured decision-making. Supports three output types per request: choice (select from options), score (probability distribution over ordered levels), and noul (binary yes/no confidence). Multiple outputs can be evaluated in a single call against the same input. - DecisionScoringRequest/Response schema, translators, and queue definitions - Gateway DecisionScoringRequestor and dispatcher registration - Laya-backed service at trustgraph-flow/trustgraph/model/decision_scoring/laya/ - Auto-routes by input language; optional model override via flow parameter - Added laya dependency and decision-scoring-laya entry point to trustgraph-flow --- docs/tech-specs/decision-scoring-service.md | 296 ++++++++++++++++++ .../trustgraph/messaging/__init__.py | 7 + .../messaging/translators/__init__.py | 1 + .../messaging/translators/decision_scoring.py | 45 +++ trustgraph-base/trustgraph/schema/__init__.py | 1 + .../trustgraph/schema/decision_scoring.py | 27 ++ trustgraph-flow/pyproject.toml | 2 + .../gateway/dispatch/decision_scoring.py | 31 ++ .../trustgraph/gateway/dispatch/manager.py | 2 + .../model/decision_scoring/__init__.py | 1 + .../model/decision_scoring/laya/__init__.py | 2 + .../model/decision_scoring/laya/__main__.py | 6 + .../model/decision_scoring/laya/service.py | 111 +++++++ 13 files changed, 532 insertions(+) create mode 100644 docs/tech-specs/decision-scoring-service.md create mode 100644 trustgraph-base/trustgraph/messaging/translators/decision_scoring.py create mode 100644 trustgraph-base/trustgraph/schema/decision_scoring.py create mode 100644 trustgraph-flow/trustgraph/gateway/dispatch/decision_scoring.py create mode 100644 trustgraph-flow/trustgraph/model/decision_scoring/__init__.py create mode 100644 trustgraph-flow/trustgraph/model/decision_scoring/laya/__init__.py create mode 100644 trustgraph-flow/trustgraph/model/decision_scoring/laya/__main__.py create mode 100644 trustgraph-flow/trustgraph/model/decision_scoring/laya/service.py diff --git a/docs/tech-specs/decision-scoring-service.md b/docs/tech-specs/decision-scoring-service.md new file mode 100644 index 000000000..2af8923c5 --- /dev/null +++ b/docs/tech-specs/decision-scoring-service.md @@ -0,0 +1,296 @@ +--- +layout: default +title: "Decision Scoring Service Technical Specification" +parent: "Tech Specs" +--- + +# Decision Scoring Service Technical Specification + +## Problem + +TrustGraph currently relies on LLM text-completion services for all +decision-making, including tasks that require only a simple judgment: +choosing between options, scoring relevance, or making a yes/no call. +These are not generative tasks — they need discrimination, not prose. + +Using an LLM for these decisions has three costs: + +1. **Latency**. A full text-completion round-trip for a decision that + could be expressed as a single number or a selection from a short + list. + +2. **Token cost**. Input and output tokens are consumed for every + decision, regardless of complexity. High-frequency decisions — + per-query routing, per-chunk scoring, per-candidate filtering — + accumulate cost with no corresponding value from the LLM's + generative capability. + +3. **Brittleness**. Extracting a structured decision from free-text + LLM output requires parsing, validation, and fallback handling. + The failure modes are disproportionate to the simplicity of the + task. + +A new class of models has emerged that targets exactly this gap. +Models such as Laya (open) and Jev (commercial) are purpose-built for +fast, structured decision-making — not generation. They accept an +input and a specification of the desired output, and produce one of +three result types: + +- **Choice** — select one option from a provided set +- **Score** — given a set of ordered levels (e.g. 0=unrelated, + 1=adjacent, 2=some direct, 3=deep direct), the model produces a + probability distribution across the levels. The output is the + expected value (weighted sum of level × probability) and a + confidence (the highest individual probability). +- **Noul** — a binary confidence value from 0.0 (no) to 1.0 (yes) + +A single request can specify multiple outputs, each independently +configured as choice, score, or noul. For example, a single call might +ask for a topic choice, a relevance score, and a yes/no safety noul — +all evaluated together against the same input. + +These models are designed for System 1 thinking: fast, automatic +pattern-matching rather than slow deliberative reasoning. Multiple +models now support this approach, and more are expected. TrustGraph +needs a service layer that can integrate them. + +## Solution + +A new TrustGraph service — `decision-scoring-laya` — that wraps the +Laya open model behind the standard request/response pattern. The +service name follows the existing convention (e.g. +`text-completion-openai`): `decision-scoring` is the abstract service +kind, `-laya` identifies the backend. Additional backends (e.g. +`decision-scoring-jev`) can follow the same pattern. + +### Output Specification + +Each request includes the input text and an output specification — a +map of named outputs, each independently configured. The specification +is caller-defined: criteria are arbitrary user-provided strings, not a +fixed vocabulary. + +```json +{ + "department": { + "type": "choice", + "instructions": "Which team should handle this ticket?", + "criteria": [ + "billing", + "technical", + "sales" + ] + }, + "urgency": { + "type": "score", + "instructions": "How urgent is this request?", + "criteria": [ + "none", + "low", + "medium", + "high", + "critical" + ] + }, + "refund_requested": { + "type": "noul", + "instructions": "Does the customer explicitly request a refund?" + } +} +``` + +- **Choice** outputs require `criteria` — the set of options to select + from. The result is the selected option and a confidence. +- **Score** outputs require `criteria` — an ordered list of levels. The + model produces a probability distribution across the levels; the + result is the expected value (weighted sum) and a confidence. Level + ordering is positional (index 0 is lowest). +- **Noul** outputs take no criteria. The result is a single value from + 0.0 (no) to 1.0 (yes). + +Each output has `instructions` — natural language guidance telling the +model what to evaluate. + +The decision scoring service does not replace existing LLM-based +decision-making. It is another tool in the armoury — a lightweight, +fast option that sits alongside text-completion services. Some decisions +will always need the full generative capability of an LLM; others are +better served by a purpose-built decision model. The choice of which to +use for a given task is a design decision, not a migration. + +## Technical Design + +### Laya Integration + +The service loads the Laya `Router` in-process with preloaded +checkpoints. Inference is ~35ms per call — no external API round-trip. + +```python +from laya import Router +router = Router(preload=True) +``` + +Laya auto-routes to the appropriate model variant based on input +language: ModernBERT-large for English, mmBERT-base for multilingual. +The service can also accept an explicit model override via flow +parameter. + +### Request/Response Data Shapes + +#### Request + +```python +@dataclass +class DecisionScoringRequest: + state: dict = field(default_factory=dict) + questions: dict = field(default_factory=dict) +``` + +- **state** — freeform input data. Can be any structure: a simple + string, a dict with fields like `from`, `subject`, `body`, or any + application-specific shape. This is what the model evaluates. +- **questions** — the output specification map, as described in the + Solution section. Each key is an output name, each value specifies + type, instructions, and criteria. + +Note: for **choice** outputs, criteria can be either a list of options +or a dict mapping options to descriptions that guide the model: + +```json +{ + "criteria": { + "billing": "invoices, payments, refunds", + "technical": "bugs, outages, system errors", + "sales": "pricing, new contracts" + } +} +``` + +#### Response + +```python +@dataclass +class DecisionScoringResponse: + error: Error | None = None + answers: dict = field(default_factory=dict) + model: str = "" +``` + +- **answers** — a map keyed by the same output names as the request's + questions. Each entry contains type-specific results: + + For **choice**: + ```json + { + "department": { + "type": "choice", + "choice": "billing", + "confidence": 0.94 + } + } + ``` + + For **score**: + ```json + { + "urgency": { + "type": "score", + "score": 1.85, + "confidence": 0.72, + "distribution": [0.05, 0.23, 0.72] + } + } + ``` + + For **noul**: + ```json + { + "refund_requested": { + "type": "noul", + "value": 0.92 + } + } + ``` + +- **model** — which Laya model variant handled the request (e.g. + `english`, `multilingual`). + +#### Queues + +```python +decision_scoring_request_queue = queue('decision-scoring', cls='request') +decision_scoring_response_queue = queue('decision-scoring', cls='response') +``` + +#### Gateway Integration + +A `DecisionScoringRequestor` is added to the gateway's dispatcher +mapping: + +```python +request_response_dispatchers = { + ... + "decision-scoring": DecisionScoringRequestor, +} +``` + +Translators are registered in the translator registry: + +```python +TranslatorRegistry.register_service( + "decision-scoring", + DecisionScoringRequestTranslator(), + DecisionScoringResponseTranslator(), +) +``` + +### Service Shape + +The service extends `FlowProcessor`. Unlike the cloze model service, +it does not train — Laya's models are pre-trained checkpoints loaded +at startup. The service is stateless beyond the loaded model weights. + +```python +class DecisionScoringService(FlowProcessor): + def __init__(self, **params): + super().__init__(**params) + + self.register_specification( + ParameterSpec(name="model") + ) + + self.register_specification( + ConsumerSpec( + name="request", + schema=DecisionScoringRequest, + handler=self.on_request, + ) + ) + + self.register_specification( + ProducerSpec( + name="response", + schema=DecisionScoringResponse, + ) + ) +``` + +The `model` parameter is optional. If set, it overrides Laya's +auto-routing and forces a specific checkpoint (e.g. +`typed-decisions`). If omitted, Laya selects the appropriate model +based on the input language. + +### Concurrency + +Laya's `Router` exposes a threads setting, suggesting it supports +concurrent inference. The service should configure `ConsumerSpec` +concurrency and the Router's thread count to match, allowing multiple +requests to be processed in parallel. + +### CLI Tooling + +A command-line tool (e.g. `tg-decision-score`) for interactive testing +of the decision scoring service. Takes state and a questions +specification, submits to the service, and prints the results. Useful +for validating question specifications and debugging scoring behaviour +during development. diff --git a/trustgraph-base/trustgraph/messaging/__init__.py b/trustgraph-base/trustgraph/messaging/__init__.py index ff379d3d2..574df1e90 100644 --- a/trustgraph-base/trustgraph/messaging/__init__.py +++ b/trustgraph-base/trustgraph/messaging/__init__.py @@ -30,6 +30,7 @@ from .translators.diagnosis import StructuredDataDiagnosisRequestTranslator, StructuredDataDiagnosisResponseTranslator from .translators.reranker import RerankerRequestTranslator, RerankerResponseTranslator from .translators.cloze import ClozeRequestTranslator, ClozeResponseTranslator +from .translators.decision_scoring import DecisionScoringRequestTranslator, DecisionScoringResponseTranslator from .translators.collection import CollectionManagementRequestTranslator, CollectionManagementResponseTranslator from .translators.sparql_query import SparqlQueryRequestTranslator, SparqlQueryResponseTranslator @@ -184,6 +185,12 @@ ClozeResponseTranslator() ) +TranslatorRegistry.register_service( + "decision-scoring", + DecisionScoringRequestTranslator(), + DecisionScoringResponseTranslator() +) + # Register single-direction translators for document loading TranslatorRegistry.register_request("document", DocumentTranslator()) TranslatorRegistry.register_request("text-document", TextDocumentTranslator()) diff --git a/trustgraph-base/trustgraph/messaging/translators/__init__.py b/trustgraph-base/trustgraph/messaging/translators/__init__.py index cfd6ea188..58dabb82c 100644 --- a/trustgraph-base/trustgraph/messaging/translators/__init__.py +++ b/trustgraph-base/trustgraph/messaging/translators/__init__.py @@ -22,3 +22,4 @@ from .diagnosis import StructuredDataDiagnosisRequestTranslator, StructuredDataDiagnosisResponseTranslator from .reranker import RerankerRequestTranslator, RerankerResponseTranslator from .cloze import ClozeRequestTranslator, ClozeResponseTranslator +from .decision_scoring import DecisionScoringRequestTranslator, DecisionScoringResponseTranslator diff --git a/trustgraph-base/trustgraph/messaging/translators/decision_scoring.py b/trustgraph-base/trustgraph/messaging/translators/decision_scoring.py new file mode 100644 index 000000000..ae2f301a0 --- /dev/null +++ b/trustgraph-base/trustgraph/messaging/translators/decision_scoring.py @@ -0,0 +1,45 @@ +from typing import Dict, Any, Tuple +from ...schema import DecisionScoringRequest, DecisionScoringResponse +from .base import MessageTranslator + + +class DecisionScoringRequestTranslator(MessageTranslator): + + def decode(self, data: Dict[str, Any]) -> DecisionScoringRequest: + return DecisionScoringRequest( + state=data.get("state", {}), + questions=data.get("questions", {}), + ) + + def encode(self, obj: DecisionScoringRequest) -> Dict[str, Any]: + return { + "state": obj.state, + "questions": obj.questions, + } + + +class DecisionScoringResponseTranslator(MessageTranslator): + + def decode(self, data: Dict[str, Any]) -> DecisionScoringResponse: + raise NotImplementedError( + "Response translation to internal not typically needed" + ) + + def encode(self, obj: DecisionScoringResponse) -> Dict[str, Any]: + result = { + "answers": obj.answers, + "model": obj.model, + } + + if obj.error: + result["error"] = { + "type": obj.error.type, + "message": obj.error.message, + } + + return result + + def encode_with_completion( + self, obj: DecisionScoringResponse + ) -> Tuple[Dict[str, Any], bool]: + return self.encode(obj), True diff --git a/trustgraph-base/trustgraph/schema/__init__.py b/trustgraph-base/trustgraph/schema/__init__.py index 357e5ea06..283992301 100644 --- a/trustgraph-base/trustgraph/schema/__init__.py +++ b/trustgraph-base/trustgraph/schema/__init__.py @@ -16,3 +16,4 @@ from .policy_event import * from .user_context import * from .cloze import * +from .decision_scoring import * diff --git a/trustgraph-base/trustgraph/schema/decision_scoring.py b/trustgraph-base/trustgraph/schema/decision_scoring.py new file mode 100644 index 000000000..7b9a8a297 --- /dev/null +++ b/trustgraph-base/trustgraph/schema/decision_scoring.py @@ -0,0 +1,27 @@ + +from dataclasses import dataclass, field + +from .core.primitives import Error +from .core.topic import queue + +############################################################################ + +# Decision scoring service + +@dataclass +class DecisionScoringRequest: + state: dict = field(default_factory=dict) + questions: dict = field(default_factory=dict) + +@dataclass +class DecisionScoringResponse: + error: Error | None = None + answers: dict = field(default_factory=dict) + model: str = "" + +############################################################################ + +# Queue definitions + +decision_scoring_request_queue = queue('decision-scoring', cls='request') +decision_scoring_response_queue = queue('decision-scoring', cls='response') diff --git a/trustgraph-flow/pyproject.toml b/trustgraph-flow/pyproject.toml index 32c143113..c3e02664a 100644 --- a/trustgraph-flow/pyproject.toml +++ b/trustgraph-flow/pyproject.toml @@ -42,6 +42,7 @@ dependencies = [ "tabulate", "tiktoken", "urllib3", + "laya", ] classifiers = [ "Programming Language :: Python :: 3", @@ -128,6 +129,7 @@ triples-write-falkordb = "trustgraph.storage.triples.falkordb:run" triples-write-memgraph = "trustgraph.storage.triples.memgraph:run" triples-write-neo4j = "trustgraph.storage.triples.neo4j:run" wikipedia-lookup = "trustgraph.external.wikipedia:run" +decision-scoring-laya = "trustgraph.model.decision_scoring.laya:run" joke-service = "trustgraph.tool_service.joke:run" [tool.setuptools.packages.find] diff --git a/trustgraph-flow/trustgraph/gateway/dispatch/decision_scoring.py b/trustgraph-flow/trustgraph/gateway/dispatch/decision_scoring.py new file mode 100644 index 000000000..c944265ba --- /dev/null +++ b/trustgraph-flow/trustgraph/gateway/dispatch/decision_scoring.py @@ -0,0 +1,31 @@ + +from ... schema import DecisionScoringRequest, DecisionScoringResponse +from ... messaging import TranslatorRegistry + +from . requestor import ServiceRequestor + +class DecisionScoringRequestor(ServiceRequestor): + def __init__( + self, backend, request_queue, response_queue, timeout, + consumer, subscriber, + ): + + super(DecisionScoringRequestor, self).__init__( + backend=backend, + request_queue=request_queue, + response_queue=response_queue, + request_schema=DecisionScoringRequest, + response_schema=DecisionScoringResponse, + subscription = subscriber, + consumer_name = consumer, + timeout=timeout, + ) + + self.request_translator = TranslatorRegistry.get_request_translator("decision-scoring") + self.response_translator = TranslatorRegistry.get_response_translator("decision-scoring") + + def to_request(self, body): + return self.request_translator.decode(body) + + def from_response(self, message): + return self.response_translator.encode_with_completion(message) diff --git a/trustgraph-flow/trustgraph/gateway/dispatch/manager.py b/trustgraph-flow/trustgraph/gateway/dispatch/manager.py index 0dfa4a9a2..3088758cf 100644 --- a/trustgraph-flow/trustgraph/gateway/dispatch/manager.py +++ b/trustgraph-flow/trustgraph/gateway/dispatch/manager.py @@ -40,6 +40,7 @@ from . mcp_tool import McpToolRequestor from . reranker import RerankerRequestor from . cloze import ClozeRequestor +from . decision_scoring import DecisionScoringRequestor from . passthrough import PassthroughRequestor from . text_load import TextLoad from . document_load import DocumentLoad @@ -81,6 +82,7 @@ "reranker": RerankerRequestor, "image-to-text": ImageToTextRequestor, "cloze": ClozeRequestor, + "decision-scoring": DecisionScoringRequestor, } system_dispatchers = { diff --git a/trustgraph-flow/trustgraph/model/decision_scoring/__init__.py b/trustgraph-flow/trustgraph/model/decision_scoring/__init__.py new file mode 100644 index 000000000..8b1378917 --- /dev/null +++ b/trustgraph-flow/trustgraph/model/decision_scoring/__init__.py @@ -0,0 +1 @@ + diff --git a/trustgraph-flow/trustgraph/model/decision_scoring/laya/__init__.py b/trustgraph-flow/trustgraph/model/decision_scoring/laya/__init__.py new file mode 100644 index 000000000..214f72728 --- /dev/null +++ b/trustgraph-flow/trustgraph/model/decision_scoring/laya/__init__.py @@ -0,0 +1,2 @@ + +from . service import * diff --git a/trustgraph-flow/trustgraph/model/decision_scoring/laya/__main__.py b/trustgraph-flow/trustgraph/model/decision_scoring/laya/__main__.py new file mode 100644 index 000000000..da5a90219 --- /dev/null +++ b/trustgraph-flow/trustgraph/model/decision_scoring/laya/__main__.py @@ -0,0 +1,6 @@ +#!/usr/bin/env python3 + +from . service import run + +if __name__ == '__main__': + run() diff --git a/trustgraph-flow/trustgraph/model/decision_scoring/laya/service.py b/trustgraph-flow/trustgraph/model/decision_scoring/laya/service.py new file mode 100644 index 000000000..081071464 --- /dev/null +++ b/trustgraph-flow/trustgraph/model/decision_scoring/laya/service.py @@ -0,0 +1,111 @@ + +""" +Decision scoring service using Laya. Provides fast, structured +decision-making with three output types: choice, score, and noul. +Input is state + questions specification, output is answers with +confidence scores. +""" + +import logging + +from .... base import FlowProcessor, ConsumerSpec, ProducerSpec +from .... base import ParameterSpec +from .... schema import DecisionScoringRequest, DecisionScoringResponse +from .... schema import Error + +logger = logging.getLogger(__name__) + +default_ident = "decision-scoring" + + +class Processor(FlowProcessor): + + def __init__(self, **params): + + model = params.get("model") + + init_params = params + if model: + init_params = params | {"model": model} + + super(Processor, self).__init__(**init_params) + + self.model_override = model + + self.register_specification( + ParameterSpec(name="model") + ) + + self.register_specification( + ConsumerSpec( + name="request", + schema=DecisionScoringRequest, + handler=self.on_request, + ) + ) + + self.register_specification( + ProducerSpec( + name="response", + schema=DecisionScoringResponse, + ) + ) + + from laya import Router + logger.info("Loading Laya router with preloaded checkpoints...") + self.router = Router(preload=True) + logger.info("Laya router loaded successfully") + + async def on_request(self, msg, consumer, flow): + + request = msg.value() + id = msg.properties()["id"] + producer = flow("response") + + model = flow("model") or self.model_override + + try: + + kwargs = {} + if model: + kwargs["model"] = model + + result = self.router.predict( + request.state, request.questions, **kwargs + ) + + answers = result.get("answers", {}) + routing = result.get("routing", {}) + model_used = routing.get("model", "") + + response = DecisionScoringResponse( + answers=answers, + model=model_used, + ) + + except Exception as e: + logger.error(f"Decision scoring error: {e}") + response = DecisionScoringResponse( + error=Error( + type="decision-scoring-error", + message=str(e), + ), + ) + + await producer.send(response, send_properties={"id": id}) + + @staticmethod + def add_args(parser): + + FlowProcessor.add_args(parser) + + parser.add_argument( + '-m', '--model', + default=None, + help='Laya model override (default: auto-route by language)' + ) + + +def run(): + + Processor.launch(default_ident, __doc__)