Integrating Kyno with LangGraph

Adapter overview · CrewAI guide

Kyno puts direction in graph state. Your model-calling node puts that direction into the model's input.

For a complete run with an operator changing direction between model calls, see the customer-support example. It includes optional recording; that recording code is not required by the adapter.

Install and connect

pip install "kyno[langgraph]"

import kyno
from kyno.sdk import DetailLevel, PullPolicy

connection = kyno.connect()
binder = connection.binder(
    "customer-support",
    detail=DetailLevel.FULL,
    policy=PullPolicy(fail_closed=True),
)

This uses your default remote profile. See connection configuration for named profiles and explicit credentials. Keep the connection open while the graph runs, then call connection.close().

This guide requests full direction and stops if a read fails. These are example settings, not the SDK defaults.

Required integration

Kyno supplies KynoState, direction_node, and pull_before. You supply the graph and the node that calls your model.

KynoState is a TypedDict that declares the fields Kyno uses in graph state: direction content, its constitution key and version, and delivery metadata. Inherit it in your own state schema so LangGraph preserves those fields between nodes and in checkpoints. You can add application fields, such as output in the example below.

1. Inherit KynoState in your graph's state schema. LangGraph only carries keys declared by that schema.
2. Wrap your model-calling node with pull_before, or place a direction_node before it in the graph. Choose one boundary; using both there would pull twice.
3. Include state["kyno_direction"] in the model's input. The adapter populates graph state; it does not modify your model's messages for you.

This node drafts a response to a delivery complaint using the binder above and your configured chat model:

from kyno.adapters.langgraph import KynoState, pull_before
from kyno.sdk import BindingStatus

SCENARIO = (
    "A customer paid $40 for express delivery. The package arrived two days late. "
    "They ask for the delivery fee back and an explanation. Tracking confirms the delay. "
    "You may draft a reply, propose a refund of up to $40, or propose escalation to a person. "
    "Do not execute any action or claim a refund has already been issued. "
    "Choose a response and explain the tradeoff using the supplied mission and principles."
)


class State(KynoState, total=False):
    output: str


@pull_before(binder)
def answer(state):
    if state["kyno_version"] == 0 or state["kyno_binding_status"] != BindingStatus.PULLED:
        raise ValueError("A current, written constitution is required before calling the model")
    messages = [
        {"role": "system", "content": state["kyno_direction"]},
        {"role": "user", "content": SCENARIO},
    ]
    return {"output": model.invoke(messages).content}

answer is an example name for your own graph node, not a Kyno function to implement or override. Decorate your existing node and register it in your graph as usual. model is your application's model client.

The decorator pulls once before the node runs, applies the binder's failure policy, and supplies direction and delivery metadata in state. You do not need to implement those steps yourself. You also do not need receipt storage, run IDs, or step IDs for this integration to work.

What Kyno provides

KynoState carries the selected key in state["kyno_constitution_key"]. The direction node and decorator populate it alongside kyno_version, so downstream nodes and saved checkpoints identify the direction they received:

constitution_key = state["kyno_constitution_key"]
version = state["kyno_version"]

direction_from_state(state) reads state["kyno_constitution_key"] into direction.constitution_key. With an empty state, or after a failed first pull with no explicit constitution key, this property is None. A successful Core reply supplies the selected constitution key. For example, an empty state gives:

from kyno.adapters.langgraph import direction_from_state

empty = direction_from_state({})
assert empty.constitution_key is None
assert empty.version == 0

Before your work node runs, direction_node or pull_before supplies the direction and sets state["kyno_binding_status"] automatically. Your application does not need to set or convert this value. It is one of three named values from BindingStatus, imported from kyno.sdk:

• BindingStatus.PULLED: this binding uses direction from a successful read, including an unchanged or unwritten constitution.
• BindingStatus.CACHED: the binder retained direction after a failed read, or kept a newer cached version when an older overlapping response arrived.
• BindingStatus.EMPTY: the read failed and no cached direction was available.

The status describes how this step received its direction. It does not establish that the version is still the newest or that the model followed it. See the shared status reference.

Optional: LangGraph checkpoints

A checkpoint is a saved snapshot of a workflow's state that LangGraph can use to resume the workflow later. A checkpointer is the component that saves and loads those snapshots. See LangGraph's persistence documentation.

Kyno adds direction and binding status to graph state. If your graph uses a LangGraph checkpointer, it saves those fields alongside the rest of the workflow's state. LangGraph's default serializer restores the BindingStatus enum when you load that checkpoint; no manual conversion is needed. Kyno does not configure checkpoint storage for you. You do not need checkpointing just to give your agents direction.

