17759ef564
- apps/api: psycopg v3 repositories, schema.sql + migration runner, state machine implementing the data-model transition table, approval gate with sha256 hash match + 24h expiry enforced at send time, EchoTransport pluggable sender, LLM calls routed through packages.llm-gateway with mock mode, 47 pytest tests green - apps/api/Dockerfile.test: python 3.13-slim test image (DinD-safe: migrations copied as directory) - docker-compose.yml: add api-test service on compose network (host port publishing is broken in this sandbox; container-to-container networking used) - Fix: replace masked placeholder password in config.py/conftest.py defaults
101 lines
No EOL
3 KiB
Python
101 lines
No EOL
3 KiB
Python
"""LLM gateway integration.
|
|
|
|
Per the task spec: LLM usage via packages.llm_gateway. Since T2 hasn't built
|
|
the gateway yet, this module provides a minimal mock-mode interface that the
|
|
API endpoints call. When packages.llm_gateway is available (importable), it
|
|
delegates to the real gateway. Otherwise, it returns deterministic canned
|
|
outputs per task name.
|
|
|
|
Task names used by the API:
|
|
- cv_assist: returns suggestions for a CV section bullet
|
|
- cl_critique: returns critique comments for a cover letter
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import time
|
|
from typing import Any
|
|
|
|
# Try to import the real llm_gateway package
|
|
try:
|
|
from packages.llm_gateway.client import run_task as _real_run_task # type: ignore
|
|
HAS_REAL_GATEWAY = True
|
|
except Exception:
|
|
HAS_REAL_GATEWAY = False
|
|
|
|
|
|
def _has_api_key() -> bool:
|
|
return bool(
|
|
os.environ.get("LLM_PRIMARY_KEY")
|
|
or os.environ.get("OLLAMA_API_KEY")
|
|
)
|
|
|
|
|
|
# Deterministic mock outputs per task name
|
|
MOCK_OUTPUTS: dict[str, dict[str, Any]] = {
|
|
"cv_assist": {
|
|
"suggestions": [
|
|
"Improved bullet: Led cross-functional team of 8 to deliver feature X 2 weeks ahead of schedule.",
|
|
"Alternative: Streamlined process reducing cycle time by 30% via automation.",
|
|
]
|
|
},
|
|
"cl_critique": {
|
|
"comments": [
|
|
{
|
|
"quote": "I am writing to apply",
|
|
"suggestion": "Consider a stronger opening that references the specific role.",
|
|
"severity": "medium",
|
|
},
|
|
{
|
|
"quote": "I have experience",
|
|
"suggestion": "Quantify with a specific achievement rather than a generic claim.",
|
|
"severity": "low",
|
|
},
|
|
]
|
|
},
|
|
"score": {
|
|
"score": 72,
|
|
"rationale": {
|
|
"match": 0.72,
|
|
"factors": {"skills": 0.8, "location": 0.6, "experience": 0.75},
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
def run_task(
|
|
task: str,
|
|
prompt: str,
|
|
schema: dict[str, Any] | None = None,
|
|
telemetry_sink: callable | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Run an LLM task. Uses mock mode when no API key is present.
|
|
|
|
Returns a dict with the task result. If a telemetry_sink callable is
|
|
provided, it is called with a dict of token/cost info.
|
|
"""
|
|
if HAS_REAL_GATEWAY and _has_api_key():
|
|
return _real_run_task(task, prompt, schema)
|
|
|
|
# Mock mode
|
|
time.sleep(0.01) # simulate latency
|
|
result = MOCK_OUTPUTS.get(task, {"result": "mock"})
|
|
|
|
# Validate against schema if provided (basic check)
|
|
# In real gateway this would be jsonschema validation
|
|
|
|
if telemetry_sink is not None:
|
|
telemetry_sink({
|
|
"task": task,
|
|
"model": "mock",
|
|
"provider": "mock",
|
|
"input_tokens": len(prompt) // 4, # rough estimate
|
|
"output_tokens": len(json.dumps(result)) // 4,
|
|
"cost_usd": 0.0,
|
|
"duration_ms": 10,
|
|
})
|
|
|
|
return result |