130 lines
4.4 KiB
Python
130 lines
4.4 KiB
Python
"""VerifyFactTool -- validate claims against the platform knowledge base.
|
|
|
|
Wraps the KV API ``/api/truth/validate`` endpoint, returning a
|
|
confidence score, validity flag, and supporting documents.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from typing import Any, ClassVar
|
|
|
|
import httpx
|
|
|
|
from ..base import Tool, ToolParameter, ToolResult
|
|
|
|
# Read from KV_API_URL environment variable, fallback to localhost
|
|
_DEFAULT_KV_API_URL = os.environ.get("KV_API_URL", "http://localhost:41233")
|
|
_REQUEST_TIMEOUT = 30.0
|
|
|
|
|
|
class VerifyFactTool(Tool):
|
|
"""Verify a factual claim against the Crystal knowledge base.
|
|
|
|
Submits a statement to the KV API validation endpoint and returns
|
|
whether the claim is valid, a confidence score, and any supporting
|
|
or contradicting documents.
|
|
"""
|
|
|
|
name: ClassVar[str] = "verify_fact"
|
|
description: ClassVar[str] = (
|
|
"Verify a factual statement against the Lilith platform knowledge base. "
|
|
"Returns a validity flag, confidence score (0-1), and supporting documents. "
|
|
"Use this to check claims about platform economics, terminology, or features."
|
|
)
|
|
parameters: ClassVar[list[ToolParameter]] = [
|
|
ToolParameter(
|
|
name="statement",
|
|
type="string",
|
|
description="The factual claim to verify",
|
|
),
|
|
ToolParameter(
|
|
name="kv_api_url",
|
|
type="string",
|
|
description="KV API base URL",
|
|
required=False,
|
|
default=_DEFAULT_KV_API_URL,
|
|
),
|
|
]
|
|
|
|
async def execute(self, **kwargs: Any) -> ToolResult:
|
|
statement: str = kwargs["statement"]
|
|
kv_api_url: str = kwargs.get("kv_api_url", _DEFAULT_KV_API_URL)
|
|
|
|
async with httpx.AsyncClient(
|
|
base_url=kv_api_url, timeout=_REQUEST_TIMEOUT
|
|
) as client:
|
|
try:
|
|
resp = await client.post(
|
|
"/api/truth/validate",
|
|
json={"content": statement, "source": "tool"},
|
|
)
|
|
except httpx.ConnectError:
|
|
return ToolResult.fail(
|
|
f"Cannot connect to KV API at {kv_api_url}. "
|
|
"Start it with: cd codebase/tools/platform-knowledge-ai && ./run start"
|
|
)
|
|
except httpx.TimeoutException:
|
|
return ToolResult.fail(
|
|
f"KV API request timed out after {_REQUEST_TIMEOUT}s"
|
|
)
|
|
|
|
if resp.status_code == 503:
|
|
return ToolResult.fail(
|
|
"Validator service not ready. "
|
|
"The embedding model may still be loading -- try again shortly."
|
|
)
|
|
|
|
if resp.status_code >= 500:
|
|
return ToolResult.fail(
|
|
f"KV API internal error (HTTP {resp.status_code}). "
|
|
"Check service logs for details."
|
|
)
|
|
|
|
if resp.status_code != 200:
|
|
return ToolResult.fail(
|
|
f"KV API returned unexpected status {resp.status_code}"
|
|
)
|
|
|
|
data = resp.json()
|
|
confidence: float = data.get("confidence", 0.0)
|
|
valid: bool = data.get("valid", False)
|
|
raw_docs = data.get("relevantDocs", [])
|
|
|
|
supporting_docs: list[dict[str, Any]] = []
|
|
if isinstance(raw_docs, list):
|
|
for doc in raw_docs:
|
|
if not isinstance(doc, dict):
|
|
continue
|
|
supporting_docs.append({
|
|
"score": doc.get("score", 0),
|
|
"path": doc.get("path", "unknown"),
|
|
"excerpt": doc.get("excerpt", "")[:200],
|
|
})
|
|
|
|
# Log low-confidence validations for feedback (best-effort)
|
|
if confidence < 0.5:
|
|
try:
|
|
from ...feedback import FeedbackLogger, ValidationEvent
|
|
|
|
logger = FeedbackLogger()
|
|
event = ValidationEvent.now(
|
|
content=statement,
|
|
valid=valid,
|
|
confidence=confidence,
|
|
)
|
|
logger.log_validation(event)
|
|
except Exception:
|
|
# Silently ignore feedback logging errors
|
|
pass
|
|
|
|
return ToolResult.success(
|
|
{
|
|
"statement": statement,
|
|
"valid": valid,
|
|
"confidence": confidence,
|
|
"supporting_docs": supporting_docs,
|
|
},
|
|
confidence=confidence,
|
|
valid=valid,
|
|
)
|