LangChain + Tork Governance

Policy Enforcement

Block unsafe content automatically

Chain Governance

Wrap any LangChain chain

Tool Safety

Callbacks govern tool inputs and outputs

LCEL Support

Works with modern LangChain

pip install tork-governance langchain langchain-openai

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from tork_governance.adapters.langchain import TorkCallbackHandler

# The handler builds its own on-device Tork() instance. No API key is
# needed: PII detection and the decision run on this machine.
handler = TorkCallbackHandler()

# Attach it to any LangChain component that accepts callbacks
llm = ChatOpenAI(model="gpt-4o", callbacks=[handler])

prompt = ChatPromptTemplate.from_template("Tell me about {topic}")
chain = prompt | llm

result = chain.invoke({"topic": "machine learning"})

# on_llm_start governs every prompt, on_llm_end every generation.
# Each call appends a dict: {"type", "receipt", "action"}.
for entry in handler.receipts:
    print(entry["type"], entry["action"], entry["receipt"].receipt_id)

Callback Events

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from tork_governance.adapters.langchain import create_governed_chain

llm = ChatOpenAI(model="gpt-4o")

# A plain LCEL chain
prompt = ChatPromptTemplate.from_template(
    "You are a helpful assistant. Answer: {question}"
)
chain = prompt | llm

# Wrap it: every string input is governed before the chain runs,
# and the output (str, or a message with .content) after it.
governed_chain = create_governed_chain(chain)

# Use it like the original chain
result = governed_chain.invoke({"question": "What is Python?"})
print(result.content)

# The last governance decision is kept on the wrapper
last = governed_chain.last_result
print(last.action.value, last.receipt.receipt_id)

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from tork_governance import Tork
from tork_governance.adapters.langchain import TorkCallbackHandler, TorkGovernedChain

# One Tork instance shared by both patterns
tork = Tork(policy_version="1.0.0")

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful coding assistant."),
    ("human", "{input}"),
])

llm = ChatOpenAI(model="gpt-4o")
output_parser = StrOutputParser()

chain = prompt | llm | output_parser

# Option 1: callback handler — governs LLM prompts and generations
handler = TorkCallbackHandler(tork=tork)
result = chain.invoke(
    {"input": "Write a hello world in Python"},
    config={"callbacks": [handler]},
)

# Option 2: wrap the whole chain — governs the chain's inputs and outputs
governed = TorkGovernedChain(chain, tork=tork)
result = governed.invoke({"input": "Explain recursion"})

from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.tools import tool
from tork_governance import Tork
from tork_governance.adapters.langchain import TorkCallbackHandler

tork = Tork()
handler = TorkCallbackHandler(tork=tork)

@tool
def search_database(query: str) -> str:
    """Search the internal database for information."""
    return f"Results for: {query}"

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """Send an email to the specified recipient."""
    # Govern the arguments yourself before the side effect: the adapter
    # has no per-tool wrapper, but Tork.govern() works on any string.
    governed_body = tork.govern(body).output
    return f"Email sent to {to} with body: {governed_body}"

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

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful research assistant."),
    MessagesPlaceholder(variable_name="chat_history", optional=True),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])

agent = create_openai_functions_agent(
    llm=llm,
    tools=[search_database, send_email],
    prompt=prompt,
)

executor = AgentExecutor(
    agent=agent,
    tools=[search_database, send_email],
    callbacks=[handler],
    verbose=True,
)

result = executor.invoke({
    "input": "Search for Q4 results and email a summary to manager@company.com"
})

# on_tool_start / on_tool_end govern each tool's input and output string
for entry in handler.receipts:
    print(f"{entry['type']}: {entry['action']} ({entry['receipt'].receipt_id})")

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from tork_governance import Tork, GovernanceAction
from tork_governance.adapters.langchain import TorkGovernedChain

# The default action on PII is REDACT (the text goes through with the PII
# replaced). To block instead, construct Tork with default_action=DENY.
strict = Tork(default_action=GovernanceAction.DENY)

llm = ChatOpenAI(model="gpt-4o")
prompt = ChatPromptTemplate.from_template("{input}")
chain = prompt | llm

governed = TorkGovernedChain(chain, tork=strict)

def safe_invoke(user_input: str) -> str:
    """Invoke the chain; a DENY decision raises ValueError."""
    try:
        result = governed.invoke({"input": user_input})
        return result.content if hasattr(result, "content") else str(result)
    except ValueError as e:
        # Message is "Input blocked: <receipt_id>" — the receipt stays
        # available on the Tork instance's decision, nothing else is exposed.
        return "I cannot process requests containing personal information."

response = safe_invoke("What's the weather like?")     # allowed
response = safe_invoke("My SSN is 123-45-6789")        # blocked: PII detected

# Note: the on-device adapter detects PII. It does not detect jailbreak
# attempts — that check is a separate server-side endpoint, POST /api/v1/jailbreak.

from tork_governance import Tork, GovernanceAction, GovernanceResult, Receipt
from tork_governance.adapters.langchain import (
    TorkCallbackHandler,    # Callback handler (prompts, generations, tool I/O)
    TorkGovernedChain,      # Chain wrapper (inputs, outputs)
    create_governed_chain,  # Factory for TorkGovernedChain
)