Chris Olson
39a338990f
Closed-loop account-intelligence system: harvest -> relevance filter vs. Capability Library -> synthesize -> score -> deliver -> feedback. - 40-capability Broadcom + VMware library - Live Exa.ai news connector with hard free-plan request ceilings - Provider-agnostic LLM layer (Gemini/Anthropic) with offline mock fallback - LLM confidence wired into composite scoring + low-confidence floor - 7-account pilot list Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
106 lines
3.5 KiB
Python
106 lines
3.5 KiB
Python
"""LLM access layer (provider-agnostic).
|
|
|
|
Dispatches to a configured provider — Gemini or Anthropic — via plain urllib
|
|
(no SDKs). When no key is configured it falls back to a deterministic mock
|
|
reasoner so the full loop runs offline. All paths return parsed JSON dicts.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import urllib.request
|
|
import urllib.error
|
|
|
|
from . import config
|
|
|
|
|
|
class LLMError(RuntimeError):
|
|
pass
|
|
|
|
|
|
def _extract_json(text: str) -> dict:
|
|
"""Pull the first JSON object out of a model response."""
|
|
text = text.strip()
|
|
# Strip ```json fences if present.
|
|
fence = re.search(r"```(?:json)?\s*(\{.*?\})\s*```", text, re.DOTALL)
|
|
if fence:
|
|
text = fence.group(1)
|
|
start = text.find("{")
|
|
end = text.rfind("}")
|
|
if start == -1 or end == -1:
|
|
raise LLMError(f"No JSON object found in LLM output: {text[:200]}")
|
|
return json.loads(text[start : end + 1])
|
|
|
|
|
|
def _post(url: str, payload: dict, headers: dict, provider: str) -> dict:
|
|
req = urllib.request.Request(
|
|
url,
|
|
data=json.dumps(payload).encode("utf-8"),
|
|
headers={"content-type": "application/json", **headers},
|
|
method="POST",
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=60) as resp:
|
|
return json.loads(resp.read().decode("utf-8"))
|
|
except urllib.error.HTTPError as e:
|
|
detail = e.read().decode("utf-8")[:400] if hasattr(e, "read") else ""
|
|
raise LLMError(f"{provider} API error {e.code}: {detail}")
|
|
except urllib.error.URLError as e:
|
|
raise LLMError(f"Network error calling {provider} API: {e}")
|
|
|
|
|
|
def _call_gemini(system: str, user: str, max_tokens: int) -> dict:
|
|
url = (
|
|
"https://generativelanguage.googleapis.com/v1beta/models/"
|
|
f"{config.GEMINI_MODEL}:generateContent"
|
|
)
|
|
payload = {
|
|
"system_instruction": {"parts": [{"text": system}]},
|
|
"contents": [{"role": "user", "parts": [{"text": user}]}],
|
|
"generationConfig": {
|
|
"maxOutputTokens": max_tokens,
|
|
"temperature": 0.2,
|
|
"responseMimeType": "application/json",
|
|
},
|
|
}
|
|
body = _post(url, payload, {"x-goog-api-key": config.GEMINI_API_KEY}, "Gemini")
|
|
candidates = body.get("candidates", [])
|
|
if not candidates:
|
|
raise LLMError(f"Gemini returned no candidates: {json.dumps(body)[:300]}")
|
|
parts = candidates[0].get("content", {}).get("parts", [])
|
|
text = "".join(p.get("text", "") for p in parts)
|
|
return _extract_json(text)
|
|
|
|
|
|
def _call_anthropic(system: str, user: str, max_tokens: int) -> dict:
|
|
payload = {
|
|
"model": config.LLM_MODEL,
|
|
"max_tokens": max_tokens,
|
|
"system": system,
|
|
"messages": [{"role": "user", "content": user}],
|
|
}
|
|
headers = {
|
|
"x-api-key": config.ANTHROPIC_API_KEY,
|
|
"anthropic-version": "2023-06-01",
|
|
}
|
|
body = _post("https://api.anthropic.com/v1/messages", payload, headers, "Anthropic")
|
|
text = "".join(b.get("text", "") for b in body.get("content", []))
|
|
return _extract_json(text)
|
|
|
|
|
|
def call_json(system: str, user: str, max_tokens: int = 1500) -> dict:
|
|
"""Single-turn LLM call that returns parsed JSON."""
|
|
if config.USE_MOCK_LLM:
|
|
raise _MockSignal() # callers catch this and run their mock path
|
|
if config.LLM_PROVIDER == "gemini":
|
|
return _call_gemini(system, user, max_tokens)
|
|
return _call_anthropic(system, user, max_tokens)
|
|
|
|
|
|
class _MockSignal(Exception):
|
|
"""Internal sentinel: tells the caller to use its deterministic mock branch."""
|
|
|
|
|
|
def is_mock() -> bool:
|
|
return config.USE_MOCK_LLM
|