72 lines
2.3 KiB
Python
72 lines
2.3 KiB
Python
"""Async HTTP client for the Knowledge Verification API.
|
|
|
|
Provides typed methods for each KV API endpoint.
|
|
Ported from the original CLI client for use in the TUI.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from types import TracebackType
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
|
|
@dataclass
|
|
class KVClient:
|
|
"""Async context-managed client for the KV API."""
|
|
|
|
base_url: str = "http://localhost:41233"
|
|
timeout: float = 60.0
|
|
_client: httpx.AsyncClient = field(init=False, repr=False)
|
|
|
|
async def __aenter__(self) -> KVClient:
|
|
self._client = httpx.AsyncClient(base_url=self.base_url, timeout=self.timeout)
|
|
return self
|
|
|
|
async def __aexit__(
|
|
self,
|
|
exc_type: type[BaseException] | None,
|
|
exc_val: BaseException | None,
|
|
exc_tb: TracebackType | None,
|
|
) -> None:
|
|
await self._client.aclose()
|
|
|
|
async def health(self) -> dict[str, Any]:
|
|
"""Check KV API health status."""
|
|
resp = await self._client.get("/api/truth/health")
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
async def search(self, query: str, limit: int = 5) -> list[dict[str, Any]]:
|
|
"""Semantic search across platform knowledge base."""
|
|
resp = await self._client.get(
|
|
"/api/truth/search", params={"q": query, "limit": str(limit)}
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json().get("results", [])
|
|
|
|
async def validate(self, content: str, source: str = "knowledge-platform") -> dict[str, Any]:
|
|
"""Validate content accuracy against platform facts."""
|
|
resp = await self._client.post(
|
|
"/api/truth/validate", json={"content": content, "source": source}
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
async def correct(self, content: str, use_reasoning: bool = False) -> dict[str, Any]:
|
|
"""Request LLM-powered content correction."""
|
|
resp = await self._client.post(
|
|
"/api/truth/correct",
|
|
json={"content": content, "useReasoning": use_reasoning},
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
async def llm_health(self) -> dict[str, Any]:
|
|
"""Check LLM service availability."""
|
|
resp = await self._client.get("/api/truth/llm/health")
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|