用 Sciverse 做科学问答的 Citation Grounding

!pip install httpx
import os
os.environ["SCIVERSE_API_TOKEN"] = "sv-your-token-here"  # 替换为你的真实值

import os
import httpx

BASE = "https://api.sciverse.space"
TOKEN = os.environ["SCIVERSE_API_TOKEN"]
HEADERS = {"Authorization": f"Bearer {TOKEN}"}

def split_claims(draft: str) -> list:
    """将草稿拆分为独立论点句子"""
    sentences = [s.strip() for s in draft.split("。") if s.strip()]
    return [s for s in sentences if len(s) > 10]

async def search_evidence(claim: str):
    """对单个论点检索支持证据"""
    async with httpx.AsyncClient(timeout=30) as client:
        resp = await client.post(
            f"{BASE}/agentic-search",
            headers=HEADERS,
            json={"query": claim, "top_k": 5}
        )
        resp.raise_for_status()
        return (resp.json().get("hits") or [])

draft = "mRNA 疫苗使用可电离脂质纳米颗粒(iLNP)包裹 mRNA。其中 MC3 是最广泛使用的可电离脂质。LNP 的粒径通常在 80-100nm。"
claims = split_claims(draft)
print(f"Split into {len(claims)} claims")

async def verify_with_content(hit: dict, claim: str) -> dict:
    """读取原文验证证据是否真正支持论点"""
    async with httpx.AsyncClient(timeout=30) as client:
        resp = await client.get(
            f"{BASE}/content",
            headers=HEADERS,
            params={"doc_id": hit["doc_id"], "offset": hit.get("offset", 0), "limit": 1000}
        )
        resp.raise_for_status()
        data = resp.json()
        # 检查原文中是否包含与论点相关的关键词
        text = data["text"].lower()
        claim_keywords = [w for w in claim.lower().split() if len(w) > 3]
        match_count = sum(1 for kw in claim_keywords if kw in text)
        match_ratio = match_count / max(len(claim_keywords), 1)
        return {
            "doc_id": hit["doc_id"],
            "offset": hit.get("offset", 0),
            "quote": data["text"][:150],
            "match_ratio": match_ratio,
            "verified": match_ratio >= 0.3 and hit["score"] >= 0.7
        }

async def ground_claims(claims: list):
    results = []
    for claim in claims:
        hits = await search_evidence(claim)
        if hits and hits[0]["score"] >= 0.6:
            verification = await verify_with_content(hits[0], claim)
            results.append({"claim": claim, **verification})
        else:
            results.append({"claim": claim, "verified": False, "doc_id": None})
        status = "\\u2713" if results[-1]["verified"] else "\\u2717"
        print(f"  {status} {claim[:50]}...")
    return results

results = await ground_claims(claims)

def build_grounded_answer(results: list) -> dict:
    citations = []
    grounded_parts = []
    unverified = []

    for r in results:
        if r["verified"]:
            cite_id = len(citations) + 1
            citations.append({
                "id": cite_id,
                "doc_id": r["doc_id"],
                "offset": r.get("offset", 0),
                "quote": r.get("quote", ""),
                "verified": True
            })
            grounded_parts.append(f"{r['claim']} [{cite_id}]")
        else:
            grounded_parts.append(f"{r['claim']} [unverified]")
            unverified.append(r["claim"])

    return {
        "grounded_answer": "\\u3002".join(grounded_parts) + "\\u3002",
        "citations": citations,
        "unverified_claims": unverified
    }

final = build_grounded_answer(results)
print(f"\
Grounded answer:\
{final['grounded_answer']}")
print(f"\
Citations: {len(final['citations'])}")
print(f"Unverified: {len(final['unverified_claims'])}")
for c in final["citations"]:
    print(f"  [{c['id']}] {c['doc_id']} (offset: {c['offset']})")