Catch a Hallucination: Ground, Quote and Audit Every Answer

"""Catch a hallucination: answer from a source, prove it with a quote, audit every claim."""
import json
import re

from openai import OpenAI

CLIENT = OpenAI()
ASSISTANT = "meta/llama-3.1-8b-instruct"   # answers customers
CHECKER = "meta/llama-3.3-70b-instruct"    # audits the answers
NOT_FOUND = "I can't find that in our policy."
HANDOFF = "I'm not sure about that one, so I've passed your question to a colleague who will reply by email."


def chat(model, system, user, max_tokens=200, client=None):
    client = client or CLIENT
    r = client.chat.completions.create(model=model, temperature=0, max_tokens=max_tokens,
                                       messages=[{"role": "system", "content": system}, {"role": "user", "content": user}])
    return r.choices[0].message.content.strip()


def json_in(text, open_ch="{", close_ch="}"):
    return json.loads(text[text.find(open_ch): text.rfind(close_ch) + 1])


# ---------- Step 1: no source ----------
def ask_free(question, client=None):
    """The assistant with a friendly system prompt and no policy: what it 'knows'."""
    # TODO (Step 1): return chat(ASSISTANT, <a friendly system prompt>, question, client=client)
    #   system prompt: "You are the customer assistant of Brightline Books, a UK bookshop. Answer briefly."
    #   (no policy: this is what the model says from its own training)
    raise NotImplementedError("Step 1: write ask_free()")


# ---------- Step 2: answer only from the source ----------
def ask_grounded(question, policy, client=None):
    """Answer from `policy` only; reply exactly NOT_FOUND when the policy does not say."""
    raise NotImplementedError("ask_grounded() arrives in Step 2")


# ---------- Step 3: show the evidence ----------
def normalise(text):
    return re.sub(r"\s+", " ", text).strip().lower()


def quote_is_verbatim(quote, policy):
    """True if `quote` appears word for word in the policy (ignoring case and spacing) and is not trivially short."""
    raise NotImplementedError("quote_is_verbatim() arrives in Step 3")


def ask_with_quote(question, policy, client=None):
    """{"answer": ..., "quote": the exact sentence from the policy that supports it ("" for NOT_FOUND)}."""
    raise NotImplementedError("ask_with_quote() arrives in Step 3")


# ---------- Step 4: audit every claim ----------
CLAIMS_PROMPT = ("Split the answer into short, separate factual claims, one fact each, keeping numbers exactly. "
                 'Reply with a JSON list of strings only, e.g. ["Delivery costs £3.95.", "Delivery takes 2 to 4 days."]')
VERDICT_PROMPT = ("You check one claim against a shop's policy. Reply with one word:\n"
                  "supported - the policy states it or it follows directly from the policy\n"
                  "contradicted - the policy says something different\n"
                  "not_in_policy - the policy does not mention it")


def extract_claims(answer, client=None):
    """The answer broken into a list of single-fact claims (CHECKER model)."""
    raise NotImplementedError("extract_claims() arrives in Step 4")


def verdict(claim, policy, client=None):
    """"supported", "contradicted" or "not_in_policy" (CHECKER model); anything unclear counts as not_in_policy."""
    raise NotImplementedError("verdict() arrives in Step 4")


def audit(answer, policy, client=None):
    """[{"claim": ..., "verdict": ...}, ...] for every claim in the answer."""
    raise NotImplementedError("audit() arrives in Step 4")


# ---------- Step 5: only send what survives ----------
ON_TOPIC_PROMPT = ("Does the answer directly answer the customer's question? A true statement about something else "
                   "does not count. Reply with one word: yes or no.")


def on_topic(question, answer, client=None):
    """True if the CHECKER says the answer addresses the question."""
    reply = chat(CHECKER, ON_TOPIC_PROMPT, f"Question: {question}\nAnswer: {answer}", max_tokens=3, client=client)
    return reply.strip().lower().startswith("yes")


def guarded_answer(question, policy, client=None):
    """Ask with a quote; hand off unless the quote is verbatim, the answer is on topic and every claim is supported.
    Returns {"question", "final", "status", "draft", "quote", "audit"} where status is
    "answered", "not_found" or "handed_off"."""
    raise NotImplementedError("guarded_answer() arrives in Step 5")