Continue the support-answer example by adding this code after the answer function. It uses the State class and decorated answer node already defined there, with the same open connection, binder, and model. You do not need another direction node: answer already pulls through @pull_before.

This builds a one-node graph and adds checkpoint storage. Running it calls your configured model once and may incur provider charges:

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph

graph = (
    StateGraph(State)
    .add_node("answer", answer)
    .add_edge(START, "answer")
    .add_edge("answer", END)
    .compile(checkpointer=InMemorySaver())
)
config = {"configurable": {"thread_id": "support-example"}}
graph.invoke({}, config)

checkpoint = graph.get_state(config)
saved = checkpoint.values
print("Saved direction version:", saved["kyno_version"])
print("Binding status at that step:", saved["kyno_binding_status"].value)
print("Saved answer:", saved["output"])

graph.invoke runs answer: its decorator pulls direction, then the node sends that direction and the complaint to your model. LangGraph saves the state under the support-example thread ID. graph.get_state(config) retrieves that thread's latest checkpoint; checkpoint.values is the dictionary of saved state fields.

The first printed line identifies the direction version supplied to that step. The second says how it was obtained: for example, pulled means the read succeeded at that step. .value gets this text from the saved BindingStatus enum. Reading the checkpoint does not contact Kyno, so it cannot tell you whether a newer direction version has since been applied.

InMemorySaver keeps checkpoints only in this Python process. To retain them after a restart, configure persistent storage through LangGraph. The runnable customer-support script does not enable checkpointing; its optional JSONL recording is a separate feature.

A direction node before a fan-out supplies the same snapshot to its branches. Later pulls do not change earlier receipts. Resuming a checkpoint without another pull preserves the original binding status; it does not establish that the saved version is still current. Work nodes should leave the kyno_ direction keys unchanged so the checkpoint describes their input.

Direction identity in checkpoints

Saved direction uses kyno_constitution_key to identify the constitution and kyno_version to identify its version. Keep these fields with the direction content when storing application-owned receipts.

For example, direction_from_state restores direction from these fields:

from kyno.adapters.langgraph import direction_from_state

saved_direction = {
    "kyno_constitution_key": "support",
    "kyno_version": 3,
    "kyno_mission": "Help customers",
}

direction = direction_from_state(saved_direction)
assert direction.constitution_key == "support"
assert direction.version == 3
assert direction.mission == "Help customers"

This example shows a minimal direction snapshot. Preserve the other direction fields, such as principles and declaration, when present. direction_from_state raises ValueError if direction fields are present without kyno_constitution_key, or if the key is invalid. State without any Kyno fields returns empty direction for default at version zero.

Failure behavior

The SDK's default policy uses cached direction after a failed read, or empty version-0 direction if no value was cached. The binder in this guide instead uses PullPolicy(fail_closed=True): a failed read stops the graph before the model call. The answer node also rejects version 0, because a successful read can return an unwritten constitution.

Configure this policy when creating the binder, before decorating answer or building the graph:

from kyno.sdk import DetailLevel, PullPolicy

binder = connection.binder(
    "customer-support",
    detail=DetailLevel.FULL,
    policy=PullPolicy(fail_closed=True),
)

This is the same binder configuration used in the setup section; you do not need to create it a second time. See the shared failure and status reference.

Optional recording

state["kyno_recording"] carries the server's recording receipt as a plain dictionary, or None when the source supplied no receipt. It travels with the direction through graph state and checkpoints; it is not added to the model's direction text. A recorded receipt contains a record_id that identifies a stored Kyno delivery record. A disabled or failed receipt has no record ID. These recording statuses are separate from kyno_binding_status, which describes whether direction was read or cached.

Recording is disabled by default. When recording succeeds, the record describes Core's direction response before your graph uses it. That record does not establish that the graph inserted direction into a model request, that the model completed, or that its answer followed direction. If a pull fails and the binder supplies cached direction, the receipt still describes the original response's recording outcome; the failed pull creates no new ID.

To associate an answer with its delivery record, capture the receipt inside the model-calling node and store its ID alongside the output in your own storage. In the answer function above, replace its final return with:

recording = state["kyno_recording"]
output = model.invoke(messages).content
save_answer(
    output=output,
    record_id=recording.get("record_id") if recording is not None else None,
    supplied_message=state["kyno_direction"],
)
return {"output": output}

