diff --git a/examples/01_standalone_sdk/27_observability_laminar.py b/examples/01_standalone_sdk/27_observability_laminar.py index 883682254f1..48c71a97243 100644 --- a/examples/01_standalone_sdk/27_observability_laminar.py +++ b/examples/01_standalone_sdk/27_observability_laminar.py @@ -3,6 +3,10 @@ This example demonstrates enabling OpenTelemetry tracing with Laminar in the OpenHands SDK. Set LMNR_PROJECT_API_KEY and run the script to see traces. + +Decision models are traced the same way: each DecisionModel.decide() call +emits a ``decision.`` span with its questions, answers, tokens and cost. +See 60_decision_model.py and 61_decision_model_agent_gate.py. """ import os diff --git a/examples/01_standalone_sdk/60_decision_model.py b/examples/01_standalone_sdk/60_decision_model.py new file mode 100644 index 00000000000..30daf738e3f --- /dev/null +++ b/examples/01_standalone_sdk/60_decision_model.py @@ -0,0 +1,187 @@ +"""Ask a decision model typed questions with DecisionModel. + +An LLM writes text. A decision model answers typed questions about a ``state`` +and returns values your code can branch on: + +- ``Noul``: a yes/no question. The answer is the probability of yes. +- ``Choice``: one option from a closed set, with a probability per option. +- ``Score``: a position on ordered levels, with a probability per level. + +Choice and Score answers also carry a ``confidence`` from 0 to 1. + +This example triages one GitHub issue with TypeSafe's Jev model and follows the +practices in https://docs.typesafe.ai/concepts/how-to-build-with-system-one.md: + +1. Keep control flow in code. Ask the model only for narrow judgments. +2. Send structured state, and point questions at fields with backticked paths. + Options and levels can be structured too, which helps tell apart options + that are easy to confuse. +3. Ask many atomic questions in one call, including speculative ones that only + matter on some code paths. Questions run in parallel against the same state, + so each extra question adds little latency and costs only its own tokens. + Through LiteLLM a call takes up to 128 questions, and a Choice up to 255 + options. More raises LLMBadRequestError before anything is sent, so split + them across calls that send the same state. +4. Combine the answers in code, and route on uncertainty instead of guessing. + +Set JEV_API_KEY to a TypeSafe API key. DECISION_MODEL defaults to +typesafe/jev-latest; any LiteLLM decision model works. + +With Laminar or OTEL configured (see 27_observability_laminar.py), each +decide() call emits a ``decision.`` span with the questions, answers, +tokens and cost. The state is not recorded on the span. +""" + +import os + +from openhands.sdk import DecisionModel +from openhands.sdk.llm import Choice, Noul, Score + + +api_key = os.getenv("JEV_API_KEY") +assert api_key, "Set JEV_API_KEY to a TypeSafe API key." + +decision_model = DecisionModel( + model=os.getenv("DECISION_MODEL", "typesafe/jev-latest"), + api_key=api_key, +) + +# Send only what the questions need, as structured JSON. +state = { + "issue": { + "title": "Agent fails when the workspace path contains a space", + "body": ( + "Running the hello world example with workspace='/tmp/my project' " + "raises FileNotFoundError on the first terminal command. Paths " + "without spaces work.\n" + "Steps: 1) mkdir '/tmp/my project' 2) run 01_hello_world.py with " + "that workspace.\n" + "This blocks everyone on our team, please fix it ASAP!" + ), + }, +} + +# One request, many questions. Question names are only for your code: they are +# not sent to the model, so each `instructions` must be a complete question. +questions = { + # Bug reports and feature requests are easy to confuse, so each option + # says what it covers, what it does not, and gives examples. Using the same + # keys for every option lets the model compare them directly. A plain + # string is enough for an option that is not easily confused. + "kind": Choice( + instructions="What does `issue` mainly ask for?", + criteria={ + "bug": { + "what": "Documented or expected behavior that is broken.", + "not_for": "Asking for behavior the software never had.", + "examples": ["Crashes on startup", "Export writes an empty file"], + }, + "feature": { + "what": "A new capability or a change to working behavior.", + "not_for": "Something that is supposed to work but fails.", + "examples": ["Support YAML config", "Add a --quiet flag"], + }, + "question": { + "what": "Help using the software; nothing is broken.", + "not_for": "Reports of failures or requests for changes.", + "examples": ["How do I set the model?", "Where are logs kept?"], + }, + "other": "None of the above.", + }, + ), + # Speculative: severity and repro steps only matter if this is a bug. + "severity": Score( + instructions="How badly does the problem in `issue` affect users?", + criteria=[ + "Cosmetic; nothing is broken.", + "A feature is broken, but a workaround exists.", + "Work is blocked and there is no workaround.", + ], + ), + "has_repro_steps": Noul( + instructions="Does `issue.body` give steps to reproduce the problem?", + ), + "is_security": Noul( + instructions=( + "Does `issue` report a security vulnerability, such as leaked " + "secrets or unauthorized access?" + ), + ), + "well_scoped": Noul( + instructions=( + "Is `issue` specific enough that a developer could start work " + "without asking follow-up questions?" + ), + ), + "frustration": Score( + instructions="How frustrated does the author of `issue` sound?", + criteria=[ + "Calm and matter-of-fact.", + "Annoyed but constructive.", + "Angry or threatening to leave.", + ], + ), +} + +result = decision_model.decide(state=state, questions=questions) + +# Typed accessors raise if the name is missing or the answer is another type. +kind = result.choice("kind") +severity = result.score("severity") +frustration = result.score("frustration") +has_repro_steps = result.noul("has_repro_steps").probability +is_security = result.noul("is_security").probability +well_scoped = result.noul("well_scoped").probability + +print(f"kind: {kind.choice}, confidence {kind.confidence:.2f}") +print(f" {kind.probabilities}") +# A Score can land between levels; level 0 is the first entry in `criteria`. +print(f"severity: {severity.score:.2f}, confidence {severity.confidence:.2f}") +print(f" {severity.legend}") +print( + f"frustration: {frustration.score:.2f}, confidence {frustration.confidence:.2f}" +) +print(f"has_repro_steps: {has_repro_steps:.2f}") +print(f"is_security: {is_security:.2f}") +print(f"well_scoped: {well_scoped:.2f}") + +# A Noul near 0.5 means "unsure", not "medium". 0.7 and 0.3 are TypeSafe's +# suggested starting thresholds; tune them on your own data. +YES, NO = 0.7, 0.3 + +# Code owns the decision. Speculative answers that do not apply are ignored. +labels: list[str] = [] +if is_security >= YES: + route = "security team (private)" +elif kind.confidence < 0.6: + route = "human triage: unclear what the issue is about" +elif kind.choice == "bug": + labels.append("bug") + # Low confidence means the probability is spread across levels, so the + # score alone is not enough to act on. + if severity.score >= 1.5 and severity.confidence >= 0.5: + labels.append("priority:high") + if has_repro_steps <= NO: + labels.append("needs-repro") + route = "agent" if well_scoped >= YES and has_repro_steps >= YES else "backlog" +elif kind.choice == "feature": + labels.append("enhancement") + route = "agent" if well_scoped >= YES else "needs-design" +else: + route = "community support" + +if frustration.score >= 1.5: + labels.append("reply-soon") + +print(f"\nroute: {route}") +print(f"labels: {labels}") + +# Log the versioned model that answered: an alias like jev-latest moves when a +# new release ships. Pin that version if you tune thresholds against it. +print(f"\nanswered by: {result.model}") +usage = decision_model.metrics.accumulated_token_usage +assert usage is not None +print(f"tokens: {usage.prompt_tokens} in, {usage.completion_tokens} out") + +cost = decision_model.metrics.accumulated_cost +print(f"EXAMPLE_COST: {cost}") diff --git a/examples/01_standalone_sdk/61_decision_model_agent_gate.py b/examples/01_standalone_sdk/61_decision_model_agent_gate.py new file mode 100644 index 00000000000..db6e1322c9b --- /dev/null +++ b/examples/01_standalone_sdk/61_decision_model_agent_gate.py @@ -0,0 +1,207 @@ +"""Use a decision model to choose which issues an agent works on. + +A decision model call is fast and costs a small fraction of a cent, so it fits +in front of an LLM agent as a gate: triage every incoming issue, and spend +agent time only on the ones that are ready for it. + +This example: + +1. Triages a queue of issues concurrently with ``adecide()``. Each issue gets + one request, and every question about that issue goes in that request. +2. Picks issues in plain code: well scoped, small, and not a security report. +3. Hands each picked issue to an agent. +4. Checks the agent's work. Code checks what code can (did any file change?). + A second decision request judges the agent's reply. A second request is + right here because its state did not exist before the agent ran; questions + about the original state belong in the first request. + +Set JEV_API_KEY for the decision model and LLM_API_KEY for the agent. + +With Laminar or OTEL configured (see 27_observability_laminar.py), ``@observe`` +opens a ``triage_issue`` span per issue and the ``decision.`` span nests +under it. The agent's conversation is traced as its own session, as in 27. +""" + +import asyncio +import os +import tempfile +from pathlib import Path + +from openhands.sdk import LLM, Agent, Conversation, DecisionModel, Tool +from openhands.sdk.conversation import get_agent_final_response +from openhands.sdk.llm import Choice, Noul, Score +from openhands.sdk.observability import observe +from openhands.tools.file_editor import FileEditorTool +from openhands.tools.terminal import TerminalTool + + +jev_api_key = os.getenv("JEV_API_KEY") +assert jev_api_key, "Set JEV_API_KEY to a TypeSafe API key." + +decision_model = DecisionModel( + model=os.getenv("DECISION_MODEL", "typesafe/jev-latest"), + api_key=jev_api_key, +) + +ISSUES = [ + { + "number": 101, + "title": "Add a slugify helper", + "body": ( + "Please add `slugify(text: str) -> str` to a new file `slug.py`. " + "It should lowercase the text, replace runs of whitespace with a " + "single hyphen, and drop every character that is not a letter, " + "digit or hyphen. Example: 'Hello, World!' -> 'hello-world'." + ), + }, + { + "number": 102, + "title": "Things are slow", + "body": "The app feels slow sometimes. Can you make it faster?", + }, + { + "number": 103, + "title": "API key shows up in debug logs", + "body": ( + "With DEBUG=1 the full LLM API key is printed in the log line for " + "every request. Anyone with log access can read it." + ), + }, + { + "number": 104, + "title": "Move storage to Postgres", + "body": ( + "We should replace the file-based conversation store with " + "Postgres, including migrations for existing users." + ), + }, +] + +TRIAGE_QUESTIONS = { + "kind": Choice( + instructions="What does `issue` mainly ask for?", + criteria={ + "bug": "Something is broken.", + "feature": "A new capability or a change in behavior.", + "other": "Neither a bug nor a feature request.", + }, + ), + "well_scoped": Noul( + instructions=( + "Does `issue.body` say exactly what to change and what the result " + "should be, so a developer could start without follow-up questions?" + ), + ), + "size": Score( + instructions="How much code would it take to do what `issue` asks?", + criteria=[ + "A few lines in one file.", + "Changes across several files or a new module.", + "A redesign or migration across the codebase.", + ], + ), + "is_security": Noul( + instructions=( + "Does `issue` report a security problem, such as exposed secrets " + "or unauthorized access?" + ), + ), +} + +YES, NO = 0.7, 0.3 + + +# ignore_input keeps the issue text off the span, as DecisionModel does. +@observe(name="triage_issue", ignore_input=True) +async def triage(issue: dict) -> tuple[str, str]: + """Return (kind, route) for one issue.""" + result = await decision_model.adecide( + state={"issue": issue}, questions=TRIAGE_QUESTIONS + ) + kind = result.choice("kind").choice + size = result.score("size") + if result.noul("is_security").probability >= YES: + return kind, "security team" + if size.score >= 1.5: + return kind, "human planning" + if result.noul("well_scoped").probability < YES: + return kind, "needs more information" + # Agents get only small changes the model is sure about. + if size.score > 0.5 or size.confidence < 0.5: + return kind, "human planning" + return kind, "agent" + + +async def triage_queue(issues: list[dict]) -> list[tuple[str, str]]: + # One request per issue, sent concurrently. Each request already asks + # every question about its issue, so no issue needs a second round trip. + return await asyncio.gather(*(triage(issue) for issue in issues)) + + +triaged = asyncio.run(triage_queue(ISSUES)) +for issue, (kind, route) in zip(ISSUES, triaged): + print(f"#{issue['number']} [{kind}] {issue['title']!r} -> {route}") + +llm = LLM( + model=os.getenv("LLM_MODEL", "gpt-5.5"), + api_key=os.getenv("LLM_API_KEY"), + base_url=os.getenv("LLM_BASE_URL"), + usage_id="agent", +) +agent = Agent( + llm=llm, + tools=[Tool(name=TerminalTool.name), Tool(name=FileEditorTool.name)], +) + +for issue, (_, route) in zip(ISSUES, triaged): + if route != "agent": + continue + + workspace = tempfile.mkdtemp() + conversation = Conversation(agent=agent, workspace=workspace) + conversation.send_message( + f"Resolve issue #{issue['number']}: {issue['title']}\n\n{issue['body']}\n\n" + "Finish with a short summary of what you changed, and say if you left " + "anything undone." + ) + conversation.run() + reply = get_agent_final_response(conversation.state.events) + + # Use code when you can: an agent that changed nothing did not do the work. + if not any(Path(workspace).iterdir()): + print(f"#{issue['number']}: no files changed; sending to a human") + continue + + # Use the decision model for the judgment code cannot make. + review = decision_model.decide( + state={"issue": issue, "agent_reply": reply}, + questions={ + "claims_done": Noul( + instructions=( + "Does `agent_reply` say that the change requested in " + "`issue.body` was made?" + ), + ), + "left_work_undone": Noul( + instructions=( + "Does `agent_reply` say that part of the request in " + "`issue.body` was left undone or done differently?" + ), + ), + }, + ) + claims_done = review.noul("claims_done").probability + left_work_undone = review.noul("left_work_undone").probability + print( + f"#{issue['number']}: claims_done={claims_done:.2f} " + f"left_work_undone={left_work_undone:.2f}" + ) + if claims_done >= YES and left_work_undone <= NO: + print(f"#{issue['number']}: ready for review") + else: + print(f"#{issue['number']}: sending to a human") + +decision_cost = decision_model.metrics.accumulated_cost +agent_cost = llm.metrics.accumulated_cost +print(f"decision model: ${decision_cost:.6f} agent LLM: ${agent_cost:.4f}") +print(f"EXAMPLE_COST: {decision_cost + agent_cost}") diff --git a/examples/01_standalone_sdk/62_decision_model_custom_provider.py b/examples/01_standalone_sdk/62_decision_model_custom_provider.py new file mode 100644 index 00000000000..995232b5ac6 --- /dev/null +++ b/examples/01_standalone_sdk/62_decision_model_custom_provider.py @@ -0,0 +1,190 @@ +"""Reach a decision model through LiteLLM, or through your own code. + +``DecisionModel`` takes the SDK's question types (``Noul``, ``Choice``, +``Score``) and returns the SDK's answer types. Where it sends the questions is +up to you: + +1. Through LiteLLM, the default. Set ``model`` to a LiteLLM model string. With + the LiteLLM release the SDK uses (1.105), that covers TypeSafe Jev + (``typesafe/jev-latest``), OpenAI (``openai/gpt-6-luna``), OpenRouter, + Perplexity, Cloudflare Clef, Microsoft Foundry (``azure_ai/``), + Strands Decider and self-hosted vLLM (``hosted_vllm/``). Databricks, + Fireworks AI, Laya, Bespoke Nimble and ``Box`` questions need LiteLLM 1.106 + or later. See https://docs.litellm.ai/docs/decisions for the current list. + +2. Through your own code. Subclass ``DecisionModel`` and override ``send()`` + (and ``asend()`` if you call ``adecide()``). Each method makes one request + and returns a ``DecisionReply`` built from SDK types. ``decide()`` still + adds retries, error mapping, metrics and the ``decision.`` span, and + checks that every question got one answer of the right type. + +Use route 2 for a model only you can reach, such as one you host privately or +an internal service. For a generally available model, contribute a provider to +LiteLLM (https://github.com/BerriAI/litellm) instead. Every LiteLLM user gets +it, and SDK users get it through route 1 when the SDK takes that LiteLLM +release. + +This example writes route 2 for TypeSafe's HTTP API. LiteLLM already supports +TypeSafe, so the example can ask both routes the same questions and show that +the code that reads the answers does not change. +""" + +import asyncio +import os +from collections.abc import Mapping +from typing import Any + +import httpx +from pydantic import SecretStr + +from openhands.sdk import DecisionModel, DecisionResult +from openhands.sdk.llm import ( + Choice, + ChoiceAnswer, + DecisionReply, + DecisionUsage, + Noul, + NoulAnswer, + Question, + Score, + ScoreAnswer, +) +from openhands.sdk.llm.exceptions import ( + LLMRateLimitError, + LLMResponseError, + LLMServiceUnavailableError, + LLMTimeoutError, +) + + +class TypeSafeHTTPDecisionModel(DecisionModel): + """Call TypeSafe's ``/v1/systemone`` API directly, without LiteLLM.""" + + def send(self, state: Any, questions: Mapping[str, Question]) -> DecisionReply: + try: + response = httpx.post(**self._request(state, questions)) + except httpx.TimeoutException as e: + raise LLMTimeoutError(str(e)) from e + return _reply(response) + + async def asend( + self, state: Any, questions: Mapping[str, Question] + ) -> DecisionReply: + try: + async with httpx.AsyncClient() as client: + response = await client.post(**self._request(state, questions)) + except httpx.TimeoutException as e: + raise LLMTimeoutError(str(e)) from e + return _reply(response) + + def _request(self, state: Any, questions: Mapping[str, Question]) -> dict: + api_key = self.api_key + assert isinstance(api_key, SecretStr), "Set api_key." + return { + "url": f"{self.base_url or 'https://api.typesafe.ai'}/v1/systemone", + "headers": {"Authorization": f"Bearer {api_key.get_secret_value()}"}, + "json": { + "model": self.model, + "state": state, + # exclude_none drops optional fields left unset, such as + # Noul.criteria. + "questions": { + name: q.model_dump(exclude_none=True) + for name, q in questions.items() + }, + }, + "timeout": self.timeout, + } + + +def _reply(response: httpx.Response) -> DecisionReply: + # DecisionModel retries these SDK errors; any other error is raised as is. + if response.status_code == 429: + raise LLMRateLimitError(response.text) + if response.status_code >= 500: + raise LLMServiceUnavailableError(response.text) + response.raise_for_status() + body = response.json() + return DecisionReply( + answers={name: _answer(a) for name, a in body["answers"].items()}, + model=body["model"], + usage=DecisionUsage( + input_tokens=body["usage"]["input_tokens"], + output_tokens=body["usage"]["output_tokens"], + ), + # TypeSafe's response has no price, so cost stays unset. Set it if + # your provider reports one or you keep a price table. + raw_response=body, + ) + + +def _answer(answer: dict) -> NoulAnswer | ChoiceAnswer | ScoreAnswer: + match answer["type"]: + case "noul": + return NoulAnswer(probability=answer["noul"]) + case "choice": + return ChoiceAnswer( + choice=answer["choice"], + confidence=answer["confidence"], + probabilities=answer["probabilities"], + ) + case "score": + return ScoreAnswer( + score=answer["score"], + confidence=answer["confidence"], + probabilities=answer["probabilities"], + legend=answer["legend"], + ) + raise LLMResponseError(f"Unsupported answer type: {answer['type']!r}") + + +api_key = os.getenv("JEV_API_KEY") +assert api_key, "Set JEV_API_KEY to a TypeSafe API key." + +state = { + "issue": { + "title": "Export to CSV drops the header row", + "body": "Since 2.3, exported CSV files have no header row. " + "Steps: open a table, click Export, choose CSV. Expected a header row.", + } +} +questions: dict[str, Question] = { + "is_bug": Noul(instructions="Does `issue` report broken behavior?"), + "area": Choice( + instructions="Which part of the product does `issue` concern?", + criteria={ + "import_export": "Reading or writing files", + "ui": "Screens, buttons and layout", + "other": "Anything else", + }, + ), + "severity": Score( + instructions="How badly does `issue` affect the user's work?", + criteria=["Cosmetic", "Workaround available", "Work blocked"], + ), +} + + +def summarize(result: DecisionResult) -> str: + """Read the answers. This code does not know which route answered.""" + severity = result.score("severity") + return ( + f"is_bug={result.noul('is_bug').probability:.2f} " + f"area={result.choice('area').choice} " + f"severity={severity.score:.2f} (confidence {severity.confidence:.2f}) " + f"model={result.model}" + ) + + +direct = TypeSafeHTTPDecisionModel(model="jev-latest", api_key=api_key) +via_litellm = DecisionModel(model="typesafe/jev-latest", api_key=api_key) + +print("direct HTTP:", summarize(direct.decide(state, questions))) +print("direct, async:", summarize(asyncio.run(direct.adecide(state, questions)))) +print("LiteLLM:", summarize(via_litellm.decide(state, questions))) + +usage = direct.metrics.accumulated_token_usage +assert usage is not None +print(f"\ndirect route tokens over 2 calls: {usage.prompt_tokens} in") +print(f"LiteLLM route cost: ${via_litellm.metrics.accumulated_cost:.6f}") +print(f"EXAMPLE_COST: {via_litellm.metrics.accumulated_cost}") diff --git a/tests/examples/test_examples.py b/tests/examples/test_examples.py index 0800d769928..ec84b5a4a06 100644 --- a/tests/examples/test_examples.py +++ b/tests/examples/test_examples.py @@ -58,6 +58,10 @@ "examples/01_standalone_sdk/16_llm_security_analyzer.py", "examples/01_standalone_sdk/27_observability_laminar.py", "examples/01_standalone_sdk/35_subscription_login.py", + # Require JEV_API_KEY for the TypeSafe decision model, which CI does not set + "examples/01_standalone_sdk/60_decision_model.py", + "examples/01_standalone_sdk/61_decision_model_agent_gate.py", + "examples/01_standalone_sdk/62_decision_model_custom_provider.py", # Requires interactive input() which fails in CI with EOFError "examples/02_remote_agent_server/05_vscode_with_docker_sandboxed_server.py", # Requires a Kubernetes cluster with agent-sandbox and the agent-sandbox extra