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: Datagoat gets 800 ms to answer; the SDK waits 2,500 ms so a call the server already logged never times out on the client). 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   # {"fit": {...} | None, "steer": bool}

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

def write_outreach(state):
    fit = (state.get("datagoat") or {}).get("fit")
    if fit:
        ...  # new behaviour; branch on ids: any(m["id"] == "c_trial_started" for m in fit["missing"])
    else:
        ...  # exactly the old node

Add the key to your state: LangGraph drops keys it doesn't know without an error. fit is set only when Datagoat answered and the case is in the test group; when it is None, run the node exactly as before (those cases are the unchanged group of the A/B test). Never give fit or the pattern to the node Datagoat wraps. Outside LangGraph, a decorated function gains with_decision(case), which returns (result, decision).

Any function

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

Outcomes

# observed_at: when it happened, not when you send it (an outcome only counts for a decision made before it)
dg.outcomes("sdr.qualify", [{"case_id": "L-20419", "value": 1, "observed_at": "2026-10-21T14:02:00Z"}])

Trying the agent out

Set DATAGOAT_TEST=1 (or Datagoat(test=True)) for local runs, CI and checking the wiring. Every call is answered as usual, but Datagoat keeps nothing: a try-out never becomes a case's decision or teaches Datagoat anything, even with a real case id. Results aren't reported and outcomes() sends nothing. Never set it where the agent really runs.

Map your agent

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