codenib.llm

LLM module for CodeNib.

Modules:

Name	Description
decisions	Typed decisions over OpenRouter's Decisions API, separate from chat completion.
diagnostics	Static diagnostics for LiteLLM-backed product configuration.
litellm_chat	Lightweight LLM chat wrapper around LiteLLM.
options	Validated LiteLLM request options shared by CLI and web runtimes.
usage	Token usage and cost tracking for LLM calls.

Classes:

Name	Description
ChoiceAnswer	Selected alternative, its full distribution, and model confidence.
ChoiceQuestion	Choose among named alternatives described by their criteria.
DecisionResult	Answers and provider metadata, including actual token usage and cost.
NoulAnswer	The probability of yes; workflow thresholds remain the caller's policy.
NoulQuestion	Ask for the probability that a condition holds, without choosing a threshold.
OpenRouterDecisions	Small synchronous client for Jev's state + questions -> answers API.
ScoreAnswer	Expected position on the question's scale, with its full distribution.
ScoreQuestion	Score on an ordered scale from zero to len(criteria) - 1.
ChatMessage	A single chat message with role and content.
LiteLLMChat	Thin wrapper around litellm.completion.
RetryConfig	Exponential-backoff retry policy for transient LLM errors.

ChoiceAnswer

Bases: BaseModel

Selected alternative, its full distribution, and model confidence.

ChoiceQuestion

Bases: _Question

Choose among named alternatives described by their criteria.

DecisionResult

Bases: BaseModel

Answers and provider metadata, including actual token usage and cost.

Methods:

Name	Description
validate_questions	Reject mismatched answers before they can influence application policy.

validate_questions

validate_questions(questions: Mapping[str, DecisionQuestion]) -> None

Reject mismatched answers before they can influence application policy.

Source code in codenib/llm/decisions.py

def validate_questions(self, questions: Mapping[str, DecisionQuestion]) -> None:
    """Reject mismatched answers before they can influence application policy."""
    if self.answers.keys() != questions.keys():
        raise ValueError("Decision answer IDs do not match the requested questions")
    for name, question in questions.items():
        answer = self.answers[name]
        if answer.type != question.type:
            raise ValueError(
                f"Decision answer type does not match question {name!r}"
            )
        if isinstance(answer, NoulAnswer):
            continue
        if isinstance(question, ChoiceQuestion):
            expected = set(question.criteria)
            if answer.choice not in expected:
                raise ValueError(f"Unknown choice for question {name!r}")
        else:
            expected = {str(i) for i in range(len(question.criteria))}
            if answer.score > len(question.criteria) - 1:
                raise ValueError(f"Score outside the scale for question {name!r}")
            expected_legend = {
                str(i): criterion for i, criterion in enumerate(question.criteria)
            }
            if answer.legend != expected_legend:
                raise ValueError(f"Score legend does not match question {name!r}")
        # Jev rounds each probability to two decimal places on the wire.
        # Allow half a rounding unit per entry; retain the original values
        # and score rather than normalizing or recomputing provider output.
        rounding_tolerance = 0.005 * len(expected) + 1e-9
        if set(answer.probabilities) != expected or not math.isclose(
            math.fsum(answer.probabilities.values()),
            1.0,
            rel_tol=0,
            abs_tol=rounding_tolerance,
        ):
            raise ValueError(
                f"Invalid probability distribution for question {name!r}"
            )

NoulAnswer

Bases: BaseModel

The probability of yes; workflow thresholds remain the caller's policy.

NoulQuestion

Bases: _Question

Ask for the probability that a condition holds, without choosing a threshold.

OpenRouterDecisions dataclass

OpenRouterDecisions(
    model: str = "~typesafe/jev-latest",
    api_key: str | None = None,
    timeout: float = 30.0,
    max_retries: int = 2,
    trust_env: bool = True,
)

Small synchronous client for Jev's state + questions -> answers API.

Reads OPENROUTER_API_KEY at invocation time unless api_key is supplied. Only transport failures, HTTP 408/429, and HTTP 5xx are retried. This client has no chat, tool-calling, or text-generation interface.

Methods:

Name	Description
decide	Evaluate a batch of typed questions against the same JSON-compatible state.

decide

decide(
    *, state: Any, questions: Mapping[str, DecisionQuestion | dict[str, Any]]
) -> DecisionResult

Evaluate a batch of typed questions against the same JSON-compatible state.

Source code in codenib/llm/decisions.py

