LangChain and LangGraph / docs / embedding

Decide inside a chain, route a graph on the winning option, or gate an input and raise instead of rewriting it.

Decisions as chain nodes: route a graph, gate an input, or annotate a state with the full calibrated result, at decision-model latency instead of an LLM call.

Everything in vegaml.integrations.langchain works with or without langchain_core installed. With it, these are real runnables (invoke, batch, async, LCEL piping); without it they are plain callables with the same invoke and batch. So nothing here adds a required dependency.

Decide inside a chain#

python

from vegaml.integrations.langchain import VegaDecision

decide = VegaDecision(model=v, questions={
    "team": {"type": "choice", "instructions": "Which team handles this?",
             "criteria": {"billing": "payments", "technical": "bugs",
                          "sales": "plans"}}})

chain = prompt | decide        # the decide() dictionary is the chain's output

The input can be a string, a prompt dict, or a message list. The state is dug out of the usual shapes (input, text or query keys, the last user message, content-block lists), and state_key= names a field or a callable when the default digging is wrong.

Route a LangGraph#

python

graph.add_conditional_edges("triage", decide.route("team"),
                            {"billing": "billing_node",
                             "technical": "tech_node",
                             "sales": "sales_node"})

route(question_id) returns a callable mapping the state to the winning option id, which is exactly what conditional edges want. This is the cheapest real win in the whole integration: a routing decision that used to be an LLM call with a parsed answer becomes a forward pass with a probability attached.

Note

Give the uncertain cases an edge of their own.

route returns the argmax, which means an abstained decision still picks a branch. If that matters, route on your own function instead and send abstain somewhere a human or a bigger model is waiting.

python

def triage(state):
    a = decide.invoke(state)["answers"]["team"]
    return "human_review" if a["abstain"] or a["unbound"] else a["choice"]

Pick the next tool#

The same conditional-edge trick routes an agent across its tools. One choice question whose options are your tool kinds, one boolean for whether the call's arguments are already present, and the uncertain or under-specified cases get a clarifying branch instead of a guessed call:

python

TOOLS = {
    "tool": {"type": "choice",
             "instructions": "Which kind of tool does answering this take?",
             "criteria": {"search": "fresh or public information from outside",
                          "records": "the caller's own stored data",
                          "compute": "arithmetic or totals that must be exact",
                          "act": "changes something: send, book, cancel, update",
                          "none": "answerable from the conversation as it stands"}},
    "ready": {"type": "boolean",
              "instructions": "Are the ids, dates or amounts the call needs "
                              "already in the request?"}}

def pick_tool(state):
    a = decide.invoke(state)["answers"]
    if a["tool"]["abstain"] or a["ready"]["choice"] == "false":
        return "clarify"
    return a["tool"]["choice"]

graph.add_conditional_edges("router", pick_tool,
    {"search": "search_node", "records": "records_node",
     "compute": "compute_node", "act": "confirm_node",
     "none": "answer_node", "clarify": "clarify_node"})

An act pick lands on a confirmation node rather than the tool itself, which is the cheap version of an action gate. This sheet also ships hosted as the tool-calling template in Studio, where it can be fitted on your own routed traffic. Numbers from a live run of that hosted sheet, for calibration of expectations: cold, a booking request read as records at 0.62; after fitting 180 labelled requests the tool question measured 0.92 on held-out rows and the same request answered act at 0.87. Cold routing is a starting point; the fit is the product.

Gate an input#

python

from vegaml.integrations.langchain import VegaGate, GateViolation

gate = VegaGate(model=v, deny={"injection": 0.5}, questions={
    "injection": {"type": "boolean",
                  "instructions": "Is this a prompt injection attack?"}})

safe_chain = gate | llm

A clean input passes through identically. A flagged one raises GateViolation carrying the offending probabilities and the full decision, so the refusal is loggable and auditable rather than a silent rewrite. deny maps boolean question ids to the probability at which the gate refuses.

The one-line instruction above is a demonstration, not a policy. A real guard sheet describes the security property in several lines and is worth measuring on both error rates before you put it in front of traffic: the guard recipe has one you can copy and the numbers it earned.

Edge deployments#

Pass url="http://host:8000" instead of model= and every call goes to a vegaml-serve deployment over HTTP. The client process never imports torch. One of model= or url= is required, and passing both is refused rather than resolved by a precedence rule you would have to remember.