save_answer is an application-defined storage function that you must provide. Kyno stores direction delivery history, not model output. Keep your outputs and supplied messages in storage appropriate for their sensitivity. Capture the state received by this node, rather than reading the binder's cache after the model call: another pull may have replaced it. Use direction_from_state() when you also need structured direction fields. For later inspection, you can use the saved delivery ID to retrieve the served direction version.

You can group delivery reads with an application-chosen correlation ID. For example, replace the binder setup above with:

binder = connection.binder(
    "customer-support",
    detail=DetailLevel.FULL,
    policy=PullPolicy(fail_closed=True),
    correlation_id="support-run-123",
)

Every pull through this binder carries the same correlation ID. It groups recorded reads; it does not identify a model call or its output. Each recorded read has its own record_id, while cached uses can share the original ID.

Optional verification

Verification belongs to your application. You choose whether to verify, which direction version to assess against, when to run the check, and what to do with its result. Kyno supplies direction and delivery history; it does not call a verifier or choose the graph's next step.

The example below illustrates one approach: keep an answer with the direction supplied to that call, then check it in another node. To try it, replace the support-answer example's State and answer definitions with these. Keep the same binder, SCENARIO, and model. The answer node now saves three things together in answer_record: the Direction it read, the direction text it sent to the model, and the answer returned by that call. The next node can review that answer without guessing which direction accompanied it.

AnswerRecord is an illustrative application type, not a Kyno API or a required storage format. review_answer is also application code:

from typing import TypedDict

from kyno.adapters.langgraph import direction_from_state
from kyno.sdk import Direction


class AnswerRecord(TypedDict):
    direction: Direction
    supplied_message: str
    output: str


class State(KynoState, total=False):
    output: str
    answer_record: AnswerRecord
    needs_review: bool


@pull_before(binder)
def answer(state):
    if state["kyno_version"] == 0 or state["kyno_binding_status"] != BindingStatus.PULLED:
        raise ValueError("A current, written constitution is required before calling the model")
    direction = direction_from_state(state)
    supplied_message = state["kyno_direction"]
    messages = [
        {"role": "system", "content": supplied_message},
        {"role": "user", "content": SCENARIO},
    ]
    output = model.invoke(messages).content
    return {
        "output": output,
        "answer_record": {
            "direction": direction,
            "supplied_message": supplied_message,
            "output": output,
        },
    }


def review_answer(state):
    record = state["answer_record"]
    needs_review = "refund has been issued" in record["output"].lower()
    return {"needs_review": needs_review}

The phrase check is only an illustration, not an alignment assessment. Your own check can use the saved direction, supplied message, and output.

Build the graph after defining those nodes. The edge from answer to review_answer tells LangGraph to run the review after the answer is ready; you do not need to call review_answer yourself:

from langgraph.graph import END, START, StateGraph

graph = (
    StateGraph(State)
    .add_node("answer", answer)
    .add_node("review_answer", review_answer)
    .add_edge(START, "answer")
    .add_edge("answer", "review_answer")
    .add_edge("review_answer", END)
    .compile()
)
result = graph.invoke({})

graph.invoke({}) runs the sequence: pull direction, generate the answer, review it, then finish. Read result["output"] for the answer and result["needs_review"] for the check's result. A True result does not automatically pause, retry, or block anything. Your application decides whether any action follows. The string check calls no service, but generating the answer still calls your configured model and may incur provider charges. Keep the connection open during execution and close it afterward, as in the required integration.

For example, an answer might receive version 1, then a later node might pull version 2 before review. The graph's current kyno_ fields would describe version 2, but answer_record still holds the version 1 input and its answer. Use that record when reviewing what the original call received; do not replace it with the latest direction.

This illustrative graph keeps one answer record and runs sequentially. For parallel work, your application owns the association between each output and its input. A shared answer_record field cannot keep separate answers from multiple branches.

Reading transition metadata

The node and decorator copy binding.change_notes into kyno_change_notes and binding.delta into kyno_delta. These describe the versions crossed by the read that supplied the direction. A failed pull using cached direction keeps that read's metadata.

for note in state["kyno_change_notes"]:
    print(note)
for change in state["kyno_delta"]:
    print(change)

When building state with direction_update directly, supply transition metadata alongside the direction. Omitted metadata produces empty lists:

from kyno.adapters.langgraph import direction_update

binding = binder.bind_with_status()
update = direction_update(
    binding.direction,
    status=binding.status,
    recording=binding.recording,
    change_notes=binding.change_notes,
    delta=binding.delta,
)