def decide(
    self,
    *,
    state: Any,
    questions: Mapping[str, DecisionQuestion | dict[str, Any]],
) -> DecisionResult:
    """Evaluate a batch of typed questions against the same JSON-compatible state."""
    parsed = _QUESTIONS.validate_python(dict(questions))
    if not parsed:
        raise ValueError("At least one decision question is required")
    payload = {
        "model": self.model,
        "state": state,
        "questions": {
            name: question.model_dump(exclude_none=True)
            for name, question in parsed.items()
        },
    }
    # Fail locally on non-JSON state (including NaN), before sending any data.
    json.dumps(payload, allow_nan=False)
    key = (
        self.api_key
        if self.api_key is not None
        else os.getenv("OPENROUTER_API_KEY")
    )
    if not key or not key.strip():
        raise ValueError("Set OPENROUTER_API_KEY to use OpenRouter Decisions")

    import requests

    for attempt in range(self.max_retries + 1):
        try:
            with requests.Session() as session:
                session.trust_env = self.trust_env
                response = session.post(
                    "https://openrouter.ai/api/alpha/decisions",
                    headers={
                        "Authorization": f"Bearer {key.strip()}",
                        "X-OpenRouter-Title": "CodeNib",
                    },
                    json=payload,
                    timeout=self.timeout,
                    allow_redirects=False,
                )
            response.raise_for_status()
            break
        except requests.HTTPError as exc:
            status = exc.response.status_code if exc.response is not None else 0
            if attempt == self.max_retries or not (
                status in (408, 429) or 500 <= status < 600
            ):
                raise
        except (requests.ConnectionError, requests.Timeout):
            if attempt == self.max_retries:
                raise
        time.sleep(min(0.5 * 2**attempt, 8.0))

    result = DecisionResult.model_validate(response.json())
    result.validate_questions(parsed)
    return result

ScoreAnswer

Bases: BaseModel

Expected position on the question's scale, with its full distribution.

ScoreQuestion

Bases: _Question

Score on an ordered scale from zero to len(criteria) - 1.

ChatMessage dataclass

ChatMessage(role: str, content: str)

A single chat message with role and content.

LiteLLMChat dataclass

LiteLLMChat(
    model: str,
    temperature: float | None = 0.0,
    max_tokens: int | None = 8192,
    api_key: str | None = None,
    api_base: str | None = None,
    extra_kwargs: dict[str, Any] = dict(),
    retry: RetryConfig = RetryConfig(),
)

Thin wrapper around litellm.completion.

Provides .invoke() and .with_structured_output() matching the interface that callers already use via LangChain chat models.

Methods:

Name	Description
invoke	Send messages and return the assistant content string.
complete	Complete raw LiteLLM message dictionaries and return text content.
with_structured_output	Return a callable that parses LLM responses into schema.

Attributes:

Name	Type	Description
cache_identity	str	Stable non-secret identity for model-dependent artifact caches.

cache_identity property

cache_identity: str

Stable non-secret identity for model-dependent artifact caches.

invoke

invoke(messages: list[ChatMessage]) -> str

Send messages and return the assistant content string.

Source code in codenib/llm/litellm_chat.py

def invoke(self, messages: List[ChatMessage]) -> str:
    """Send messages and return the assistant content string."""
    response = self._call(messages)
    return response.choices[0].message.content

complete

complete(messages: list[dict[str, Any]], **overrides: Any) -> str

Complete raw LiteLLM message dictionaries and return text content.

Source code in codenib/llm/litellm_chat.py

def complete(
    self,
    messages: List[Dict[str, Any]],
    **overrides: Any,
) -> str:
    """Complete raw LiteLLM message dictionaries and return text content."""

    response = self._call_raw(messages, **overrides)
    return (response.choices[0].message.content or "").strip()

with_structured_output

with_structured_output(schema: Type[BaseModel]) -> _StructuredLLM

Return a callable that parses LLM responses into schema.

Source code in codenib/llm/litellm_chat.py

def with_structured_output(self, schema: Type[BaseModel]) -> _StructuredLLM:
    """Return a callable that parses LLM responses into *schema*."""
    return _StructuredLLM(chat=self, schema=schema)

RetryConfig dataclass

RetryConfig(
    max_retries: int = 2,
    base_delay: float = 0.5,
    max_delay: float = 8.0,
    jitter: float = 0.1,
)

Exponential-backoff retry policy for transient LLM errors.

max_retries is the number of additional attempts after the first, so the call is made at most max_retries + 1 times. base_delay is the first backoff in seconds; each subsequent wait is base_delay * 2**attempt capped at max_delay. A small random jitter spreads retries out so concurrent callers don't synchronize.

Methods:

Name	Description
backoff	Seconds to sleep before retry attempt (0-based).

backoff

backoff(attempt: int) -> float

Seconds to sleep before retry attempt (0-based).

Source code in codenib/llm/litellm_chat.py

def backoff(self, attempt: int) -> float:
    """Seconds to sleep before retry ``attempt`` (0-based)."""
    delay = min(self.base_delay * (2**attempt), self.max_delay)
    if self.jitter:
        delay += random.uniform(0, self.jitter * delay)
    return delay