Integrations

LangGraph

A one-node LangGraph graph. The judge node POSTs /v1/judge and stores the JSON on verdict.

All integrations

"""LangGraph node that POSTs /v1/judge to Screwtape.ai."""
import json
import os
import urllib.request
from typing import TypedDict

from langgraph.graph import END, StateGraph

API = "https://screwtape.ai"


def screwtape_post(path: str, payload: dict) -> dict:
    """POST JSON to Screwtape.ai. Requires SCREWTAPE_API_KEY."""
    key = os.environ["SCREWTAPE_API_KEY"]
    request = urllib.request.Request(
        API + path,
        data=json.dumps(payload).encode(),
        headers={
            "Authorization": "Bearer " + key,
            "Content-Type": "application/json",
        },
        method="POST",
    )
    with urllib.request.urlopen(request) as response:
        body = json.load(response)
        if not isinstance(body, dict):
            raise RuntimeError("expected a JSON object")
        return body


JUDGE = {
    "task": "State the crew size and launch site.",
    "agent_output": "The crew size was 4. The mission launched from Cape Canaveral.",
    "evidence": [
        "The mission launched on 2024-03-01 from Cape Canaveral.",
        "The crew size was 4."
    ],
    "criteria": [
        "crew size",
        "Cape Canaveral"
    ]
}

CONSENSUS = {
    "task": "State the crew size.",
    "answers": [
        {
            "agent_id": "alpha",
            "output": "The crew size was 4."
        },
        {
            "agent_id": "beta",
            "output": "The crew size was 4."
        },
        {
            "agent_id": "gamma",
            "output": "The crew size was 4."
        }
    ]
}

class State(TypedDict, total=False):
    task: str
    agent_output: str
    evidence: list[str]
    criteria: list[str]
    verdict: dict


def judge(state: State) -> dict:
    payload: dict = {"task": state["task"], "agent_output": state["agent_output"]}
    if "evidence" in state:
        payload["evidence"] = state["evidence"]
    if "criteria" in state:
        payload["criteria"] = state["criteria"]
    return {"verdict": screwtape_post("/v1/judge", payload)}


graph = StateGraph(State)
graph.add_node("judge", judge)
graph.set_entry_point("judge")
graph.add_edge("judge", END)
app = graph.compile()


if __name__ == "__main__":
    print(app.invoke({
        "task": JUDGE["task"],
        "agent_output": JUDGE["agent_output"],
        "evidence": JUDGE["evidence"],
        "criteria": JUDGE["criteria"],
    }))
    print(screwtape_post("/v1/consensus", CONSENSUS))