datagoat-decide (Python)

pip install datagoat-decide     # or: uv add datagoat-decide
export DATAGOAT_API_KEY=dg_live_...

Wrap a LangGraph node

from datagoat_decide import Datagoat
dg = Datagoat()  # reads DATAGOAT_API_KEY and DATAGOAT_BASE_URL (your Datagoat deployment URL)

graph.add_node("qualify", dg.wrap("sdr.qualify", qualify_lead,
    input_key="lead", output_key="qualification", case_id="lead_id"))

• Valid answer → Datagoat's answer goes into state["qualification"], your node does not run.
• No valid answer → your fallback (set in the dashboard): your node runs as before, or the case goes to a person, or no decision (None).
• Datagoat unreachable (or past the SDK's budget) → your node runs. Datagoat reachable but its engine down or slow → the outage fallback you set (your node by default). The SDK never blocks your agent past its budget (timeout, default 2.5 s). Every call carries an idempotency key, so a call that timed out after Datagoat logged it is replayed off your agent's path and still counted.

When a person decides

If the decision's fallback is Send to a person (or its outage fallback is, during an outage), a case Datagoat has no answer for goes to the dashboard Queue, and the node writes None. Keep the decision id and read the person's answer once it's in:

graph.add_node("qualify", dg.wrap("sdr.qualify", qualify_lead,
    input_key="lead", output_key="qualification", case_id="lead_id",
    on_queued=lambda d, state: save_for_later(state["lead"]["lead_id"], d.decision_id)))

# later, in a worker
s = dg.get_decision(decision_id)            # {"status": "pending" | "resolved" | "final", "answer": {...}}
answer = dg.wait_for_human(decision_id, timeout=600)   # polls; None if nobody answered in time (or the id is unknown)

answer has the same fields as Datagoat's answer (flag, label, choice) plus what the person picked (value) and their note. Async: aget_decision, await_for_human. @dg.point takes on_queued too. With a LangGraph checkpointer you can route a None to a node that calls interrupt(), and resume the run with Command(resume=answer) once the answer is in. The person's answer also goes onto the case, so training sees what was decided; it never counts as your AI's.

Using the decision in later nodes

When you change the steps after a decision (see the handoff on the decision's pattern), later nodes read the decision from the graph state:

class State(TypedDict, total=False):
    lead: dict
    qualification: dict
    datagoat: dict   # {"steered": bool, "reasons": [str]}

graph.add_node("qualify", dg.wrap("sdr.qualify", qualify_lead,
    input_key="lead", output_key="qualification", case_id="lead_id", decision_key="datagoat"))

Declare the key in your state: LangGraph drops undeclared keys without an error. Keep the old behaviour when steered is false (the unchanged cases of the A/B test), and never give the reasons to the node Datagoat wraps. Outside LangGraph, @dg.point(..., on_decision=fn) calls fn(decision, case).

Any function

@dg.point("cs.risk_tier", case_id=lambda acct: acct["account_id"])
def risk_tier(acct: dict) -> dict: ...

Outcomes

dg.outcomes("sdr.qualify", [{"case_id": "L-20419", "value": 1}])

Map your agent

datagoat-decide map my_agent.graph:graph --name "SDR agent"