Cloudflare OS：真正值得看的不是聊天框，而是连接器边界

from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Protocol


@dataclass
class Document:
    source: str
    resource_id: str
    title: str
    text: str
    updated_at: str
    allowed_users: set[str]


class Connector(Protocol):
    def search(self, query: str) -> list[Document]:
        ...


class IncidentConnector:
    def __init__(self) -> None:
        now = datetime.now(timezone.utc).isoformat()
        self.documents = [
            Document(
                source="incident-system",
                resource_id="INC-1042",
                title="API latency investigation",
                text="The API latency increase is related to a database connection pool limit.",
                updated_at=now,
                allowed_users={"alice", "bob"},
            ),
            Document(
                source="incident-system",
                resource_id="INC-1043",
                title="Internal security review",
                text="This record is visible only to the security team.",
                updated_at=now,
                allowed_users={"security-team"},
            ),
        ]

    def search(self, query: str) -> list[Document]:
        words = query.lower().split()
        return [
            doc for doc in self.documents
            if any(word in (doc.title + " " + doc.text).lower() for word in words)
        ]


def retrieve_for_user(
    query: str,
    user: str,
    connectors: list[Connector],
) -> list[Document]:
    candidates = [doc for connector in connectors for doc in connector.search(query)]
    return [doc for doc in candidates if user in doc.allowed_users]


def build_context(documents: list[Document]) -> str:
    return "\n\n".join(
        f"[{doc.source}/{doc.resource_id}] {doc.title}\n"
        f"Updated: {doc.updated_at}\n{doc.text}"
        for doc in documents
    )


if __name__ == "__main__":
    query = "API latency"
    user = "alice"
    documents = retrieve_for_user(query, user, [IncidentConnector()])

    if not documents:
        print("No authorized context found.")
    else:
        print(build_context(documents))

You are an internal assistant. Answer only from the supplied context.
If the context is insufficient, say that you do not have enough information.
Always cite the source and updated time for each important claim.
Never reveal content that was excluded by the authorization filter.

User question:
{question}

Authorized context:
{context}