ai-governance-sidecar

ai-governance-sidecar

RAG lineage reporting for the AI Governance Gateway.

When your application retrieves documents to feed an LLM, this package records which documents were used and reports them to the governance gateway. That turns "the AI gave a wrong answer" into "Employee_Handbook.pdf v7 was stale and fed these 43 answers."

It hooks into your retriever automatically — you do not change your retrieval code. You do add two lines at startup and one line after each LLM call (see What you have to change).

Despite the name, this is not a container sidecar. There is nothing to deploy. It is a library that runs inside your existing application process.

---

Requirements

• Python 3.12+
• An onboarded application on the gateway, and its gateway API key
• One of: LangChain, LlamaIndex, or Vertex AI

Install

Pick the extra that matches your stack:

pip install "ai-governance-sidecar[langchain]"
pip install "ai-governance-sidecar[llamaindex]"
pip install "ai-governance-sidecar[vertexai]"

The bare pip install ai-governance-sidecar works too — patches for libraries you don't have installed silently no-op, so installing more than one extra is safe.

Configure

Two environment variables:

GOVERNANCE_GATEWAY_URL=https://your-gateway-url   # no trailing slash needed
GOVERNANCE_API_KEY=<your gateway API key>

Optional:

If the URL or key is missing the package logs one warning and does nothing else. It never blocks or breaks your application.

What you have to change

1. Patch at startup

At the very top of your entry point, before your framework imports:

import ai_governance_sidecar
ai_governance_sidecar.auto_patch()

auto_patch() is idempotent — calling it twice is harmless.

2. Flush after each LLM call

Pass the X-Request-Id header from the gateway's response, which is what links your retrievals to the request the gateway already logged:

response = client.post(f"{GATEWAY_URL}/v1/chat/completions", json=payload)
ai_governance_sidecar.flush(response.headers.get("x-request-id", ""))

That's it. Your retriever calls in between are captured automatically.

Full example (LangChain)

import ai_governance_sidecar
ai_governance_sidecar.auto_patch()          # must precede the langchain import

import httpx
from langchain_community.vectorstores import FAISS

GATEWAY_URL = "https://your-gateway-url"
API_KEY = "..."

retriever = FAISS.load_local("index", embeddings).as_retriever()

def answer(question: str) -> str:
    docs = retriever.invoke(question)       # captured automatically
    context = "\n\n".join(d.page_content for d in docs)

    resp = httpx.post(
        f"{GATEWAY_URL}/v1/chat/completions",
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={
            "model": "gemini-2.0-flash",
            "messages": [
                {"role": "system", "content": f"Use this context:\n{context}"},
                {"role": "user", "content": question},
            ],
        },
    )
    ai_governance_sidecar.flush(resp.headers.get("x-request-id", ""))
    return resp.json()["choices"][0]["message"]["content"]

---

What gets captured

Both common ways of retrieving are covered, so you do not have to write your retrieval code a particular way:

docs = retriever.invoke(question)                       # retriever pipelines
docs = vectorstore.similarity_search(question)          # direct search
docs = vectorstore.similarity_search_with_score(q)      # scores are recorded too
docs = vectorstore.max_marginal_relevance_search(q)     # search_type="mmr"
docs = await vectorstore.asimilarity_search(question)   # async variants

A retriever built with as_retriever() runs a vector store search underneath, so one search reaches both hooks. It is still recorded once, under the question you actually asked — not under any query the retriever rewrote internally.

Retrievers that are not backed by a vector store at all (BM25, Wikipedia, Tavily, your own) are captured as well.

If nothing is being captured, check the return value of auto_patch():

print(ai_governance_sidecar.auto_patch())
# {'langchain': True, 'langchain_vectorstore': True, 'llamaindex': False, ...}

langchain covers retrievers, langchain_vectorstore covers direct searches. All-False retrieval entries mean no supported library was importable when auto_patch() ran.

---

How it works

1. auto_patch() wraps the retriever and vector store search methods of whichever supported libraries are installed — including store classes you import after calling it.
2. Each retrieval enqueues its query and documents into a contextvars.ContextVar — so each asyncio task and each thread has its own queue, with no locking and no cross-request mixing.
3. flush(request_id) drains that queue and fires a fire-and-forget POST to /api/v1/rag/lineage on the gateway.
4. The gateway attaches the lineage to the request log it already wrote, evaluates your RAG policy, and records the result.

The POST authenticates with your gateway API key. The key identifies your application, and the application determines the organisation, so lineage always lands on your own tenant.

It never raises. Network errors, timeouts and gateway errors are caught and logged. A governance outage cannot take your application down.

---

Limitations — read these

These are real constraints, not edge cases:

• Retrieval and the LLM call must happen in the same asyncio task (or thread). The correlation uses a ContextVar. If you retrieve in one task and generate in another — a worker pool, a queue, run_in_executor — the queue will be empty at flush time and no lineage is recorded.
• You must call flush() yourself. Skip it and the retrievals are silently discarded on the next drain. There is no timer and no automatic flush.
• Forking web servers need care. Under Gunicorn or uWSGI with forking workers, call auto_patch() after the fork — in Gunicorn's post_fork hook, or at the top of your app factory. Patching in the parent process is lost when workers fork.
• Streaming responses: read X-Request-Id from the initial response headers, before consuming the stream — not after it exhausts.
• In synchronous applications flush() blocks briefly. With no running event loop it falls back to asyncio.run(), adding up to timeout_s (default 5 s) in the worst case. In async applications it schedules a task and returns immediately.

Troubleshooting

4xx failures are warned about once per status code per process — enough to notice, not enough to flood your logs.