LangChain and LangGraph Compliance Witnessing

v0.7.9 RELEASED – LangChain callback handler + LangGraph proxy adapter. Available on PyPI and npm.

1. Two Integration Patterns

LangChain and LangGraph solve different problems. LangChain composes LLM calls into chains and pipelines using LCEL (LangChain Expression Language). LangGraph builds stateful, multi-step agents with explicit graph topologies. SWT3 provides a dedicated adapter for each.

Framework	Integration	What It Witnesses	When to Use
LangChain	SWT3CallbackHandler	Every LLM call and retriever result in any chain or pipeline	LCEL chains, sequential pipelines, RAG with retrievers, any LangChain-managed LLM call
LangGraph	wrap_langgraph / wrapLangGraph	Each graph invocation: input state hash, output state hash, model, latency	Stateful agents, multi-step workflows, compiled graphs with branching or cycles

The callback handler is event-driven. It hooks into LangChain's callback system and fires on every LLM call and retriever result, regardless of which provider is underneath. The LangGraph proxy intercepts graph-level invoke(), stream(), ainvoke(), and astream() calls and witnesses the complete state transition.

2. LangChain Callback Handler

The SWT3CallbackHandler plugs into LangChain's native callback system. Pass it once when constructing your LLM, and every call through that LLM -- whether it originates from a chain, an agent, or a direct invocation -- is witnessed automatically.

from langchain_openai import ChatOpenAI
from swt3_ai import Witness
from swt3_ai.adapters.langchain import SWT3CallbackHandler

witness = Witness(
    endpoint="https://your-instance.example.com/api/v1/witness",
    api_key="axm_your_key",
    tenant_id="YOUR_TENANT",
)
handler = SWT3CallbackHandler(witness)

llm = ChatOpenAI(model="gpt-4o", callbacks=[handler])
response = llm.invoke("Analyze this regulatory filing.")
# Every LLM call through this chain is witnessed automatically

witness.flush()

The callback handler captures three events:

• on_chat_model_start -- records the prompt, the serialized model name, and the invocation timestamp.
• on_llm_end -- records the response content, token usage (prompt + completion), and latency.
• on_retriever_end -- records retrieved documents as RAG context (covered in Section 3).

It works with ANY LangChain-supported LLM. OpenAI, Anthropic, Bedrock, Ollama, local models -- the handler does not care which provider is underneath. It extracts the actual model name from LangChain's internal serialization, so the witness anchor always records the real model, not a generic wrapper name.

System Prompt Witnessing

The callback handler captures system prompts automatically. When on_chat_model_start fires, the handler extracts system messages from the messages array and hashes them independently as system_prompt_hash. This works for any LangChain chat model -- OpenAI, Anthropic, Bedrock, or local -- without configuration.

llm = ChatOpenAI(model="gpt-4o", callbacks=[handler])

# System prompt passed via LangChain's standard pattern
from langchain_core.messages import SystemMessage, HumanMessage

response = llm.invoke([
    SystemMessage(content="You are a claims adjuster. Never approve claims over $50,000 without escalation."),
    HumanMessage(content="Review claim #7721."),
])
# system_prompt_hash captures the adjuster instructions for this call

3. RAG Pipeline Witnessing

When a LangChain retriever fetches documents, the callback handler automatically witnesses the retrieval as RAG context. No additional code is needed beyond what you already wrote in Section 2.

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from swt3_ai import Witness
from swt3_ai.adapters.langchain import SWT3CallbackHandler

witness = Witness(
    endpoint="https://your-instance.example.com/api/v1/witness",
    api_key="axm_your_key",
    tenant_id="YOUR_TENANT",
)
handler = SWT3CallbackHandler(witness)

embeddings = OpenAIEmbeddings()
retriever = FAISS.from_texts(texts, embeddings).as_retriever()

llm = ChatOpenAI(model="gpt-4o", callbacks=[handler])

# LCEL chain: retriever pipes into LLM
chain = retriever | llm
result = chain.invoke("What is our data retention policy?")
# Both the retrieval and the LLM call are witnessed

witness.flush()

When on_retriever_end fires, the handler captures each retrieved document's content hash, source identifier, and similarity score as RagChunk objects. It then calls witness.witness_rag_context() internally, minting an AI-RAG.1 anchor without any additional code from you.

This means every RAG pipeline built with LangChain retrievers automatically produces provenance evidence: which documents were retrieved, from which source, with what relevance, for which query -- all fingerprinted and anchored.

4. LangGraph Agent Witnessing

