实时网络 RAG：用搜索检索，带引用作答

import os
import re

import requests

API = "https://llmwebsearch.com/api/v1/search"
HEADERS = {"Authorization": "Bearer " + os.environ["LLMWEBSEARCH_API_KEY"]}


def live_chunks(query, read=3, chunk_len=1200):
    """Search, then cut each page that was read into heading-based chunks tagged with its source."""
    body = {"q": query, "max_results": 5, "read": read, "max_chars": 12000}
    results = requests.post(API, headers=HEADERS, json=body, timeout=60).json()["results"]
    chunks = []
    for r in results:
        if r.get("content_status") != "ok":
            continue
        for section in re.split(r"\n(?=#{1,3} )", r["content"]):
            section = section.strip()
            for start in range(0, len(section), chunk_len):
                chunks.append({"url": r["url"], "title": r["title"], "text": section[start:start + chunk_len]})
    return chunks


def top_chunks(chunks, question, k=6):
    """A plain keyword score; swap in your embedding model for better ranking."""
    words = set(re.findall(r"\w{4,}", question.lower()))

    def score(chunk):
        return sum(1 for w in re.findall(r"\w{4,}", chunk["text"].lower()) if w in words)

    return sorted(chunks, key=score, reverse=True)[:k]


def build_prompt(question):
    picked = top_chunks(live_chunks(question), question)
    urls = list(dict.fromkeys(c["url"] for c in picked))
    blocks = [f"[{urls.index(c['url']) + 1}] {c['title']}\n{c['text']}" for c in picked]
    sources = "\n".join(f"[{i}] {u}" for i, u in enumerate(urls, 1))
    return (
        "Use only the context to answer and cite sources as [n].\n\n"
        + "\n\n".join(blocks)
        + f"\n\nSources:\n{sources}\n\nQuestion: {question}"
    )