rankigi-langchain 1.0.0

LangChain (Python) callback handler that adds RANKIGI cryptographic receipts to every LLM, tool, agent, and retriever step.

rankigi-langchain

LangChain (Python) callback handler that adds RANKIGI cryptographic receipts to every LLM, tool, agent, retriever, and LangGraph node.

Compatible with langchain-core >= 0.1 and LangGraph.

Install

pip install rankigi-langchain

Setup

First, enroll a connector passport in your RANKIGI dashboard (Agents → New, type "connector"). Copy the credentials into your environment:

export RANKIGI_API_KEY=rnk_...
export RANKIGI_AGENT_ID=<uuid>
export RANKIGI_PASSPORT_ID=<uuid>
export RANKIGI_SIGNING_KEY=<base64 ed25519 seed>

Then attach the callback:

from rankigi_langchain import RankigiCallbackHandler
from langchain.agents import AgentExecutor

executor = AgentExecutor(
    agent=agent,
    tools=tools,
    callbacks=[RankigiCallbackHandler(agent_tag="checkout-service-prod")],
)

Or pass per-invocation via RunnableConfig:

executor.invoke(
    {"input": "..."},
    config={"callbacks": [RankigiCallbackHandler(agent_tag="checkout-service-prod")]},
)

LangGraph works the same way:

graph.invoke(
    state,
    config={"callbacks": [RankigiCallbackHandler(agent_tag="research-agent")]},
)

What gets a receipt

LangChain callback	RANKIGI action
on_llm_end / on_chat_model_start+end	llm_call
on_tool_end	tool_call
on_agent_action	agent_action
on_agent_finish	agent_finish
on_chain_start/end (LangGraph nodes + top-level only)	chain_start / chain_end
on_retriever_end	retrieval
any on_*_error callback	*_error with severity="warn"

What gets hashed

Prompts, completions, tool inputs, tool outputs, agent reasoning, retriever queries, and document contents are SHA-256-hashed locally by the SDK before any network call. Only the hashes plus safe metadata (model name, token counts, latency, run/parent IDs, LangGraph node names) reach RANKIGI.

Safety

Every callback is wrapped in try/except. SDK or network failures are absorbed silently — the callback never raises into LangChain. The underlying SDK has a durable disk queue that absorbs RANKIGI downtime and retries.

Explicit credentials

If you can't use env vars, pass the credentials directly:

RankigiCallbackHandler(
    api_key=os.environ["RANKIGI_API_KEY"],
    agent_id=os.environ["RANKIGI_AGENT_ID"],
    passport_id=os.environ["RANKIGI_PASSPORT_ID"],
    signing_private_key=os.environ["RANKIGI_SIGNING_KEY"],
    agent_tag="checkout-service-prod",
)

Test injection

For unit tests, pass a mock client:

class FakeClient:
    def track(self, **kwargs): ...

RankigiCallbackHandler(client=FakeClient())

License

MIT