Langchain integration

The LangChain integration is a single class: KyvvuLangChainHandler. It implements LangChain's BaseCallbackHandler and translates LangChain's native execution events into Kyvvu log entries. You pass it as a callback when invoking your agent — no decorators, no changes to your agent code.

pip install kyvvu langchain langchain-core langchain-openai

---

How events are mapped

LangChain fires callback events at every stage of execution. The handler maps them to Kyvvu node types automatically:

LangChain event	Kyvvu node type
Chain start (top-level)	START_NODE
LLM call	LLM_CALL
Agent decides to use a tool	DECISION
Tool executes	TOOL_CALL
Agent finishes	END_NODE
Any error	ERROR

Nested chains (e.g. a chain calling another chain) are tracked by depth — only the outermost chain generates START_NODE and END_NODE, keeping the task structure clean.

---

Setup

Instantiate KyvvuLangChainHandler with your agent's registration fields. This also registers the agent with Kyvvu immediately:

from kyvvu.integrations.langchain import KyvvuLangChainHandler

handler = KyvvuLangChainHandler(
    api_key="KvKey-your-api-key",
    api_url="http://localhost:8000",
    agent_key="my-langchain-agent",        # stable ID you choose
    name="My LangChain Agent",
    purpose="Answers customer questions",
    owner_id="[email protected]",
    risk_classification="limited",         # "minimal" | "limited" | "high"
    environment="production",              # "development" | "qa" | "staging" | "production"
)

Then pass the handler as a callback when invoking your agent:

result = agent_executor.invoke(
    {"input": "What is the capital of France?"},
    config={"callbacks": [handler]}
)

That's the entire integration. Every LLM call, tool invocation, and decision made during that invoke() is automatically logged to Kyvvu.

---

Advanced options

handler = KyvvuLangChainHandler(
    api_key="KvKey-...",
    agent_key="my-agent",

    # Control what gets marked as external
    llm_external_call=True,              # mark LLM calls as external (default: True)
    tool_external_call=True,             # mark tool calls as external (default: True)

    # Detect write operations by tool name keywords
    tool_write_detection=["write", "send", "update", "delete", "create", "modify"],

    # Error and incident handling (same as base Kyvvu class)
    on_error="raise",                    # "raise" | "log" | "ignore"
    on_incident="log",                   # "raise" | "log" | "ignore"
    incident_severity_threshold="critical",

    # Task control
    auto_start_task=True,                # auto-log START_NODE on first chain (default: True)
)

tool_write_detection is a list of keywords checked against the tool name. If a tool's name contains any of these words, has_write_permission=True is set on its TOOL_CALL log entry — which policies can then enforce rules against.

---

Multiple invocations

The handler is stateful between invocations: task_started tracks whether a task is open. After on_agent_finish fires, the state resets automatically, so consecutive invoke() calls each produce their own clean task in Kyvvu.

If you need to reset manually (e.g. after an error that didn't reach on_agent_finish):

handler.reset_task()

---

Error handling

If a policy is violated during execution and on_incident="raise", KyvvuPolicyViolationError is raised. Catch it the same way as with the base SDK:

from kyvvu.exceptions import KyvvuPolicyViolationError

try:
    result = executor.invoke({"input": "..."}, config={"callbacks": [handler]})
except KyvvuPolicyViolationError as e:
    for incident in e.incidents:
        print(f"[{incident['severity']}] {incident['title']}")

---

Complete example

import os
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_react_agent
from langchain_core.prompts import PromptTemplate
from langchain_core.tools import Tool
from kyvvu.integrations.langchain import KyvvuLangChainHandler
import numexpr as ne

# One handler per agent process
handler = KyvvuLangChainHandler(
    api_key=os.getenv("KYVVU_API_KEY"),
    api_url=os.getenv("KYVVU_API_URL", "http://localhost:8000"),
    agent_key="math-agent",
    name="Math Agent",
    purpose="Solves mathematical problems",
    owner_id="[email protected]",
    risk_classification="limited",
    environment="production",
)

# Standard LangChain setup — nothing Kyvvu-specific below this line
llm = ChatOpenAI(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o-mini", temperature=0.0)

tools = [
    Tool(
        name="Calculator",
        func=lambda expr: str(ne.evaluate(expr.strip())),
        description="Evaluates mathematical expressions.",
    )
]

prompt = PromptTemplate.from_template("""
Answer the question using available tools.
Tools: {tools}
Format: Thought → Action [{tool_names}] → Action Input → Observation → Final Answer
Question: {input}
Thought:{agent_scratchpad}
""")

executor = AgentExecutor(
    agent=create_react_agent(llm, tools, prompt),
    tools=tools,
    handle_parsing_errors=True,
)

# Attach Kyvvu here — one line
result = executor.invoke(
    {"input": "What is 25 * 17 + 143?"},
    config={"callbacks": [handler]}
)
print(result["output"])

For more context on when to use the handler versus @kv.log_step decorators, see Custom Python Agent.