LLLM Tactic Processor

Goal: consume SSSN events, run each payload through an LLLM tactic, and write the tactic output back to an analysis channel.

The same flow is available as an executable example at examples/lllm_tactic_processor/workflow.py.

Prerequisites

python -m pip install -e ".[dev]"
python -m pip install lllm

Files Used

• examples/lllm_tactic_processor/workflow.py contains the runnable composition.
• tests/test_examples.py verifies the example when lllm is installed.
• sssn/stores/local.py provides durable channels and subscription cursors.

Boundary

SSSN owns the durable data plane: channels, subscriptions, cursors, event lineage, artifacts, and snapshots. LLLM owns the typed compute boundary. The processor between them should stay small.

Config Refs

When channels come from a PsiHub package, keep their local binding in .psi/config.toml and let SSSN resolve only the refs it owns:

[refs."psi://demo/events/channels/raw"]
store = ".sssn"

[refs."psi://demo/events/channels/analysis"]
store = ".sssn"

from sssn import SSSNResolver

resolver = SSSNResolver.from_config(".")
store = resolver.local_store("psi://demo/events/channels/raw")

Tactic

from pydantic import BaseModel
from lllm import Tactic


class RawMessage(BaseModel):
    text: str


class AnalysisMessage(BaseModel):
    summary: str
    length: int


class AnalyzeMessage(Tactic[RawMessage, AnalysisMessage]):
    name = "analyze_message"
    input_type = RawMessage
    output_type = AnalysisMessage

    def _run(self, input_value, *, context=None):
        return AnalysisMessage(
            summary=input_value.text.upper(),
            length=len(input_value.text),
        )

Processor

from sssn import Event, LocalStore

store = LocalStore(".sssn")
store.create_channel({"name": "raw", "schema": "demo.RawMessage"})
store.create_channel({"name": "analysis", "schema": "demo.AnalysisMessage"})
subscription = store.create_subscription(
    "raw",
    consumer="lllm_tactic_processor",
    filters={"kind": "message"},
)

raw = store.append_event(
    Event(
        channel="raw",
        source="example",
        kind="message",
        payload={"text": "hello from sssn"},
        correlation_id="episode-1",
    )
)

tactic = AnalyzeMessage()
for event in store.pull_subscription(subscription.id):
    result = tactic.run(event.payload)
    store.append_event(
        Event(
            channel="analysis",
            source="lllm_tactic_processor",
            kind="analysis",
            payload=result.model_dump(),
            correlation_id=event.correlation_id or event.id,
            parent_ids=(event.id,),
        )
    )

The analysis event keeps the raw event as a parent, so downstream services can trace an LLLM result back to the SSSN input that produced it.

Verify

python examples/lllm_tactic_processor/workflow.py

Expected output:

{
  "analysis_event_ids": ["..."],
  "raw_event_id": "...",
  "root": "...",
  "tactic": "analyze_message"
}

Next, replace the local tactic with a remote LLLM service while keeping the SSSN subscription and output-channel shape unchanged.