Skip to main content

LangChain Integration

Add governance to your LangChain chains and agents with a callback handler. The callback records LangChain activity; use GovernedTool or explicit cz.guard() calls for enforcement.

Setup​

pip install controlzero langchain langchain-openai
from controlzero import Client
from controlzero.integrations.langchain import ControlZeroCallbackHandler
from langchain_openai import ChatOpenAI

cz = Client(api_key="cz_live_your_api_key_here")
handler = ControlZeroCallbackHandler(cz)

# Add the handler to an LLM to record its LLM calls
llm = ChatOpenAI(model="gpt-5.4", callbacks=[handler])

Create the handler and pass it to your LLM to record LLM calls. To record agent or chain activity, including tool invocations, also pass the handler to the agent or chain that dispatches those tools, or attach it to the tools themselves. Enforcement requires guarded tools or explicit enforcing cz.guard() calls.

What the Callback Records​

The callback handler intercepts two types of events:

LangChain EventPolicy ActionPolicy Resource
LLM call (on_llm_start)llm.generatemodel/{model_name}
Tool call (on_tool_start)tool.calltool/{tool_name}

The handler extracts model and tool metadata and records it in the audit trail; it does not evaluate or block.

Audit-only by default

ControlZeroCallbackHandler logs the LLM and tool events emitted by the components to which it is attached, but does not block actions on its own. To enforce policies and stop execution on a deny decision, wrap your tools with GovernedTool or add explicit enforcing cz.guard() calls inside each tool function.

Example: Research Agent with Policy Enforcement​

from controlzero import Client
import controlzero
from controlzero.integrations.langchain import ControlZeroCallbackHandler
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool

# --- Setup Control Zero ---
cz = Client(api_key="cz_live_your_api_key_here")
handler = ControlZeroCallbackHandler(cz)

# --- Define tools ---
@tool
def search_web(query: str) -> str:
    """Search the web for information."""
    cz.guard("search_web", args={"query": query}, raise_on_deny=True)
    return f"Results for: {query}"

@tool
def read_database(query: str) -> str:
    """Query the internal database."""
    cz.guard("read_database", args={"query": query}, raise_on_deny=True)
    return f"DB results for: {query}"

# --- Create agent with governance ---
llm = ChatOpenAI(model="gpt-5.4", callbacks=[handler])

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

agent = create_tool_calling_agent(llm, [search_web, read_database], prompt)
executor = AgentExecutor(
    agent=agent,
    tools=[search_web, read_database],
    callbacks=[handler],
)

# --- Run it ---
try:
    result = executor.invoke({"input": "Find recent sales data"})
    print(result["output"])
except controlzero.PolicyDeniedError as e:
    print(f"Blocked by policy: {e.decision.reason}")

Example Policy for the Research Agent​

Define this in the Control Zero dashboard:

{
  "name": "research-agent-policy",
  "description": "Allow web search but block database access",
  "rules": [
    { "effect": "allow", "action": "llm:generate", "resource": "model/gpt-5.4" },
    { "effect": "allow", "action": "tool:call", "resource": "tool/search_web" },
    { "effect": "deny", "action": "tool:call", "resource": "tool/read_database" }
  ]
}

What happens at runtime:

  • When the agent calls GPT-5.4 for reasoning: ALLOWED (matches rule 1).
  • When the agent tries to use search_web: ALLOWED (matches rule 2).
  • When the agent tries to use read_database: DENIED (matches rule 3). The enforcing guard raises PolicyDeniedError before the database logic runs, and the decision is logged to the audit trail.

Using with Chains​

The handler works with any LangChain component that accepts callbacks:

from langchain_core.output_parsers import StrOutputParser

chain = llm | StrOutputParser()
# The handler attached to llm records this chain's LLM calls

result = chain.invoke("Summarize this report")

Next Steps​