SourceScore

Integration guide

LangGraph + SourceScore VERITAS

A LangGraph StateGraph with two VERITAS nodes: one retrieves signed claims, the other verifies the generated answer is actually backed by one before you return it.

Install

pip install langgraph langchain-openai requests

Graph state + the retrieve node

LangGraph nodes are plain functions that take the state and return a partial update. The retrieve node turns a question into signed VERITAS claims via /search, which returns a results array of claim summaries.

import requests
from typing import TypedDict, List
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI

VERITAS = "https://sourcescore.org/api/v1"

class State(TypedDict):
    question: str
    claims: List[dict]
    answer: str
    grounded: bool

def veritas_retrieve(state: State) -> dict:
    r = requests.get(
        f"{VERITAS}/search",
        params={"q": state["question"], "limit": 5},
        timeout=8,
    )
    r.raise_for_status()
    # /search returns {"results": [...]} — each item is a signed claim summary.
    return {"claims": r.json().get("results", [])}

The generate node

Build the prompt from the retrieved claims and force a citation by claim_id on every fact.

def generate(state: State) -> dict:
    context = "\n".join(
        f"[{c['id']}] {c['statement']} (confidence {c['confidence']})"
        for c in state["claims"]
    )
    prompt = (
        "Answer using ONLY the verified claims below. Cite every fact with "
        "[claim_id]. If the claims do not cover the question, say so.\n\n"
        f"Claims:\n{context}\n\nQuestion: {state['question']}\nAnswer:"
    )
    answer = ChatOpenAI(model="gpt-4o-mini").invoke(prompt).content
    return {"answer": answer}

The verify node

A retriever returns the closest claims; it does not confirm the generated answer is consistent with them. The verify node POSTs the answer to /verify, which returns a bestMatch only when a signed claim backs it at or above minConfidence.

def veritas_verify(state: State) -> dict:
    r = requests.post(
        f"{VERITAS}/verify",
        json={"claim": state["answer"], "minConfidence": 0.85},
        timeout=8,
    ).json()
    # /verify returns {"matches": [...], "bestMatch": {...} | absent}.
    return {"grounded": r.get("bestMatch") is not None}

Wire the graph

Retrieve first. If no claims match, short-circuit to END (don't let the model improvise). Otherwise generate, then verify.

def has_claims(state: State) -> str:
    return "generate" if state["claims"] else "no_claims"

g = StateGraph(State)
g.add_node("retrieve", veritas_retrieve)
g.add_node("generate", generate)
g.add_node("verify", veritas_verify)

g.set_entry_point("retrieve")
g.add_conditional_edges("retrieve", has_claims, {"generate": "generate", "no_claims": END})
g.add_edge("generate", "verify")
g.add_edge("verify", END)

app = g.compile()

out = app.invoke({"question": "Who introduced the Transformer architecture?"})
print(out["answer"])
print("grounded:", out["grounded"])

Why a verify node

The two failure modes a retriever can't catch — out-of-corpus assertions and fabricated citations — both surface here: if the answer isn't backed by a signed claim with ≥2 primary sources, bestMatch is absent and grounded is false. Route on that to retry, hand off to a human, or label the answer unverified. Free tier is 1,000 calls/month, no signup; same API the other guides use.

Next steps