LangGraph agents are stateful graphs: they take an input state, pass it through nodes (which may call LLMs, tools, or other logic), and produce an output state. The wrap_langgraph proxy intercepts graph-level invocations and witnesses the complete state transition.

Python

from langgraph.graph import StateGraph
from swt3_ai import Witness
from swt3_ai.adapters.langgraph import wrap_langgraph

witness = Witness(
    endpoint="https://your-instance.example.com/api/v1/witness",
    api_key="axm_your_key",
    tenant_id="YOUR_TENANT",
    agent_id="claims-processor",
)

graph = StateGraph(...)  # your graph definition
app = graph.compile()
witnessed_app = wrap_langgraph(app, witness)

result = witnessed_app.invoke({"messages": [("user", "Process this claim")]})
# The entire graph execution is witnessed as a single anchor

witness.flush()

TypeScript

import { StateGraph } from '@langchain/langgraph';
import { Witness, wrapLangGraph } from '@tenova/swt3-ai';

const witness = new Witness({
  endpoint: 'https://your-instance.example.com/api/v1/witness',
  apiKey: 'axm_your_key',
  tenantId: 'YOUR_TENANT',
  agentId: 'claims-processor',
});

const graph = new StateGraph({...}); // your graph definition
const app = graph.compile();
const witnessedApp = wrapLangGraph(app, witness);

const result = await witnessedApp.invoke({
  messages: [['user', 'Process this claim']],
});

await witness.flush();

The proxy intercepts four methods: invoke(), stream(), ainvoke(), and astream(). For invoke() and ainvoke(), the input state dictionary is hashed as prompt_hash and the output state is hashed as response_hash. For streaming variants, the proxy accumulates the last chunk and uses it as the response hash after the stream completes.

5. Streaming Graph Execution

LangGraph's stream() and astream() yield chunks as the graph executes. The SWT3 proxy passes every chunk through untouched -- your application receives the stream exactly as it would without witnessing. The witness anchor is minted after the stream completes.

Synchronous Streaming

for chunk in witnessed_app.stream({"messages": [("user", "Search and summarize")]}):
    print(chunk)  # Every chunk passes through immediately
# Witness anchor minted after stream completes

Async Streaming

async for chunk in witnessed_app.astream({"messages": [("user", "Search and summarize")]}):
    print(chunk)  # Every chunk passes through immediately
# Witness anchor minted after stream completes

The last-chunk pattern. During streaming, the proxy records a timestamp at stream start and accumulates a reference to the last chunk yielded. After the stream's iterator is exhausted, the proxy hashes the last chunk as the response hash, calculates total latency, and mints the anchor. Every chunk is yielded to the developer with zero added latency -- the witnessing overhead is deferred entirely to after the stream ends.

6. Combining Both Patterns

Complex architectures often use LangChain LLMs inside LangGraph agents. A LangGraph node might call a LangChain ChatModel, which might use a retriever. With both adapters active, you get a two-layer evidence chain: per-LLM-call evidence from the callback handler and per-graph-execution evidence from the proxy.

from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, MessagesState, START, END
from swt3_ai import Witness
from swt3_ai.adapters.langchain import SWT3CallbackHandler
from swt3_ai.adapters.langgraph import wrap_langgraph

witness = Witness(
    endpoint="https://your-instance.example.com/api/v1/witness",
    api_key="axm_your_key",
    tenant_id="YOUR_TENANT",
    agent_id="research-agent",
)

# Layer 1: Callback handler witnesses every LLM call
handler = SWT3CallbackHandler(witness)
llm = ChatOpenAI(model="gpt-4o", callbacks=[handler])

# Build a LangGraph agent that uses the witnessed LLM
def research_node(state: MessagesState):
    return {"messages": [llm.invoke(state["messages"])]}

def summarize_node(state: MessagesState):
    return {"messages": [llm.invoke(state["messages"])]}

graph = StateGraph(MessagesState)
graph.add_node("research", research_node)
graph.add_node("summarize", summarize_node)
graph.add_edge(START, "research")
graph.add_edge("research", "summarize")
graph.add_edge("summarize", END)

# Layer 2: Proxy witnesses the overall graph execution
witnessed_app = wrap_langgraph(graph.compile(), witness)

result = witnessed_app.invoke({"messages": [("user", "Research and summarize AI regulation")]})
# 3 anchors minted: research LLM call + summarize LLM call + graph execution

witness.flush()

7. Model-Agnostic Evidence

