b4041f6892
Remove the obsolete dashboard now that Langfuse is the analytics surface.\nIntroduce focused transport, model, and configuration modules while preserving the ai_review facade, and document the current runtime architecture.
37 lines
1.3 KiB
Python
37 lines
1.3 KiB
Python
"""Model-provider adapter for the legacy Anthropic-compatible endpoint."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
try: # Works both as `python pilot/ai_review.py` and `import pilot.model_client`.
|
|
from .gitea_client import request
|
|
except ImportError: # pragma: no cover - script-style runtime
|
|
from gitea_client import request
|
|
|
|
|
|
def parse_text_blocks(content: object) -> str:
|
|
"""Return only text blocks from an Anthropic-style response."""
|
|
if not isinstance(content, list):
|
|
return ""
|
|
return "\n".join(
|
|
block["text"]
|
|
for block in content
|
|
if isinstance(block, dict)
|
|
and block.get("type") == "text"
|
|
and isinstance(block.get("text"), str)
|
|
).strip()
|
|
|
|
|
|
def complete(base_url: str, model: str, system: str, user: str, max_tokens: int) -> str:
|
|
payload = {
|
|
"model": model,
|
|
"max_tokens": max_tokens,
|
|
"system": system,
|
|
"messages": [{"role": "user", "content": user}],
|
|
}
|
|
status, raw = request("POST", f"{base_url.rstrip('/')}/v1/messages", "ollama", payload)
|
|
if status != 200:
|
|
detail = raw[:500].decode("utf-8", errors="replace")
|
|
raise RuntimeError(f"model call failed: HTTP {status}: {detail}")
|
|
return parse_text_blocks(json.loads(raw).get("content", []))
|