LangChain abstracts the LLM provider behind a uniform interface. The SWT3 callback handler extracts the real model identity from LangChain's internal serialization, regardless of which wrapper class you use. This means the witness anchor always records the actual model and provider, not the LangChain class name.

LangChain Provider Class	Underlying Model	Provider Tag in Evidence
ChatOpenAI	gpt-4o	openai
ChatAnthropic	claude-sonnet-4	anthropic
ChatBedrock	us.anthropic.claude-sonnet-4-20250514	bedrock
ChatGoogleGenerativeAI	gemini-2.0-flash	google
OllamaLLM	llama3.2	langchain
ChatMistralAI	mistral-large-latest	mistral

This matters for multi-model pipelines. A single LCEL chain might route different queries to different models based on complexity, cost, or latency. The callback handler witnesses each call with the actual model used, so your evidence trail reflects reality -- not the abstraction layer.

8. Regulatory Evidence Mapping

Every anchor minted through LangChain or LangGraph maps to specific regulatory requirements. The evidence is produced at inference time, not reconstructed after the fact.

Obligation	Regulation	SWT3 Evidence	Procedure
Automatic logging of AI operations	EU AI Act Art. 12	Per-inference anchor with model ID, prompt hash, response hash	AI-INF.1
Transparency of AI system behavior	EU AI Act Art. 13	Model identity and provider extracted from LangChain serialization	AI-INF.1
Document data sources used by AI	EU AI Act Art. 10	RAG retrieval provenance with document hashes and source IDs	AI-RAG.1
Track AI system interfaces and dependencies	NIST AI RMF MAP 1.5	Two-layer chain: graph execution + individual LLM calls	AI-INF.1
Mechanisms to track AI behavior	NIST AI RMF MN 2.2	Decision-level witnessing across all graph nodes	AI-INF.1
Ongoing model monitoring	SR 11-7 / SR 26-2	Model hash drift detection across time	AI-INF.1
Audit events for AI decisions	SOC 2 CC7.2	Cryptographic fingerprint per orchestrated decision	AI-INF.1
Agent identity and accountability	NIST AI RMF GV 1.1	agent_id tag on every graph-level anchor	AI-ID.1

9. Quick Start

Install

# Python (LangChain + LangGraph)
pip install swt3-ai langchain-openai langgraph

# TypeScript (LangGraph only)
npm install @tenova/swt3-ai @langchain/langgraph

LangChain -- Minimal Example (Python)

from langchain_openai import ChatOpenAI
from swt3_ai import Witness
from swt3_ai.adapters.langchain import SWT3CallbackHandler

witness = Witness(
    endpoint="https://your-instance.example.com/api/v1/witness",
    api_key="axm_your_key",
    tenant_id="YOUR_TENANT",
)

llm = ChatOpenAI(model="gpt-4o", callbacks=[SWT3CallbackHandler(witness)])
response = llm.invoke("Summarize this document.")
witness.flush()

LangGraph -- Minimal Example (Python)

from langgraph.graph import StateGraph
from swt3_ai import Witness
from swt3_ai.adapters.langgraph import wrap_langgraph

witness = Witness(
    endpoint="https://your-instance.example.com/api/v1/witness",
    api_key="axm_your_key",
    tenant_id="YOUR_TENANT",
)

app = StateGraph(...).compile()  # your compiled graph
witnessed_app = wrap_langgraph(app, witness)
result = witnessed_app.invoke({"messages": [("user", "Process this request")]})
witness.flush()

LangGraph -- Minimal Example (TypeScript)

import { StateGraph } from '@langchain/langgraph';
import { Witness, wrapLangGraph } from '@tenova/swt3-ai';

const witness = new Witness({
  endpoint: 'https://your-instance.example.com/api/v1/witness',
  apiKey: 'axm_your_key',
  tenantId: 'YOUR_TENANT',
});

const app = new StateGraph({...}).compile(); // your compiled graph
const witnessedApp = wrapLangGraph(app, witness);
const result = await witnessedApp.invoke({ messages: [['user', 'Process this request']] });
await witness.flush();

10. References

• LangChain Python Documentation
• LangGraph Documentation
• Multi-Agent Witness Architecture Guide
• Trust Propagation Engine Guide
• Runtime Surface Governance Guide
• SWT3 SDK Documentation
• swt3-ai on PyPI
• @tenova/swt3-ai on npm

Other Witnessing Guides

• Anthropic Claude Compliance Witnessing
• OpenAI GPT Compliance Witnessing
• Google Vertex AI and Gemini Compliance Witnessing
• DeepSeek GPAI Compliance Witnessing
• Groq Compliance Witnessing