From de8a4f6fb8b6aebf2d1d84d9c8af62b99f39bc93 Mon Sep 17 00:00:00 2001 From: nicholascole Date: Wed, 23 Sep 2026 21:30:09 -0700 Subject: [PATCH 1/7] feat(agents): add server-backed DecisionModelTool --- docs/agents/concepts/decision-models.md | 61 ++++++++++++ src/conductor/ai/agents/__init__.py | 3 + src/conductor/ai/agents/decision.py | 110 ++++++++++++++++++++++ tests/unit/ai/test_decision_model_tool.py | 57 +++++++++++ 4 files changed, 231 insertions(+) create mode 100644 docs/agents/concepts/decision-models.md create mode 100644 src/conductor/ai/agents/decision.py create mode 100644 tests/unit/ai/test_decision_model_tool.py diff --git a/docs/agents/concepts/decision-models.md b/docs/agents/concepts/decision-models.md new file mode 100644 index 00000000..e87cc70f --- /dev/null +++ b/docs/agents/concepts/decision-models.md @@ -0,0 +1,61 @@ +# Server decision tools + +`DecisionModelTool` exposes Conductor's `DECISION_MODEL` task to an agent. The +server owns provider credentials, inference, response validation, and usage +reporting. The SDK sends tool configuration and starts no Python tool worker. +This requires a server that supports the `decision_model` tool type. + +```python +from conductor.ai.agents import Agent, AgentRuntime, DecisionModelTool, agent + +class Support: + @agent( + model="openai/gpt-4o-mini", + tools=[DecisionModelTool( + name="choose_team", + description="Choose the team responsible for a support issue.", + provider="jev", + model="jev-1.13", + questions={"team": { + "type": "choice", + "instructions": "Which team should handle the issue?", + "choices": {"billing": "Payment issues", "technical": "Software issues"}, + }}, + max_calls=1, + )], + ) + def assistant(self): + """Call choose_team with the supplied state and report its decision.""" + +with AgentRuntime() as runtime: + result = runtime.run(Agent.from_instance(Support(), "assistant"), + "My invoice was charged twice.", timeout=120) +``` + +Select an orchestration model configured on your server. It handles conversation +and tool calling; the decision model evaluates state. Both can incur inference +costs. The Jev provider and its API key must be configured on the server, not in +the agent's environment, prompt, or tool arguments. + +Provider and model are fixed tool configuration. With fixed `questions`, only +`state` is exposed as a tool argument. Omitting `questions` exposes a typed +questions schema too. Generated arguments cannot override the configured provider, +model, or fixed questions. + +Questions have `type` and `instructions`: + +| Type | Additional fields | Answer | +| --- | --- | --- | +| `choice` | `choices`: 2–255 named options with descriptions | `choice`: one of the option names | +| `score` | `scale`: 2–10 ordered descriptions | `score`: number from 0 to scale length minus one | +| `boolean` | None | `probability`: number from 0 to 1 | + +The server returns `model`, `answers`, `usage`, `latencyMs`, and optional +`requestId`. Usage fields are `inputTokens`, `outputTokens`, and optional `cost` +and `currency`. Unknown costs are omitted, not reported as zero. Confidence is +optional per answer. Read the decision task's output for exact values; the agent's +final conversational summary is model-generated. + +The compiled tool task has zero retries by default. A caller explicitly rerunning +a workflow can still trigger another inference request. No Jev-specific HTTP or +authentication code belongs in the SDK. diff --git a/src/conductor/ai/agents/__init__.py b/src/conductor/ai/agents/__init__.py index fce5b308..ad053116 100644 --- a/src/conductor/ai/agents/__init__.py +++ b/src/conductor/ai/agents/__init__.py @@ -51,6 +51,8 @@ def get_weather(city: str) -> str: ServerlessCodeExecutor, ) +from conductor.ai.agents.decision import DecisionModelTool + # Exceptions from conductor.ai.agents.exceptions import ( AgentAPIError, @@ -232,6 +234,7 @@ def resolve_credentials(task: object, names: list) -> dict: from conductor.ai.agents.tracing import is_tracing_enabled __all__ = [ + "DecisionModelTool", # OpenAI Agents SDK compatibility "Runner", "RunResult", diff --git a/src/conductor/ai/agents/decision.py b/src/conductor/ai/agents/decision.py new file mode 100644 index 00000000..b5f36f64 --- /dev/null +++ b/src/conductor/ai/agents/decision.py @@ -0,0 +1,110 @@ +"""Server-executed decision tools. Provider transport and credentials stay on Conductor.""" + +from __future__ import annotations + +from copy import deepcopy +from typing import Any, Dict, Optional + +from conductor.ai.agents.tool import ToolDef + + +class DecisionModelTool(ToolDef): + """Expose a server decision provider as an agent tool without a Python worker. + + ``provider`` and ``model`` are fixed configuration, not model-generated tool + arguments. When ``questions`` is supplied, only ``state`` is exposed to the + orchestration model. Otherwise the caller supplies both state and questions. + + Question types are ``choice`` (named ``choices``), ``score`` (ordered + ``scale``), and ``boolean`` (a probability). Each has ``instructions``. + Jev-specific wire formats are translated by the server implementation. + + Requires a Conductor server supporting the ``DECISION_MODEL`` task and + ``decision_model`` tool type. This does not register provider credentials. + """ + + def __init__( + self, + name: str, + description: str, + *, + provider: str, + model: str, + questions: Optional[Dict[str, Any]] = None, + max_calls: Optional[int] = None, + ) -> None: + if not isinstance(provider, str) or not provider.strip(): + raise ValueError("provider must be a nonempty server provider name") + if not isinstance(model, str) or not model.strip(): + raise ValueError("model must be nonempty") + if questions is not None and (not isinstance(questions, dict) or not questions): + raise ValueError("questions must be a nonempty mapping when supplied") + config: Dict[str, Any] = {"provider": provider, "model": model} + properties: Dict[str, Any] = { + "state": {"type": "string", "description": "Observed state to evaluate."}, + } + required = ["state"] + if questions is not None: + config["questions"] = deepcopy(questions) + else: + properties["questions"] = { + "type": "object", + "minProperties": 1, + "additionalProperties": { + "oneOf": [ + _question_schema( + "choice", + "choices", + { + "type": "object", + "minProperties": 2, + "maxProperties": 255, + "additionalProperties": {"type": "string", "minLength": 1}, + }, + ), + _question_schema( + "score", + "scale", + { + "type": "array", + "minItems": 2, + "maxItems": 10, + "items": {"type": "string", "minLength": 1}, + }, + ), + _question_schema("boolean"), + ], + }, + } + required.append("questions") + super().__init__( + name=name, + description=description, + input_schema={ + "type": "object", + "properties": properties, + "required": required, + "additionalProperties": False, + }, + tool_type="decision_model", + config=config, + max_calls=max_calls, + retry_count=0, + ) + + +def _question_schema(kind: str, field: str = "", schema: Optional[dict] = None) -> dict: + properties = { + "type": {"type": "string", "enum": [kind]}, + "instructions": {"type": "string", "minLength": 1}, + } + required = ["type", "instructions"] + if field: + properties[field] = schema + required.append(field) + return { + "type": "object", + "properties": properties, + "required": required, + "additionalProperties": False, + } diff --git a/tests/unit/ai/test_decision_model_tool.py b/tests/unit/ai/test_decision_model_tool.py new file mode 100644 index 00000000..5a0957a9 --- /dev/null +++ b/tests/unit/ai/test_decision_model_tool.py @@ -0,0 +1,57 @@ +from conductor.ai.agents import Agent, DecisionModelTool, agent +from conductor.ai.agents.config_serializer import AgentConfigSerializer + + +def test_fixed_questions_are_configuration_and_require_no_worker(): + questions = { + "team": { + "type": "choice", + "instructions": "Select team", + "choices": {"billing": "Payments", "support": "Technical help"}, + } + } + tool = DecisionModelTool( + "decide", "Decide team", provider="jev", model="jev-1.13", questions=questions, max_calls=1 + ) + assert tool.func is None + assert tool.credentials == [] + questions["team"]["instructions"] = "mutated" + + class Example: + @agent(model="openai/configured-model", tools=[tool]) + def assistant(self): + """Use the decision tool.""" + + definition = Agent.from_instance(Example(), "assistant") + data = AgentConfigSerializer().serialize(definition)["tools"][0] + assert data["toolType"] == "decision_model" + assert data["config"]["provider"] == "jev" + assert data["config"]["questions"]["team"]["instructions"] == "Select team" + assert set(data["inputSchema"]["properties"]) == {"state"} + assert data["maxCalls"] == 1 + assert "credentials" not in data["config"] + + +def test_dynamic_questions_use_provider_neutral_schema(): + tool = DecisionModelTool("decide", "Evaluate questions", provider="another", model="v1") + assert tool.input_schema["required"] == ["state", "questions"] + variants = tool.input_schema["properties"]["questions"]["additionalProperties"]["oneOf"] + assert [v["properties"]["type"]["enum"] for v in variants] == [ + ["choice"], + ["score"], + ["boolean"], + ] + assert "provider" not in tool.input_schema["properties"] + assert "model" not in tool.input_schema["properties"] + assert tool.config == {"provider": "another", "model": "v1"} + + +def test_invalid_config_fails_locally(): + import pytest + + with pytest.raises(ValueError): + DecisionModelTool("decide", "Test", provider="", model="v1") + with pytest.raises(ValueError): + DecisionModelTool("decide", "Test", provider="jev", model="") + with pytest.raises(ValueError): + DecisionModelTool("decide", "Test", provider="jev", model="v1", questions={}) From a19a581820f8a99cc1f7c0d51f1686da4ff6f49c Mon Sep 17 00:00:00 2001 From: nicholascole Date: Wed, 23 Sep 2026 21:36:44 -0700 Subject: [PATCH 2/7] Trim decision tool documentation and tidy imports --- docs/agents/concepts/decision-models.md | 65 +++++++---------------- src/conductor/ai/agents/__init__.py | 1 - src/conductor/ai/agents/decision.py | 14 ++--- tests/unit/ai/test_decision_model_tool.py | 4 +- 4 files changed, 23 insertions(+), 61 deletions(-) diff --git a/docs/agents/concepts/decision-models.md b/docs/agents/concepts/decision-models.md index e87cc70f..ac06c328 100644 --- a/docs/agents/concepts/decision-models.md +++ b/docs/agents/concepts/decision-models.md @@ -1,61 +1,32 @@ -# Server decision tools +# Decision tools -`DecisionModelTool` exposes Conductor's `DECISION_MODEL` task to an agent. The -server owns provider credentials, inference, response validation, and usage -reporting. The SDK sends tool configuration and starts no Python tool worker. -This requires a server that supports the `decision_model` tool type. +`DecisionModelTool` calls the server's `DECISION_MODEL` task. Configure the +provider credential on Conductor; no Python worker is needed. ```python from conductor.ai.agents import Agent, AgentRuntime, DecisionModelTool, agent class Support: - @agent( - model="openai/gpt-4o-mini", - tools=[DecisionModelTool( - name="choose_team", - description="Choose the team responsible for a support issue.", - provider="jev", - model="jev-1.13", - questions={"team": { - "type": "choice", - "instructions": "Which team should handle the issue?", - "choices": {"billing": "Payment issues", "technical": "Software issues"}, - }}, - max_calls=1, - )], - ) + @agent(model="openai/gpt-4o-mini", tools=[DecisionModelTool( + "choose_team", "Choose the responsible support team.", + provider="jev", model="jev-1.13", + questions={"team": { + "type": "choice", "instructions": "Which team should handle this?", + "choices": {"billing": "Payment issues", "technical": "Software issues"}, + }}, + )]) def assistant(self): - """Call choose_team with the supplied state and report its decision.""" + """Call choose_team and report its decision.""" with AgentRuntime() as runtime: result = runtime.run(Agent.from_instance(Support(), "assistant"), "My invoice was charged twice.", timeout=120) ``` -Select an orchestration model configured on your server. It handles conversation -and tool calling; the decision model evaluates state. Both can incur inference -costs. The Jev provider and its API key must be configured on the server, not in -the agent's environment, prompt, or tool arguments. +Provider, model, and supplied questions are fixed configuration. Omitting +`questions` lets the agent supply them. Types: `choice` with named `choices`, +`score` with an ordered `scale`, and `boolean` returning a probability. -Provider and model are fixed tool configuration. With fixed `questions`, only -`state` is exposed as a tool argument. Omitting `questions` exposes a typed -questions schema too. Generated arguments cannot override the configured provider, -model, or fixed questions. - -Questions have `type` and `instructions`: - -| Type | Additional fields | Answer | -| --- | --- | --- | -| `choice` | `choices`: 2–255 named options with descriptions | `choice`: one of the option names | -| `score` | `scale`: 2–10 ordered descriptions | `score`: number from 0 to scale length minus one | -| `boolean` | None | `probability`: number from 0 to 1 | - -The server returns `model`, `answers`, `usage`, `latencyMs`, and optional -`requestId`. Usage fields are `inputTokens`, `outputTokens`, and optional `cost` -and `currency`. Unknown costs are omitted, not reported as zero. Confidence is -optional per answer. Read the decision task's output for exact values; the agent's -final conversational summary is model-generated. - -The compiled tool task has zero retries by default. A caller explicitly rerunning -a workflow can still trigger another inference request. No Jev-specific HTTP or -authentication code belongs in the SDK. +Requires a server with [decision-model support](https://github.com/conductor-oss/conductor/pull/1664). +The chat model handles orchestration; Jev evaluates state. Both incur inference +costs. Read the decision task's output for exact answers and usage. diff --git a/src/conductor/ai/agents/__init__.py b/src/conductor/ai/agents/__init__.py index ad053116..7ceb42bf 100644 --- a/src/conductor/ai/agents/__init__.py +++ b/src/conductor/ai/agents/__init__.py @@ -50,7 +50,6 @@ def get_weather(city: str) -> str: LocalCodeExecutor, ServerlessCodeExecutor, ) - from conductor.ai.agents.decision import DecisionModelTool # Exceptions diff --git a/src/conductor/ai/agents/decision.py b/src/conductor/ai/agents/decision.py index b5f36f64..10c2ca42 100644 --- a/src/conductor/ai/agents/decision.py +++ b/src/conductor/ai/agents/decision.py @@ -9,18 +9,10 @@ class DecisionModelTool(ToolDef): - """Expose a server decision provider as an agent tool without a Python worker. + """A server decision tool with fixed provider and model. - ``provider`` and ``model`` are fixed configuration, not model-generated tool - arguments. When ``questions`` is supplied, only ``state`` is exposed to the - orchestration model. Otherwise the caller supplies both state and questions. - - Question types are ``choice`` (named ``choices``), ``score`` (ordered - ``scale``), and ``boolean`` (a probability). Each has ``instructions``. - Jev-specific wire formats are translated by the server implementation. - - Requires a Conductor server supporting the ``DECISION_MODEL`` task and - ``decision_model`` tool type. This does not register provider credentials. + Supply ``questions`` to expose only ``state`` as a tool argument. Otherwise + the agent supplies both. Requires server support for ``DECISION_MODEL``. """ def __init__( diff --git a/tests/unit/ai/test_decision_model_tool.py b/tests/unit/ai/test_decision_model_tool.py index 5a0957a9..4570ada3 100644 --- a/tests/unit/ai/test_decision_model_tool.py +++ b/tests/unit/ai/test_decision_model_tool.py @@ -1,3 +1,5 @@ +import pytest + from conductor.ai.agents import Agent, DecisionModelTool, agent from conductor.ai.agents.config_serializer import AgentConfigSerializer @@ -47,8 +49,6 @@ def test_dynamic_questions_use_provider_neutral_schema(): def test_invalid_config_fails_locally(): - import pytest - with pytest.raises(ValueError): DecisionModelTool("decide", "Test", provider="", model="v1") with pytest.raises(ValueError): From aaf0cf8f2b3c898fca1eecab055ea9bbd1af0897 Mon Sep 17 00:00:00 2001 From: nicholascole Date: Wed, 23 Sep 2026 21:47:59 -0700 Subject: [PATCH 3/7] Remove duplicate decision model documentation --- docs/agents/concepts/decision-models.md | 32 ------------------------- 1 file changed, 32 deletions(-) delete mode 100644 docs/agents/concepts/decision-models.md diff --git a/docs/agents/concepts/decision-models.md b/docs/agents/concepts/decision-models.md deleted file mode 100644 index ac06c328..00000000 --- a/docs/agents/concepts/decision-models.md +++ /dev/null @@ -1,32 +0,0 @@ -# Decision tools - -`DecisionModelTool` calls the server's `DECISION_MODEL` task. Configure the -provider credential on Conductor; no Python worker is needed. - -```python -from conductor.ai.agents import Agent, AgentRuntime, DecisionModelTool, agent - -class Support: - @agent(model="openai/gpt-4o-mini", tools=[DecisionModelTool( - "choose_team", "Choose the responsible support team.", - provider="jev", model="jev-1.13", - questions={"team": { - "type": "choice", "instructions": "Which team should handle this?", - "choices": {"billing": "Payment issues", "technical": "Software issues"}, - }}, - )]) - def assistant(self): - """Call choose_team and report its decision.""" - -with AgentRuntime() as runtime: - result = runtime.run(Agent.from_instance(Support(), "assistant"), - "My invoice was charged twice.", timeout=120) -``` - -Provider, model, and supplied questions are fixed configuration. Omitting -`questions` lets the agent supply them. Types: `choice` with named `choices`, -`score` with an ordered `scale`, and `boolean` returning a probability. - -Requires a server with [decision-model support](https://github.com/conductor-oss/conductor/pull/1664). -The chat model handles orchestration; Jev evaluates state. Both incur inference -costs. Read the decision task's output for exact answers and usage. From 7e908e918b2fcd0f8b8c944284d1b0ea3e9c3f78 Mon Sep 17 00:00:00 2001 From: nicholascole Date: Fri, 25 Sep 2026 12:25:25 -0700 Subject: [PATCH 4/7] feat(agents): replace decision model API with JevAgent --- docs/agents/reference/agent-definition.md | 24 +++ docs/agents/reference/agent-schema.json | 28 +++ examples/agents/README.md | 17 ++ examples/agents/jev_agent.py | 63 ++++++ src/conductor/ai/agents/__init__.py | 16 +- src/conductor/ai/agents/agent.py | 41 +++- src/conductor/ai/agents/config_serializer.py | 22 +- src/conductor/ai/agents/decision.py | 102 --------- src/conductor/ai/agents/jev.py | 143 +++++++++++++ src/conductor/ai/agents/result.py | 26 ++- src/conductor/ai/agents/runtime/runtime.py | 76 ++++++- tests/unit/ai/test_agent_schema_contract.py | 19 ++ tests/unit/ai/test_decision_model_tool.py | 57 ----- tests/unit/ai/test_jev_agent.py | 206 +++++++++++++++++++ 14 files changed, 666 insertions(+), 174 deletions(-) create mode 100644 examples/agents/jev_agent.py delete mode 100644 src/conductor/ai/agents/decision.py create mode 100644 src/conductor/ai/agents/jev.py delete mode 100644 tests/unit/ai/test_decision_model_tool.py create mode 100644 tests/unit/ai/test_jev_agent.py diff --git a/docs/agents/reference/agent-definition.md b/docs/agents/reference/agent-definition.md index fe630aa9..01ad2ade 100644 --- a/docs/agents/reference/agent-definition.md +++ b/docs/agents/reference/agent-definition.md @@ -10,3 +10,27 @@ Names must match `^[a-zA-Z_][a-zA-Z0-9_-]*$`. Empty models represent inherited o external-agent behavior. The complete constructor and serialization semantics are maintained in [api-reference.md](../api-reference.md) and `AgentConfigSerializer`; use those sources when adding a newly supported field. + +## Jev agents + +Use `JevAgent(name="jev_support_agent", model="jev-1.13", questions=questions)` +or `AgentDef(name="jev_support_agent", kind="jev", model="jev-1.13", questions=questions)`. +Both serialize through `AgentConfigSerializer` with `kind: "jev"` and use the +standard agent runtime APIs. Jev credentials and provider HTTP calls stay on +Conductor. A chat model and Python worker are not required. + +Each question requires `instructions`. Use `ChoiceQuestion(instructions, choices)` +for a choices map, `ScoreQuestion(instructions, scale)` for an ordered scale, or +`BooleanQuestion(instructions)` for a probability response. Plain dictionaries +with `type` and the corresponding fields are also accepted. Omit definition +questions to supply them in `context={"questions": questions}` for each run. + +`runtime.plan(agent, prompt, context=context)` calls `/agent/compile` without +inference. `runtime.start(agent, prompt, context=context)` calls `/agent/start`; +`handle.join()` polls status until complete. Check `result.is_success` and report +`result.error` on failure. `result.output["result"]` retains the structured +`model`, `answers`, `usage`, `latencyMs`, and optional `requestId` fields. + +See [the runnable Jev example](../../../examples/agents/jev_agent.py), which +compiles by default and requires `--run` to start inference. Jev is supported +only as an agent definition; there is no public Jev task or decision-model tool. diff --git a/docs/agents/reference/agent-schema.json b/docs/agents/reference/agent-schema.json index 12656b47..55d63f9a 100644 --- a/docs/agents/reference/agent-schema.json +++ b/docs/agents/reference/agent-schema.json @@ -34,12 +34,40 @@ }, "additionalProperties": true }, + "jevQuestion": { + "type": "object", + "required": ["type", "instructions"], + "properties": { + "type": { "enum": ["choice", "score", "boolean"] }, + "instructions": { "type": "string", "pattern": "\\S" }, + "choices": { + "type": "object", "minProperties": 2, "maxProperties": 255, + "additionalProperties": { "type": "string", "minLength": 1 } + }, + "scale": { + "type": "array", "minItems": 2, "maxItems": 10, + "items": { "type": "string", "minLength": 1 } + } + }, + "oneOf": [ + { "properties": { "type": { "const": "choice" } }, "required": ["choices"], "not": { "required": ["scale"] } }, + { "properties": { "type": { "const": "score" } }, "required": ["scale"], "not": { "required": ["choices"] } }, + { "properties": { "type": { "const": "boolean" } }, "not": { "anyOf": [{ "required": ["choices"] }, { "required": ["scale"] }] } } + ], + "additionalProperties": false + }, "agentConfig": { "type": "object", "required": ["name"], "properties": { "name": { "type": "string", "pattern": "^[a-zA-Z_][a-zA-Z0-9_-]*$" }, "model": { "type": ["string", "null"] }, + "kind": { "const": "jev" }, + "questions": { + "type": "object", + "minProperties": 1, + "additionalProperties": { "$ref": "#/$defs/jevQuestion" } + }, "baseUrl": { "type": ["string", "null"] }, "strategy": { "type": ["string", "null"] }, "maxTurns": { "type": ["integer", "null"], "minimum": 0 }, diff --git a/examples/agents/README.md b/examples/agents/README.md index 41bae16e..3091d0ef 100644 --- a/examples/agents/README.md +++ b/examples/agents/README.md @@ -26,3 +26,20 @@ the Conductor CLI manages the server with `conductor server start`. Framework-specific examples are in [ADK](adk/README.md), [LangGraph](langgraph/README.md), and [OpenAI Agents SDK](openai/README.md). Review tool side effects before using real credentials. + +## Jev agents + +`python examples/agents/jev_agent.py` (from the repository root) compiles the +`jev_support_agent` definition without inference. Pass `--run` to explicitly start +it and poll for completion. The example uses `JevAgent` and `AgentRuntime`, requires +no chat model or Python worker, and prints structured `output.result` data: +`model`, `answers`, `usage`, `latencyMs`, and optional `requestId`. +Configure Jev credentials only on Conductor; the SDK calls the agent APIs. + +The equivalent generic definition is +`AgentDef(name="jev_support_agent", kind="jev", model="jev-1.13", questions=questions)`. +Both forms support `runtime.plan()`, `runtime.start()`, and deployment. Questions +require `instructions`: `ChoiceQuestion` uses a choices map, `ScoreQuestion` uses +an ordered scale, and `BooleanQuestion` returns a probability. If questions are +omitted from the definition, pass `context={"questions": questions}` to `plan()` +or `start()`. There is no public Jev tool or standalone decision-model API. diff --git a/examples/agents/jev_agent.py b/examples/agents/jev_agent.py new file mode 100644 index 00000000..0eedf0dc --- /dev/null +++ b/examples/agents/jev_agent.py @@ -0,0 +1,63 @@ +"""Compile a Jev agent; pass --run to explicitly start inference. + + python examples/agents/jev_agent.py + python examples/agents/jev_agent.py --run + +Requires Conductor at http://localhost:8080/api (override CONDUCTOR_SERVER_URL). +Jev credentials and provider HTTP calls belong on the server. No chat model or +Python worker is required. View executions in the UI at http://localhost:1234. +""" + +import argparse +import json +import os + +from conductor.ai.agents import AgentRuntime, ChoiceQuestion, JevAgent +from conductor.client.configuration.configuration import Configuration + +PROMPT = "The customer reports a duplicate charge on the latest invoice." + + +def support_agent(): + return JevAgent( + name="jev_support_agent", + model="jev-1.13", + questions={ + "department": ChoiceQuestion( + instructions="Which team should handle this issue?", + choices={ + "billing": "Payment and invoice issues", + "technical": "Bugs and software issues", + "other": "Other requests", + }, + ) + }, + ) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--run", action="store_true", help="Start live Jev inference") + args = parser.parse_args() + config = Configuration( + server_api_url=os.environ.get("CONDUCTOR_SERVER_URL", "http://localhost:8080/api") + ) + with AgentRuntime(config) as runtime: + agent = support_agent() + if not args.run: + # POST /agent/compile: compilation only, with the same definition and prompt. + print(json.dumps(runtime.plan(agent, PROMPT), indent=2)) + return + + # POST /agent/start, then poll GET /agent/{executionId}/status until isComplete. + handle = runtime.start(agent, PROMPT) + print("Execution:", handle.execution_id) + result = handle.join(timeout=120) + if not result.is_success: + raise RuntimeError(f"{result.status}: {result.error}") + # Keep output.result as structured data, including answers and provider metrics. + print(json.dumps(result.output["result"], indent=2)) + + +if __name__ == "__main__": + main() diff --git a/src/conductor/ai/agents/__init__.py b/src/conductor/ai/agents/__init__.py index 7ceb42bf..c5ab0ac1 100644 --- a/src/conductor/ai/agents/__init__.py +++ b/src/conductor/ai/agents/__init__.py @@ -50,7 +50,14 @@ def get_weather(city: str) -> str: LocalCodeExecutor, ServerlessCodeExecutor, ) -from conductor.ai.agents.decision import DecisionModelTool +from conductor.ai.agents.jev import ( + JevAgent, + ChoiceQuestion, + ScoreQuestion, + BooleanQuestion, + JevAnswer, + JevResult, +) # Exceptions from conductor.ai.agents.exceptions import ( @@ -233,7 +240,12 @@ def resolve_credentials(task: object, names: list) -> dict: from conductor.ai.agents.tracing import is_tracing_enabled __all__ = [ - "DecisionModelTool", + "JevAgent", + "ChoiceQuestion", + "ScoreQuestion", + "BooleanQuestion", + "JevAnswer", + "JevResult", # OpenAI Agents SDK compatibility "Runner", "RunResult", diff --git a/src/conductor/ai/agents/agent.py b/src/conductor/ai/agents/agent.py index b199aa11..2e147479 100644 --- a/src/conductor/ai/agents/agent.py +++ b/src/conductor/ai/agents/agent.py @@ -75,6 +75,8 @@ class AgentDef: name: Agent name (becomes the Conductor workflow name). model: LLM model in ``"provider/model"`` format. Empty string means "inherit from parent agent at resolution time". + kind: Set to "jev" for a Jev agent; omitted for chat agents. + questions: Fixed Jev questions, or omit and supply context.questions at runtime. instructions: System prompt — a string or the decorated callable. tools: List of tools for the agent. guardrails: List of guardrails for the agent. @@ -109,6 +111,8 @@ class AgentDef: credentials: List[Any] = field(default_factory=list) context_window_budget: Optional[int] = None prefill_tools: List[Any] = field(default_factory=list) + kind: Optional[str] = None + questions: Optional[Dict[str, Any]] = None # ── @agent decorator ──────────────────────────────────────────────────── @@ -207,7 +211,7 @@ def wrapper(*args: Any, **kwargs: Any) -> Any: def _resolve_agent(obj: Any, parent_model: str = "") -> "Agent": - """Convert an ``@agent``-decorated function into an :class:`Agent` instance. + """Convert an AgentDef or ``@agent``-decorated function into an Agent. If *obj* is already an :class:`Agent`, it is returned as-is. @@ -215,12 +219,35 @@ def _resolve_agent(obj: Any, parent_model: str = "") -> "Agent": *parent_model* is provided, the parent's model is inherited. Raises: - TypeError: If *obj* is not an Agent or ``@agent``-decorated function. + TypeError: If *obj* is not an Agent, AgentDef or ``@agent``-decorated function. """ if isinstance(obj, Agent): return obj - if callable(obj) and hasattr(obj, "_agent_def"): - ad: AgentDef = obj._agent_def + if isinstance(obj, AgentDef) or (callable(obj) and hasattr(obj, "_agent_def")): + ad: AgentDef = obj if isinstance(obj, AgentDef) else obj._agent_def + if ad.kind == "jev": + from conductor.ai.agents.jev import JevAgent + + if ( + ad.tools + or ad.agents + or ad.guardrails + or ad.instructions + or ad.func + or ad.local_code_execution + or ad.code_execution + or ad.cli_commands + or ad.credentials + or ad.prefill_tools + or ad.max_tokens is not None + or ad.temperature is not None + ): + raise ValueError( + "Jev AgentDef supports model, questions and metadata; chat configuration is unsupported" + ) + return JevAgent(ad.name, model=ad.model, questions=ad.questions, metadata=ad.metadata) + if ad.kind is not None: + raise ValueError(f"Unsupported agent kind: {ad.kind}") # Handle ClaudeCode: don't inherit parent model for claude-code agents if isinstance(ad.model, ClaudeCode): resolved_model = ad.model @@ -229,7 +256,7 @@ def _resolve_agent(obj: Any, parent_model: str = "") -> "Agent": return Agent( name=ad.name, model=resolved_model, - instructions=ad.func, + instructions=ad.func or ad.instructions, tools=ad.tools, guardrails=ad.guardrails, agents=ad.agents, @@ -249,7 +276,9 @@ def _resolve_agent(obj: Any, parent_model: str = "") -> "Agent": context_window_budget=ad.context_window_budget, prefill_tools=ad.prefill_tools or None, ) - raise TypeError(f"Expected an Agent or @agent-decorated function, got {type(obj).__name__}") + raise TypeError( + f"Expected an Agent, AgentDef or @agent-decorated function, got {type(obj).__name__}" + ) # ── from_instance resolution helpers ──────────────────────────────────── diff --git a/src/conductor/ai/agents/config_serializer.py b/src/conductor/ai/agents/config_serializer.py index fcc8ed67..576394a4 100644 --- a/src/conductor/ai/agents/config_serializer.py +++ b/src/conductor/ai/agents/config_serializer.py @@ -36,7 +36,27 @@ def serialize(self, agent: "Agent") -> dict: return self._serialize_agent(agent) def _serialize_agent(self, agent: "Agent") -> dict: - from conductor.ai.agents.agent import PromptTemplate + from conductor.ai.agents.agent import AgentDef, PromptTemplate, _resolve_agent + + if isinstance(agent, AgentDef): + agent = _resolve_agent(agent) + if getattr(agent, "kind", None) == "jev": + from conductor.ai.agents.jev import jev_questions + + if agent.tools or agent.agents or agent.memory or agent.guardrails or agent.output_type: + raise ValueError( + "Jev agents cannot contain chat tools, agents, memory, output schemas or guardrails" + ) + config = {"name": agent.name, "kind": "jev", "model": agent.model} + if agent.questions is not None: + config["questions"] = jev_questions(agent.questions) + if agent.timeout_seconds: + config["timeoutSeconds"] = agent.timeout_seconds + if agent.metadata: + config["metadata"] = agent.metadata + if agent.masked_fields: + config["maskedFields"] = agent.masked_fields + return config # Skill agents — emit the raw skill config so the server's # SkillNormalizer can compile sub-agents (e.g. gilfoyle, dinesh) diff --git a/src/conductor/ai/agents/decision.py b/src/conductor/ai/agents/decision.py deleted file mode 100644 index 10c2ca42..00000000 --- a/src/conductor/ai/agents/decision.py +++ /dev/null @@ -1,102 +0,0 @@ -"""Server-executed decision tools. Provider transport and credentials stay on Conductor.""" - -from __future__ import annotations - -from copy import deepcopy -from typing import Any, Dict, Optional - -from conductor.ai.agents.tool import ToolDef - - -class DecisionModelTool(ToolDef): - """A server decision tool with fixed provider and model. - - Supply ``questions`` to expose only ``state`` as a tool argument. Otherwise - the agent supplies both. Requires server support for ``DECISION_MODEL``. - """ - - def __init__( - self, - name: str, - description: str, - *, - provider: str, - model: str, - questions: Optional[Dict[str, Any]] = None, - max_calls: Optional[int] = None, - ) -> None: - if not isinstance(provider, str) or not provider.strip(): - raise ValueError("provider must be a nonempty server provider name") - if not isinstance(model, str) or not model.strip(): - raise ValueError("model must be nonempty") - if questions is not None and (not isinstance(questions, dict) or not questions): - raise ValueError("questions must be a nonempty mapping when supplied") - config: Dict[str, Any] = {"provider": provider, "model": model} - properties: Dict[str, Any] = { - "state": {"type": "string", "description": "Observed state to evaluate."}, - } - required = ["state"] - if questions is not None: - config["questions"] = deepcopy(questions) - else: - properties["questions"] = { - "type": "object", - "minProperties": 1, - "additionalProperties": { - "oneOf": [ - _question_schema( - "choice", - "choices", - { - "type": "object", - "minProperties": 2, - "maxProperties": 255, - "additionalProperties": {"type": "string", "minLength": 1}, - }, - ), - _question_schema( - "score", - "scale", - { - "type": "array", - "minItems": 2, - "maxItems": 10, - "items": {"type": "string", "minLength": 1}, - }, - ), - _question_schema("boolean"), - ], - }, - } - required.append("questions") - super().__init__( - name=name, - description=description, - input_schema={ - "type": "object", - "properties": properties, - "required": required, - "additionalProperties": False, - }, - tool_type="decision_model", - config=config, - max_calls=max_calls, - retry_count=0, - ) - - -def _question_schema(kind: str, field: str = "", schema: Optional[dict] = None) -> dict: - properties = { - "type": {"type": "string", "enum": [kind]}, - "instructions": {"type": "string", "minLength": 1}, - } - required = ["type", "instructions"] - if field: - properties[field] = schema - required.append(field) - return { - "type": "object", - "properties": properties, - "required": required, - "additionalProperties": False, - } diff --git a/src/conductor/ai/agents/jev.py b/src/conductor/ai/agents/jev.py new file mode 100644 index 00000000..f0d213e8 --- /dev/null +++ b/src/conductor/ai/agents/jev.py @@ -0,0 +1,143 @@ +"""Server-executed Jev agents. Provider transport and credentials stay on Conductor.""" + +from __future__ import annotations + +from copy import deepcopy +from dataclasses import asdict, dataclass +from typing import Any, Dict, Mapping, Optional, Sequence + +from conductor.ai.agents.agent import Agent + +_MIN_OPTIONS = 2 +_MAX_CHOICES = 255 +_MAX_SCORE_LABELS = 10 + + +@dataclass(frozen=True) +class ChoiceQuestion: + instructions: str + choices: Mapping[str, str] + type: str = "choice" + + +@dataclass(frozen=True) +class ScoreQuestion: + instructions: str + scale: Sequence[str] + type: str = "score" + + +@dataclass(frozen=True) +class BooleanQuestion: + instructions: str + type: str = "boolean" + + +def jev_questions(questions: Mapping[str, Any]) -> dict: + """Validate the typed question contract before registering or executing an agent.""" + if not isinstance(questions, Mapping) or not questions: + raise ValueError("questions must be a nonempty mapping") + result = {} + for name, question in questions.items(): + q = ( + asdict(question) + if isinstance(question, (ChoiceQuestion, ScoreQuestion, BooleanQuestion)) + else deepcopy(question) + ) + if not isinstance(name, str) or not name.strip() or not isinstance(q, dict): + raise ValueError("invalid Jev question") + kind = q.get("type") + fields = {"type", "instructions"} | ( + {"choices"} if kind == "choice" else {"scale"} if kind == "score" else set() + ) + if ( + kind not in ("choice", "score", "boolean") + or set(q) != fields + or not isinstance(q.get("instructions"), str) + or not q["instructions"].strip() + ): + raise ValueError("invalid Jev question") + if kind == "choice": + choices = q["choices"] + if ( + not isinstance(choices, dict) + or not _MIN_OPTIONS <= len(choices) <= _MAX_CHOICES + or any( + not isinstance(k, str) + or not k.strip() + or not isinstance(v, str) + or not v.strip() + for k, v in choices.items() + ) + ): + raise ValueError("choice requires 2..255 named choices") + if kind == "score": + if ( + not isinstance(q["scale"], (list, tuple)) + or not _MIN_OPTIONS <= len(q["scale"]) <= _MAX_SCORE_LABELS + or any(not isinstance(v, str) or not v.strip() for v in q["scale"]) + ): + raise ValueError("score requires 2..10 labels") + q["scale"] = list(q["scale"]) + result[name] = q + return result + + +class JevAgent(Agent): + """A Jev agent with the standard agent execution lifecycle. + + Pass state as the runtime prompt (strings or JSON-serializable mappings). + With no fixed questions, supply ``context={"questions": ...}`` when starting. + Credentials and inference stay on Conductor; no local tool workers are needed. + """ + + kind = "jev" + + def __init__( + self, + name: str, + *, + model: str, + questions: Optional[Mapping[str, Any]] = None, + timeout_seconds: int = 0, + metadata: Optional[Dict[str, Any]] = None, + masked_fields: Optional[list[str]] = None, + ): + if not isinstance(model, str) or not model.strip(): + raise ValueError("Jev model must be nonempty") + super().__init__( + name=name, + model=model, + timeout_seconds=timeout_seconds, + metadata=metadata, + masked_fields=masked_fields, + ) + self.questions = jev_questions(questions) if questions is not None else None + + +@dataclass(frozen=True) +class JevAnswer: + type: str + choice: Optional[str] = None + score: Optional[float] = None + probability: Optional[float] = None + confidence: Optional[float] = None + + +@dataclass(frozen=True) +class JevResult: + model: str + answers: Dict[str, JevAnswer] + usage: Optional[Dict[str, Any]] = None + latency_ms: Optional[int] = None + request_id: Optional[str] = None + + @classmethod + def from_dict(cls, value: dict) -> "JevResult": + return cls( + model=value["model"], + answers={k: JevAnswer(**v) for k, v in value["answers"].items()}, + usage=value.get("usage"), + latency_ms=value.get("latencyMs"), + request_id=value.get("requestId"), + ) diff --git a/src/conductor/ai/agents/result.py b/src/conductor/ai/agents/result.py index 29ccc92f..c1b831a1 100644 --- a/src/conductor/ai/agents/result.py +++ b/src/conductor/ai/agents/result.py @@ -8,7 +8,10 @@ from dataclasses import dataclass, field from enum import Enum -from typing import Any, AsyncIterator, Callable, Dict, Iterator, List, Optional +from typing import TYPE_CHECKING, Any, AsyncIterator, Callable, Dict, Iterator, List, Optional + +if TYPE_CHECKING: + from conductor.ai.agents.jev import JevResult # ── Status & FinishReason enums ──────────────────────────────────────── @@ -94,7 +97,9 @@ class AgentResult: Attributes: output: The agent's final answer as a dict. Always contains a - ``"result"`` key whose value is a string (or ``None``). + ``"result"`` key whose value is a string (or ``None``) for chat + agents, or a structured result for Jev agents. The latter + is also available through :attr:`jev`. If ``output_type`` was set on the agent, this is a validated instance of that type instead. execution_id: The Conductor execution ID (for debugging in the UI). @@ -125,6 +130,21 @@ class AgentResult: events: List["AgentEvent"] = field(default_factory=list) sub_results: Dict[str, Any] = field(default_factory=dict) + @property + def jev(self) -> Optional[JevResult]: + """Typed view of a successful Jev-shaped result; raw data stays in output.""" + from conductor.ai.agents.jev import JevResult + + if ( + self.is_success + and isinstance(self.output, dict) + and isinstance(self.output.get("result"), dict) + ): + value = self.output["result"] + if isinstance(value.get("model"), str) and isinstance(value.get("answers"), dict): + return JevResult.from_dict(value) + return None + @property def is_success(self) -> bool: """Whether the agent completed successfully.""" @@ -598,7 +618,7 @@ def _build_result(self, status: "AgentStatus") -> "AgentResult": correlation_id=self.correlation_id, status=status.status, finish_reason=self._runtime._derive_finish_reason(status.status, status.output), - error=status.reason if status.status in ("FAILED", "TERMINATED") else None, + error=status.reason if status.status in ("FAILED", "TERMINATED", "TIMED_OUT") else None, token_usage=token_usage, metadata=metadata, ) diff --git a/src/conductor/ai/agents/runtime/runtime.py b/src/conductor/ai/agents/runtime/runtime.py index e53b921a..0793c175 100644 --- a/src/conductor/ai/agents/runtime/runtime.py +++ b/src/conductor/ai/agents/runtime/runtime.py @@ -842,6 +842,14 @@ def _start_via_server( serializer = AgentConfigSerializer() config_json = serializer.serialize(agent) + if getattr(agent, "kind", None) == "jev" and agent.questions is None: + from conductor.ai.agents.jev import jev_questions + + context = { + **(context or {}), + "questions": jev_questions((context or {}).get("questions")), + } + # Per-run LLM overrides (model/temperature/…) mutate the serialized # agentConfig before compile+register+start, so they flow into the # LLM tasks without a new server field. @@ -906,6 +914,14 @@ async def _start_via_server_async( serializer = AgentConfigSerializer() config_json = serializer.serialize(agent) + if getattr(agent, "kind", None) == "jev" and agent.questions is None: + from conductor.ai.agents.jev import jev_questions + + context = { + **(context or {}), + "questions": jev_questions((context or {}).get("questions")), + } + # Per-run LLM overrides (see :meth:`_start_via_server`). rs = RunSettings.coerce(run_settings) if rs is not None: @@ -1105,6 +1121,10 @@ def prepare(self, agent: Any) -> None: handle = runtime.start(agent, prompt) """ from conductor.ai.agents.frameworks.serializer import detect_framework + from conductor.ai.agents.agent import AgentDef, _resolve_agent + + if isinstance(agent, AgentDef): + agent = _resolve_agent(agent) if isinstance(agent, str): return # nothing to prepare for run-by-name @@ -2003,6 +2023,9 @@ def _resolve_prompt(self, prompt: Any) -> str: return "" if isinstance(prompt, str): return prompt + if isinstance(prompt, (dict, list)): + # Structured agent input must be JSON, not Python's repr with single quotes. + return json.dumps(prompt, separators=(",", ":"), allow_nan=False) if not isinstance(prompt, PromptTemplate): return str(prompt) @@ -2233,7 +2256,7 @@ def _ensure_models_for_agent(self, agent: Agent) -> None: def _collect(a: Agent) -> None: if not isinstance(a, Agent): return - if a.model and a.model not in seen: + if a.model and a.model not in seen and getattr(a, "kind", None) != "jev": seen.add(a.model) for sub in a.agents: _collect(sub) @@ -2299,20 +2322,32 @@ def _has_worker_tools(self, agent: Agent) -> bool: # ── Plan (compile without executing) ──────────────────────────── - def plan(self, agent: Agent) -> Any: + def plan( + self, + agent: Agent, + prompt: Optional[str] = None, + *, + context: Optional[Dict[str, Any]] = None, + ) -> Any: """Compile an agent to a Conductor workflow definition and return it. This does NOT register, start workers, or execute. Useful for inspecting, debugging, or exporting the compiled workflow. Args: - agent: The agent to compile. + agent: The Agent or AgentDef to compile. + prompt: Optional input to include in the compilation request. + context: Optional run context, including dynamic Jev questions. Returns: The raw server response dict with ``workflowDef`` and ``requiredWorkers`` keys. """ from conductor.ai.agents.frameworks.serializer import detect_framework + from conductor.ai.agents.agent import AgentDef, _resolve_agent + + if isinstance(agent, AgentDef): + agent = _resolve_agent(agent) framework = detect_framework(agent) if framework: @@ -2330,6 +2365,14 @@ def plan(self, agent: Agent) -> Any: config_json = serializer.serialize(agent) payload = {"agentConfig": config_json} + if prompt is not None: + payload["prompt"] = self._resolve_prompt(prompt) + if context is not None: + if getattr(agent, "kind", None) == "jev" and agent.questions is None: + from conductor.ai.agents.jev import jev_questions + + context = {**context, "questions": jev_questions(context.get("questions"))} + payload["context"] = context return self._agent_client.compile_agent(payload) # ── Deploy (CI/CD) ───────────────────────────────────────────── @@ -2375,6 +2418,10 @@ def deploy( results = [] for agent in all_agents: from conductor.ai.agents.frameworks.serializer import detect_framework + from conductor.ai.agents.agent import AgentDef, _resolve_agent + + if isinstance(agent, AgentDef): + agent = _resolve_agent(agent) framework = detect_framework(agent) @@ -2413,6 +2460,10 @@ async def deploy_async( results = [] for agent in all_agents: from conductor.ai.agents.frameworks.serializer import detect_framework + from conductor.ai.agents.agent import AgentDef, _resolve_agent + + if isinstance(agent, AgentDef): + agent = _resolve_agent(agent) framework = detect_framework(agent) @@ -2545,9 +2596,12 @@ def serve( # Register local Python worker functions for each agent from conductor.ai.agents.frameworks.serializer import detect_framework + from conductor.ai.agents.agent import AgentDef, _resolve_agent has_new = False for agent in all_agents: + if isinstance(agent, AgentDef): + agent = _resolve_agent(agent) framework = detect_framework(agent) # Register the agent (workflow + task defs) on the server before # bringing up local workers. Mirrors run()'s deploy-then-register @@ -2697,6 +2751,10 @@ def run( # Check for foreign framework agent from conductor.ai.agents.frameworks.serializer import detect_framework + from conductor.ai.agents.agent import AgentDef, _resolve_agent + + if isinstance(agent, AgentDef): + agent = _resolve_agent(agent) framework = detect_framework(agent) @@ -3807,6 +3865,10 @@ def start( # Check for foreign framework agent from conductor.ai.agents.frameworks.serializer import detect_framework + from conductor.ai.agents.agent import AgentDef, _resolve_agent + + if isinstance(agent, AgentDef): + agent = _resolve_agent(agent) framework = detect_framework(agent) if framework is not None: @@ -4066,6 +4128,10 @@ async def run_async( # Foreign framework check from conductor.ai.agents.frameworks.serializer import detect_framework + from conductor.ai.agents.agent import AgentDef, _resolve_agent + + if isinstance(agent, AgentDef): + agent = _resolve_agent(agent) framework = detect_framework(agent) @@ -4237,6 +4303,10 @@ async def start_async( ) from conductor.ai.agents.frameworks.serializer import detect_framework + from conductor.ai.agents.agent import AgentDef, _resolve_agent + + if isinstance(agent, AgentDef): + agent = _resolve_agent(agent) framework = detect_framework(agent) if framework is not None: diff --git a/tests/unit/ai/test_agent_schema_contract.py b/tests/unit/ai/test_agent_schema_contract.py index eecd1db3..a2d31dc4 100644 --- a/tests/unit/ai/test_agent_schema_contract.py +++ b/tests/unit/ai/test_agent_schema_contract.py @@ -49,3 +49,22 @@ def test_agent_schema_rejects_unknown_root_fields(): schema = json.loads(SCHEMA_PATH.read_text()) with pytest.raises(jsonschema.ValidationError): jsonschema.validate({"name": "valid_name", "unknownField": True}, schema) + + +def test_agent_schema_accepts_jev_definition(): + from conductor.ai.agents import JevAgent, ChoiceQuestion, ScoreQuestion, BooleanQuestion + + definition = JevAgent( + "jev", + model="jev-1.13", + questions={ + "team": ChoiceQuestion("Choose", {"billing": "Payments", "technical": "Bugs"}), + "priority": ScoreQuestion("Score", ["low", "high"]), + "ready": BooleanQuestion("Ready?"), + }, + ) + schema = json.loads(SCHEMA_PATH.read_text()) + jsonschema.validate(AgentConfigSerializer().serialize(definition), schema) + jsonschema.validate( + AgentConfigSerializer().serialize(JevAgent("dynamic", model="jev-1.13")), schema + ) diff --git a/tests/unit/ai/test_decision_model_tool.py b/tests/unit/ai/test_decision_model_tool.py deleted file mode 100644 index 4570ada3..00000000 --- a/tests/unit/ai/test_decision_model_tool.py +++ /dev/null @@ -1,57 +0,0 @@ -import pytest - -from conductor.ai.agents import Agent, DecisionModelTool, agent -from conductor.ai.agents.config_serializer import AgentConfigSerializer - - -def test_fixed_questions_are_configuration_and_require_no_worker(): - questions = { - "team": { - "type": "choice", - "instructions": "Select team", - "choices": {"billing": "Payments", "support": "Technical help"}, - } - } - tool = DecisionModelTool( - "decide", "Decide team", provider="jev", model="jev-1.13", questions=questions, max_calls=1 - ) - assert tool.func is None - assert tool.credentials == [] - questions["team"]["instructions"] = "mutated" - - class Example: - @agent(model="openai/configured-model", tools=[tool]) - def assistant(self): - """Use the decision tool.""" - - definition = Agent.from_instance(Example(), "assistant") - data = AgentConfigSerializer().serialize(definition)["tools"][0] - assert data["toolType"] == "decision_model" - assert data["config"]["provider"] == "jev" - assert data["config"]["questions"]["team"]["instructions"] == "Select team" - assert set(data["inputSchema"]["properties"]) == {"state"} - assert data["maxCalls"] == 1 - assert "credentials" not in data["config"] - - -def test_dynamic_questions_use_provider_neutral_schema(): - tool = DecisionModelTool("decide", "Evaluate questions", provider="another", model="v1") - assert tool.input_schema["required"] == ["state", "questions"] - variants = tool.input_schema["properties"]["questions"]["additionalProperties"]["oneOf"] - assert [v["properties"]["type"]["enum"] for v in variants] == [ - ["choice"], - ["score"], - ["boolean"], - ] - assert "provider" not in tool.input_schema["properties"] - assert "model" not in tool.input_schema["properties"] - assert tool.config == {"provider": "another", "model": "v1"} - - -def test_invalid_config_fails_locally(): - with pytest.raises(ValueError): - DecisionModelTool("decide", "Test", provider="", model="v1") - with pytest.raises(ValueError): - DecisionModelTool("decide", "Test", provider="jev", model="") - with pytest.raises(ValueError): - DecisionModelTool("decide", "Test", provider="jev", model="v1", questions={}) diff --git a/tests/unit/ai/test_jev_agent.py b/tests/unit/ai/test_jev_agent.py new file mode 100644 index 00000000..87946e11 --- /dev/null +++ b/tests/unit/ai/test_jev_agent.py @@ -0,0 +1,206 @@ +import json + +import pytest + +from conductor.ai.agents import ( + Agent, + AgentResult, + BooleanQuestion, + ChoiceQuestion, + JevAgent, + ScoreQuestion, +) +from conductor.ai.agents.config_serializer import AgentConfigSerializer +from conductor.ai.agents.jev import jev_questions +from conductor.ai.agents.runtime.runtime import AgentRuntime + + +def test_jev_agent_serializes_as_an_agent_and_as_a_child(): + definition = JevAgent( + "choose", + model="jev-1.13", + questions={ + "action": ChoiceQuestion("Choose an action", {"go": "Proceed", "wait": "Wait"}), + "quality": ScoreQuestion("Evaluate quality", ["low", "high"]), + "ready": BooleanQuestion("Ready to proceed?"), + }, + ) + assert isinstance(definition, Agent) + config = AgentConfigSerializer().serialize(definition) + assert config["kind"] == "jev" + assert "decisionProvider" not in config + assert config["questions"]["action"]["choices"] == {"go": "Proceed", "wait": "Wait"} + assert "tools" not in config + assert "credentials" not in config + parent = Agent("parent", agents=[definition], strategy="sequential") + assert AgentConfigSerializer().serialize(parent)["agents"][0]["kind"] == "jev" + assert "kind" not in AgentConfigSerializer().serialize(Agent("chat", model="openai/model")) + + +def test_dynamic_questions_and_json_state(): + agent = JevAgent("choose", model="jev-1.13") + assert "questions" not in AgentConfigSerializer().serialize(agent) + runtime = AgentRuntime.__new__(AgentRuntime) + assert json.loads(runtime._resolve_prompt({"ready": True, "value": None})) == { + "ready": True, + "value": None, + } + # No attempt to register Jev models as chat providers. + runtime._ensure_models_for_agent(agent) + assert jev_questions({"ready": BooleanQuestion("Ready?")})["ready"]["type"] == "boolean" + + +@pytest.mark.parametrize( + "question", + [ + {"type": "choice", "instructions": "Choose", "choices": {"only": "Only"}}, + {"type": "score", "instructions": "Score", "scale": []}, + {"type": "boolean", "instructions": "", "choices": {}}, + ], +) +def test_invalid_question_contract_is_rejected(question): + with pytest.raises(ValueError, match=r"requires|invalid"): + JevAgent("bad", model="jev-1.13", questions={"q": question}) + + +def test_typed_result_preserves_provider_details_without_chat_wrapping(): + raw = { + "model": "jev-1.13", + "answers": { + "action": {"type": "choice", "choice": "go", "confidence": 0.9}, + "score": {"type": "score", "score": 1.2}, + "ready": {"type": "boolean", "probability": 0.8}, + }, + "usage": {"inputTokens": 10, "outputTokens": 3, "cost": 0.001, "currency": "USD"}, + "requestId": "req", + "latencyMs": 42, + } + result = AgentResult(output={"result": raw}) + assert result.jev.answers["action"].choice == "go" + assert result.jev.answers["score"].score == raw["answers"]["score"]["score"] + assert result.jev.answers["ready"].probability == raw["answers"]["ready"]["probability"] + assert result.jev.usage == raw["usage"] + assert result.jev.latency_ms == raw["latencyMs"] + assert result.jev.request_id == "req" + assert AgentResult(output={"result": "chat"}).jev is None + + +def test_failed_jev_has_no_typed_answer(): + result = AgentResult(status="FAILED", output={"agentKind": "jev", "result": {}}) + assert result.jev is None + + +@pytest.fixture +def runtime(): + from unittest.mock import patch + from conductor.ai.agents.runtime.config import AgentConfig + + with patch("conductor.client.orkes_clients.OrkesClients"): + with AgentRuntime(settings=AgentConfig(auto_start_workers=False)) as runtime: + yield runtime + + +def test_agent_def_and_convenience_class_serialize_identically(runtime): + from conductor.ai.agents import AgentDef + + questions = {"ready": BooleanQuestion("Ready?")} + definition = AgentDef(name="ready", kind="jev", model="jev-1.13", questions=questions) + expected = { + "name": "ready", + "kind": "jev", + "model": "jev-1.13", + "questions": {"ready": {"type": "boolean", "instructions": "Ready?"}}, + } + assert AgentConfigSerializer().serialize(definition) == expected + assert ( + AgentConfigSerializer().serialize(JevAgent("ready", model="jev-1.13", questions=questions)) + == expected + ) + runtime.plan(definition, "Ready to ship") + runtime._agent_client.compile_agent.assert_called_once_with( + {"agentConfig": expected, "prompt": "Ready to ship"} + ) + runtime._agent_client.start_agent.assert_not_called() + + +@pytest.mark.parametrize("status", ["COMPLETED", "FAILED", "TIMED_OUT", "TERMINATED"]) +def test_start_and_poll_preserves_structured_result_and_failure_reason(runtime, status): + from unittest.mock import patch + from conductor.ai.agents import AgentDef + + definition = AgentDef(name="ready", kind="jev", model="jev-1.13") + raw = { + "model": "jev-1.13", + "answers": {"ready": {"type": "boolean", "probability": 0.8}}, + "usage": {"inputTokens": 10}, + "latencyMs": 42, + "requestId": "request-1", + } + runtime._agent_client.start_agent.return_value = { + "executionId": "execution-1", + "requiredWorkers": [], + } + runtime._agent_client.get_status.side_effect = [ + {"isComplete": False, "status": "RUNNING"}, + { + "isComplete": True, + "status": status, + "output": {"result": raw}, + "reasonForIncompletion": "provider failure" if status != "COMPLETED" else None, + }, + ] + handle = runtime.start( + definition, "Ready?", context={"questions": {"ready": BooleanQuestion("Ready?")}} + ) + assert handle.execution_id == "execution-1" + payload = runtime._agent_client.start_agent.call_args.args[0] + assert payload["agentConfig"] == {"name": "ready", "kind": "jev", "model": "jev-1.13"} + assert payload["context"]["questions"]["ready"] == {"type": "boolean", "instructions": "Ready?"} + assert not runtime._workers_started + with patch("time.sleep"): + result = handle.join(timeout=5) + assert runtime._agent_client.get_status.call_count == 2 + assert result.output["result"] == raw + assert result.is_success == (status == "COMPLETED") + assert result.error == (None if status == "COMPLETED" else "provider failure") + + +def test_dynamic_questions_required_before_start(runtime): + with pytest.raises(ValueError, match="questions"): + runtime.start(JevAgent("ready", model="jev-1.13"), "Ready?") + runtime._agent_client.start_agent.assert_not_called() + + +def test_async_start_and_dynamic_compile(runtime): + import asyncio + from unittest.mock import AsyncMock + from conductor.ai.agents import AgentDef + + definition = AgentDef(name="ready", kind="jev", model="jev-1.13") + context = {"questions": {"ready": BooleanQuestion("Ready?")}} + runtime.plan(definition, "Ready?", context=context) + assert ( + runtime._agent_client.compile_agent.call_args.args[0]["context"]["questions"]["ready"][ + "type" + ] + == "boolean" + ) + runtime._agent_client.start_agent_async = AsyncMock( + return_value={"executionId": "async-1", "requiredWorkers": []} + ) + handle = asyncio.run(runtime.start_async(definition, "Ready?", context=context)) + assert handle.execution_id == "async-1" + assert ( + runtime._agent_client.start_agent_async.call_args.args[0]["context"]["questions"]["ready"][ + "type" + ] + == "boolean" + ) + + +def test_jev_has_no_tool_or_internal_task_exports(): + import conductor.ai.agents as agents + + assert not hasattr(agents, "DecisionModelTool") + assert not hasattr(agents, "DecisionAgent") + assert not hasattr(agents, "JEV_AGENT") From 2fd3226aeb8be842b8ad9f7c9f4f5b8b8d99c091 Mon Sep 17 00:00:00 2001 From: nicholascole Date: Fri, 25 Sep 2026 12:58:01 -0700 Subject: [PATCH 5/7] feat(agents): simplify Jev support and add routing examples --- docs/agents/reference/agent-definition.md | 33 ++--- docs/agents/reference/agent-schema.json | 28 +--- examples/agents/README.md | 23 +-- examples/agents/jev_agent.py | 24 +--- examples/agents/jev_nested_triage.py | 86 +++++++++++ examples/agents/jev_specialists.py | 122 ++++++++++++++++ examples/agents/luna_jev_triage.py | 63 ++++++++ src/conductor/ai/agents/__init__.py | 14 +- src/conductor/ai/agents/agent.py | 15 +- src/conductor/ai/agents/config_serializer.py | 8 +- src/conductor/ai/agents/jev.py | 132 +---------------- src/conductor/ai/agents/result.py | 27 +--- src/conductor/ai/agents/runtime/runtime.py | 44 ++---- tests/unit/ai/test_agent_schema_contract.py | 23 ++- tests/unit/ai/test_jev_agent.py | 144 +++++-------------- tests/unit/ai/test_tool_extraction.py | 31 ++++ 16 files changed, 405 insertions(+), 412 deletions(-) create mode 100644 examples/agents/jev_nested_triage.py create mode 100644 examples/agents/jev_specialists.py create mode 100644 examples/agents/luna_jev_triage.py diff --git a/docs/agents/reference/agent-definition.md b/docs/agents/reference/agent-definition.md index 01ad2ade..6b4a1acd 100644 --- a/docs/agents/reference/agent-definition.md +++ b/docs/agents/reference/agent-definition.md @@ -13,24 +13,21 @@ maintained in [api-reference.md](../api-reference.md) and ## Jev agents -Use `JevAgent(name="jev_support_agent", model="jev-1.13", questions=questions)` -or `AgentDef(name="jev_support_agent", kind="jev", model="jev-1.13", questions=questions)`. -Both serialize through `AgentConfigSerializer` with `kind: "jev"` and use the -standard agent runtime APIs. Jev credentials and provider HTTP calls stay on -Conductor. A chat model and Python worker are not required. +Use `JevAgent(name, model="jev-1.13", questions=questions)` or +`AgentDef(name=name, kind="jev", model="jev-1.13", questions=questions)`. +Questions are dictionaries with `instructions` and a `type`: +`choice` uses a `choices` map, `score` uses an ordered `scale`, and `boolean` +returns a probability. Omit questions to supply `context={"questions": questions}`. -Each question requires `instructions`. Use `ChoiceQuestion(instructions, choices)` -for a choices map, `ScoreQuestion(instructions, scale)` for an ordered scale, or -`BooleanQuestion(instructions)` for a probability response. Plain dictionaries -with `type` and the corresponding fields are also accepted. Omit definition -questions to supply them in `context={"questions": questions}` for each run. +Use `runtime.plan(agent, prompt)` to compile or `runtime.start(agent, prompt)` +to run. Call `handle.join()` and check `result.is_success` or `result.error`. +`result.output["result"]` preserves `model`, `answers`, `usage`, `latencyMs` +and optional `requestId`. Credentials and inference stay on Conductor. +No chat model or Python worker is needed. -`runtime.plan(agent, prompt, context=context)` calls `/agent/compile` without -inference. `runtime.start(agent, prompt, context=context)` calls `/agent/start`; -`handle.join()` polls status until complete. Check `result.is_success` and report -`result.error` on failure. `result.output["result"]` retains the structured -`model`, `answers`, `usage`, `latencyMs`, and optional `requestId` fields. +[Example](../../../examples/agents/jev_agent.py) -See [the runnable Jev example](../../../examples/agents/jev_agent.py), which -compiles by default and requires `--run` to start inference. Jev is supported -only as an agent definition; there is no public Jev task or decision-model tool. +For server-side Jev routing, use `Agent(strategy="router", router=selector, +agents=children)`. The Jev selector must have one fixed choice question whose +choice keys match the child agent names. No parent chat model is needed. +Routing runs one child and preserves its structured result. diff --git a/docs/agents/reference/agent-schema.json b/docs/agents/reference/agent-schema.json index 55d63f9a..425d9019 100644 --- a/docs/agents/reference/agent-schema.json +++ b/docs/agents/reference/agent-schema.json @@ -34,28 +34,6 @@ }, "additionalProperties": true }, - "jevQuestion": { - "type": "object", - "required": ["type", "instructions"], - "properties": { - "type": { "enum": ["choice", "score", "boolean"] }, - "instructions": { "type": "string", "pattern": "\\S" }, - "choices": { - "type": "object", "minProperties": 2, "maxProperties": 255, - "additionalProperties": { "type": "string", "minLength": 1 } - }, - "scale": { - "type": "array", "minItems": 2, "maxItems": 10, - "items": { "type": "string", "minLength": 1 } - } - }, - "oneOf": [ - { "properties": { "type": { "const": "choice" } }, "required": ["choices"], "not": { "required": ["scale"] } }, - { "properties": { "type": { "const": "score" } }, "required": ["scale"], "not": { "required": ["choices"] } }, - { "properties": { "type": { "const": "boolean" } }, "not": { "anyOf": [{ "required": ["choices"] }, { "required": ["scale"] }] } } - ], - "additionalProperties": false - }, "agentConfig": { "type": "object", "required": ["name"], @@ -63,11 +41,7 @@ "name": { "type": "string", "pattern": "^[a-zA-Z_][a-zA-Z0-9_-]*$" }, "model": { "type": ["string", "null"] }, "kind": { "const": "jev" }, - "questions": { - "type": "object", - "minProperties": 1, - "additionalProperties": { "$ref": "#/$defs/jevQuestion" } - }, + "questions": { "type": "object", "additionalProperties": { "type": "object" } }, "baseUrl": { "type": ["string", "null"] }, "strategy": { "type": ["string", "null"] }, "maxTurns": { "type": ["integer", "null"], "minimum": 0 }, diff --git a/examples/agents/README.md b/examples/agents/README.md index 3091d0ef..690f6f45 100644 --- a/examples/agents/README.md +++ b/examples/agents/README.md @@ -27,19 +27,10 @@ Framework-specific examples are in [ADK](adk/README.md), [LangGraph](langgraph/README.md), and [OpenAI Agents SDK](openai/README.md). Review tool side effects before using real credentials. -## Jev agents - -`python examples/agents/jev_agent.py` (from the repository root) compiles the -`jev_support_agent` definition without inference. Pass `--run` to explicitly start -it and poll for completion. The example uses `JevAgent` and `AgentRuntime`, requires -no chat model or Python worker, and prints structured `output.result` data: -`model`, `answers`, `usage`, `latencyMs`, and optional `requestId`. -Configure Jev credentials only on Conductor; the SDK calls the agent APIs. - -The equivalent generic definition is -`AgentDef(name="jev_support_agent", kind="jev", model="jev-1.13", questions=questions)`. -Both forms support `runtime.plan()`, `runtime.start()`, and deployment. Questions -require `instructions`: `ChoiceQuestion` uses a choices map, `ScoreQuestion` uses -an ordered scale, and `BooleanQuestion` returns a probability. If questions are -omitted from the definition, pass `context={"questions": questions}` to `plan()` -or `start()`. There is no public Jev tool or standalone decision-model API. +Jev: [jev_agent.py](jev_agent.py) compiles by default. Pass `--run` for inference. + +- `jev_nested_triage.py`: Jev department selection, Jev specialist selection, then a Jev specialist. +- `luna_jev_triage.py --model INTEGRATION/luna-6`: Luna selects one of ten Jev specialists. + +Both compile by default. Pass `--run` for inference. These require the server's +Jev router support and structured output for single-turn routers without synthesis. diff --git a/examples/agents/jev_agent.py b/examples/agents/jev_agent.py index 0eedf0dc..6923467d 100644 --- a/examples/agents/jev_agent.py +++ b/examples/agents/jev_agent.py @@ -1,18 +1,10 @@ -"""Compile a Jev agent; pass --run to explicitly start inference. - - python examples/agents/jev_agent.py - python examples/agents/jev_agent.py --run - -Requires Conductor at http://localhost:8080/api (override CONDUCTOR_SERVER_URL). -Jev credentials and provider HTTP calls belong on the server. No chat model or -Python worker is required. View executions in the UI at http://localhost:1234. -""" +"""Compile a Jev agent. Pass --run for inference. Configure credentials on Conductor.""" import argparse import json import os -from conductor.ai.agents import AgentRuntime, ChoiceQuestion, JevAgent +from conductor.ai.agents import AgentRuntime, JevAgent from conductor.client.configuration.configuration import Configuration PROMPT = "The customer reports a duplicate charge on the latest invoice." @@ -23,14 +15,15 @@ def support_agent(): name="jev_support_agent", model="jev-1.13", questions={ - "department": ChoiceQuestion( - instructions="Which team should handle this issue?", - choices={ + "department": { + "type": "choice", + "instructions": "Which team should handle this issue?", + "choices": { "billing": "Payment and invoice issues", "technical": "Bugs and software issues", "other": "Other requests", }, - ) + } }, ) @@ -45,17 +38,14 @@ def main(): with AgentRuntime(config) as runtime: agent = support_agent() if not args.run: - # POST /agent/compile: compilation only, with the same definition and prompt. print(json.dumps(runtime.plan(agent, PROMPT), indent=2)) return - # POST /agent/start, then poll GET /agent/{executionId}/status until isComplete. handle = runtime.start(agent, PROMPT) print("Execution:", handle.execution_id) result = handle.join(timeout=120) if not result.is_success: raise RuntimeError(f"{result.status}: {result.error}") - # Keep output.result as structured data, including answers and provider metrics. print(json.dumps(result.output["result"], indent=2)) diff --git a/examples/agents/jev_nested_triage.py b/examples/agents/jev_nested_triage.py new file mode 100644 index 00000000..220237f0 --- /dev/null +++ b/examples/agents/jev_nested_triage.py @@ -0,0 +1,86 @@ +"""Three levels: Jev department selector, Jev specialist selector, Jev specialist. + +Requires server support for Jev routers. Compile by default, --run for inference. +""" + +import argparse +import json +import os + +from conductor.ai.agents import Agent, AgentRuntime, JevAgent, Strategy +from conductor.client.configuration.configuration import Configuration +from jev_specialists import SPECIALTIES, specialists + +PROMPT = "Our latest invoice has two settled charges with different transaction IDs for the same purchase." + + +def routing_team(name, children, descriptions): + return Agent( + name=name, + strategy=Strategy.ROUTER, + router=JevAgent( + name=f"{name}_selector", + model="jev-1.13", + questions={ + "agent": { + "type": "choice", + "instructions": "Choose the agent best suited to handle this request.", + "choices": descriptions, + } + }, + ), + agents=children, + max_turns=1, + synthesize=False, + ) + + +def triage_agent(): + leaves = specialists() + departments = [] + for department in ("billing", "technical", "account"): + members = { + name: agent for name, agent in leaves.items() if SPECIALTIES[name][0] == department + } + departments.append( + routing_team( + f"jev_{department}_team", + list(members.values()), + {agent.name: SPECIALTIES[name][1] for name, agent in members.items()}, + ) + ) + return routing_team( + "jev_nested_triage", + departments, + { + "jev_billing_team": "Charges, refunds and subscriptions", + "jev_technical_team": "API errors, outages, integrations and setup", + "jev_account_team": "Access, security and privacy", + }, + ) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--run", action="store_true") + parser.add_argument("--prompt", default=PROMPT) + args = parser.parse_args() + config = Configuration( + server_api_url=os.getenv("CONDUCTOR_SERVER_URL", "http://localhost:8080/api") + ) + with AgentRuntime(config) as runtime: + agent = triage_agent() + if not args.run: + print(json.dumps(runtime.plan(agent, args.prompt), indent=2)) + return + runtime.deploy(agent) + handle = runtime.start(agent, args.prompt) + print("Execution:", handle.execution_id) + result = handle.join(timeout=180) + if not result.is_success: + raise RuntimeError(result.error) + print(json.dumps(result.output, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/examples/agents/jev_specialists.py b/examples/agents/jev_specialists.py new file mode 100644 index 00000000..1275163d --- /dev/null +++ b/examples/agents/jev_specialists.py @@ -0,0 +1,122 @@ +"""Ten Jev specialists used by the routing examples.""" + +from conductor.ai.agents import JevAgent + +SPECIALTIES = { + "duplicate_charge": ( + "billing", + "Duplicate payments or repeated invoice charges", + { + "check_transactions": "Compare charge identifiers and payment status", + "refund_duplicate": "Refund a confirmed duplicate settled charge", + "explain_hold": "Explain a pending authorization hold", + }, + ), + "refund": ( + "billing", + "Refund eligibility and refund progress", + { + "review_eligibility": "Check purchase date and refund policy", + "approve_refund": "Approve a refund supported by the supplied policy and evidence", + "trace_refund": "Trace an already approved refund", + }, + ), + "subscription": ( + "billing", + "Subscription renewals, cancellation and plan changes", + { + "review_renewal": "Check renewal timing and notice", + "cancel_renewal": "Schedule cancellation at the end of the current term", + "change_plan": "Review the requested plan change", + }, + ), + "api_errors": ( + "technical", + "API request failures, rate limits and invalid payloads", + { + "inspect_request": "Collect request identifiers and validate the payload", + "backoff": "Apply retry backoff for documented rate limiting", + "escalate_bug": "Escalate a reproducible server defect", + }, + ), + "outage": ( + "technical", + "Service outages and widespread availability failures", + { + "check_scope": "Verify the affected services and customers", + "open_incident": "Open an incident for confirmed widespread impact", + "update_incident": "Attach evidence to an existing incident", + }, + ), + "integration": ( + "technical", + "Webhook delivery and third-party integration failures", + { + "inspect_delivery": "Inspect delivery logs and endpoint responses", + "repair_config": "Correct an identified integration configuration error", + "replay_delivery": "Replay a failed delivery after checking idempotency", + }, + ), + "onboarding": ( + "technical", + "Initial setup, configuration and first successful request", + { + "collect_requirements": "Clarify the intended setup and environment", + "provide_setup": "Provide the relevant setup steps", + "validate_setup": "Verify the completed configuration", + }, + ), + "access": ( + "account", + "Login problems, account recovery and permissions", + { + "verify_identity": "Verify identity before changing account access", + "recover_account": "Use the approved account recovery process", + "review_permissions": "Review the required role and authorization", + }, + ), + "security": ( + "account", + "Suspected compromise, leaked credentials and abuse", + { + "investigate": "Collect evidence and assess the scope of compromise", + "contain": "Recommend revoking confirmed compromised credentials", + "escalate_security": "Escalate an active compromise to security response", + }, + ), + "privacy": ( + "account", + "Personal data access, export and deletion requests", + { + "verify_request": "Verify the requester and scope of the request", + "export_data": "Route a verified data export request", + "delete_data": "Route a verified deletion request through policy review", + }, + ), +} + + +def specialists(): + return { + name: JevAgent( + name=f"jev_{name}", + model="jev-1.13", + questions={ + "action": { + "type": "choice", + "instructions": f"Recommend the next step for {description}. Use only supplied evidence. Do not claim any action was executed.", + "choices": actions, + }, + "urgency": { + "type": "score", + "instructions": "Rate urgency from low to critical based on customer impact.", + "scale": ["low", "normal", "high", "critical"], + }, + "needs_more_information": { + "type": "boolean", + "instructions": "Is more information needed before taking the recommended action?", + }, + }, + ) + for name, (_, description, actions) in SPECIALTIES.items() + } diff --git a/examples/agents/luna_jev_triage.py b/examples/agents/luna_jev_triage.py new file mode 100644 index 00000000..16f63da2 --- /dev/null +++ b/examples/agents/luna_jev_triage.py @@ -0,0 +1,63 @@ +"""Luna-6 selects one of ten Jev specialists using server-side routing. + +Pass --model INTEGRATION/luna-6. Compile by default, --run for inference. +""" + +import argparse +import json +import os + +from conductor.ai.agents import Agent, AgentRuntime, Strategy +from conductor.client.configuration.configuration import Configuration +from jev_specialists import SPECIALTIES, specialists + + +def triage_agent(model): + candidates = specialists() + descriptions = "\n".join( + f"{agent.name}: {SPECIALTIES[name][1]}" for name, agent in candidates.items() + ) + return Agent( + name="luna_jev_triage", + model=model, + strategy=Strategy.ROUTER, + router=Agent( + name="luna_jev_selector", + model=model, + instructions=f"Select exactly one specialist for the request. Return only its agent name.\n{descriptions}", + ), + agents=list(candidates.values()), + max_turns=1, + synthesize=False, + ) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--model", default=os.getenv("CONDUCTOR_AGENT_LLM_MODEL")) + parser.add_argument("--run", action="store_true") + parser.add_argument( + "--prompt", default="Our webhook endpoint returns 503 and order notifications are missing." + ) + args = parser.parse_args() + if not args.model: + parser.error("Pass --model with your server's integration/luna-6 identifier") + config = Configuration( + server_api_url=os.getenv("CONDUCTOR_SERVER_URL", "http://localhost:8080/api") + ) + with AgentRuntime(config) as runtime: + agent = triage_agent(args.model) + if not args.run: + print(json.dumps(runtime.plan(agent, args.prompt), indent=2)) + return + runtime.deploy(agent) + handle = runtime.start(agent, args.prompt) + print("Execution:", handle.execution_id) + result = handle.join(timeout=180) + if not result.is_success: + raise RuntimeError(result.error) + print(json.dumps(result.output, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/src/conductor/ai/agents/__init__.py b/src/conductor/ai/agents/__init__.py index c5ab0ac1..b42b7934 100644 --- a/src/conductor/ai/agents/__init__.py +++ b/src/conductor/ai/agents/__init__.py @@ -50,14 +50,7 @@ def get_weather(city: str) -> str: LocalCodeExecutor, ServerlessCodeExecutor, ) -from conductor.ai.agents.jev import ( - JevAgent, - ChoiceQuestion, - ScoreQuestion, - BooleanQuestion, - JevAnswer, - JevResult, -) +from conductor.ai.agents.jev import JevAgent # Exceptions from conductor.ai.agents.exceptions import ( @@ -241,11 +234,6 @@ def resolve_credentials(task: object, names: list) -> dict: __all__ = [ "JevAgent", - "ChoiceQuestion", - "ScoreQuestion", - "BooleanQuestion", - "JevAnswer", - "JevResult", # OpenAI Agents SDK compatibility "Runner", "RunResult", diff --git a/src/conductor/ai/agents/agent.py b/src/conductor/ai/agents/agent.py index 2e147479..583b11bc 100644 --- a/src/conductor/ai/agents/agent.py +++ b/src/conductor/ai/agents/agent.py @@ -75,7 +75,7 @@ class AgentDef: name: Agent name (becomes the Conductor workflow name). model: LLM model in ``"provider/model"`` format. Empty string means "inherit from parent agent at resolution time". - kind: Set to "jev" for a Jev agent; omitted for chat agents. + kind: Set to "jev" for a Jev agent. Omit for chat agents. questions: Fixed Jev questions, or omit and supply context.questions at runtime. instructions: System prompt — a string or the decorated callable. tools: List of tools for the agent. @@ -242,9 +242,7 @@ def _resolve_agent(obj: Any, parent_model: str = "") -> "Agent": or ad.max_tokens is not None or ad.temperature is not None ): - raise ValueError( - "Jev AgentDef supports model, questions and metadata; chat configuration is unsupported" - ) + raise ValueError("Jev AgentDef does not support chat configuration") return JevAgent(ad.name, model=ad.model, questions=ad.questions, metadata=ad.metadata) if ad.kind is not None: raise ValueError(f"Unsupported agent kind: {ad.kind}") @@ -942,11 +940,12 @@ def is_claude_code(self) -> bool: def external(self) -> bool: """``True`` if this agent references an external workflow (no local definition). - An agent with no ``model`` is treated as external — the server - produces a ``SubWorkflowTask`` referencing the workflow by name - instead of compiling the agent inline. + An agent with no model references an existing workflow, except a + ROUTER with a Jev selector, which is compiled locally by the server. """ - return not self.model + return not self.model and not ( + self.strategy == Strategy.ROUTER and getattr(self.router, "kind", None) == "jev" + ) # ── Instance-method resolution ────────────────────────────────────── diff --git a/src/conductor/ai/agents/config_serializer.py b/src/conductor/ai/agents/config_serializer.py index 576394a4..9800bb6e 100644 --- a/src/conductor/ai/agents/config_serializer.py +++ b/src/conductor/ai/agents/config_serializer.py @@ -41,21 +41,15 @@ def _serialize_agent(self, agent: "Agent") -> dict: if isinstance(agent, AgentDef): agent = _resolve_agent(agent) if getattr(agent, "kind", None) == "jev": - from conductor.ai.agents.jev import jev_questions - if agent.tools or agent.agents or agent.memory or agent.guardrails or agent.output_type: raise ValueError( "Jev agents cannot contain chat tools, agents, memory, output schemas or guardrails" ) config = {"name": agent.name, "kind": "jev", "model": agent.model} if agent.questions is not None: - config["questions"] = jev_questions(agent.questions) - if agent.timeout_seconds: - config["timeoutSeconds"] = agent.timeout_seconds + config["questions"] = agent.questions if agent.metadata: config["metadata"] = agent.metadata - if agent.masked_fields: - config["maskedFields"] = agent.masked_fields return config # Skill agents — emit the raw skill config so the server's diff --git a/src/conductor/ai/agents/jev.py b/src/conductor/ai/agents/jev.py index f0d213e8..f9d3e70f 100644 --- a/src/conductor/ai/agents/jev.py +++ b/src/conductor/ai/agents/jev.py @@ -1,95 +1,13 @@ -"""Server-executed Jev agents. Provider transport and credentials stay on Conductor.""" - -from __future__ import annotations +"""Jev agent definitions. Inference and credentials stay on Conductor.""" from copy import deepcopy -from dataclasses import asdict, dataclass -from typing import Any, Dict, Mapping, Optional, Sequence +from typing import Any, Dict, Optional from conductor.ai.agents.agent import Agent -_MIN_OPTIONS = 2 -_MAX_CHOICES = 255 -_MAX_SCORE_LABELS = 10 - - -@dataclass(frozen=True) -class ChoiceQuestion: - instructions: str - choices: Mapping[str, str] - type: str = "choice" - - -@dataclass(frozen=True) -class ScoreQuestion: - instructions: str - scale: Sequence[str] - type: str = "score" - - -@dataclass(frozen=True) -class BooleanQuestion: - instructions: str - type: str = "boolean" - - -def jev_questions(questions: Mapping[str, Any]) -> dict: - """Validate the typed question contract before registering or executing an agent.""" - if not isinstance(questions, Mapping) or not questions: - raise ValueError("questions must be a nonempty mapping") - result = {} - for name, question in questions.items(): - q = ( - asdict(question) - if isinstance(question, (ChoiceQuestion, ScoreQuestion, BooleanQuestion)) - else deepcopy(question) - ) - if not isinstance(name, str) or not name.strip() or not isinstance(q, dict): - raise ValueError("invalid Jev question") - kind = q.get("type") - fields = {"type", "instructions"} | ( - {"choices"} if kind == "choice" else {"scale"} if kind == "score" else set() - ) - if ( - kind not in ("choice", "score", "boolean") - or set(q) != fields - or not isinstance(q.get("instructions"), str) - or not q["instructions"].strip() - ): - raise ValueError("invalid Jev question") - if kind == "choice": - choices = q["choices"] - if ( - not isinstance(choices, dict) - or not _MIN_OPTIONS <= len(choices) <= _MAX_CHOICES - or any( - not isinstance(k, str) - or not k.strip() - or not isinstance(v, str) - or not v.strip() - for k, v in choices.items() - ) - ): - raise ValueError("choice requires 2..255 named choices") - if kind == "score": - if ( - not isinstance(q["scale"], (list, tuple)) - or not _MIN_OPTIONS <= len(q["scale"]) <= _MAX_SCORE_LABELS - or any(not isinstance(v, str) or not v.strip() for v in q["scale"]) - ): - raise ValueError("score requires 2..10 labels") - q["scale"] = list(q["scale"]) - result[name] = q - return result - class JevAgent(Agent): - """A Jev agent with the standard agent execution lifecycle. - - Pass state as the runtime prompt (strings or JSON-serializable mappings). - With no fixed questions, supply ``context={"questions": ...}`` when starting. - Credentials and inference stay on Conductor; no local tool workers are needed. - """ + """Supply questions here or through context.questions at runtime.""" kind = "jev" @@ -98,46 +16,8 @@ def __init__( name: str, *, model: str, - questions: Optional[Mapping[str, Any]] = None, - timeout_seconds: int = 0, + questions: Optional[Dict[str, Any]] = None, metadata: Optional[Dict[str, Any]] = None, - masked_fields: Optional[list[str]] = None, ): - if not isinstance(model, str) or not model.strip(): - raise ValueError("Jev model must be nonempty") - super().__init__( - name=name, - model=model, - timeout_seconds=timeout_seconds, - metadata=metadata, - masked_fields=masked_fields, - ) - self.questions = jev_questions(questions) if questions is not None else None - - -@dataclass(frozen=True) -class JevAnswer: - type: str - choice: Optional[str] = None - score: Optional[float] = None - probability: Optional[float] = None - confidence: Optional[float] = None - - -@dataclass(frozen=True) -class JevResult: - model: str - answers: Dict[str, JevAnswer] - usage: Optional[Dict[str, Any]] = None - latency_ms: Optional[int] = None - request_id: Optional[str] = None - - @classmethod - def from_dict(cls, value: dict) -> "JevResult": - return cls( - model=value["model"], - answers={k: JevAnswer(**v) for k, v in value["answers"].items()}, - usage=value.get("usage"), - latency_ms=value.get("latencyMs"), - request_id=value.get("requestId"), - ) + super().__init__(name=name, model=model, metadata=metadata) + self.questions = deepcopy(questions) diff --git a/src/conductor/ai/agents/result.py b/src/conductor/ai/agents/result.py index c1b831a1..41e8b335 100644 --- a/src/conductor/ai/agents/result.py +++ b/src/conductor/ai/agents/result.py @@ -8,10 +8,7 @@ from dataclasses import dataclass, field from enum import Enum -from typing import TYPE_CHECKING, Any, AsyncIterator, Callable, Dict, Iterator, List, Optional - -if TYPE_CHECKING: - from conductor.ai.agents.jev import JevResult +from typing import Any, AsyncIterator, Callable, Dict, Iterator, List, Optional # ── Status & FinishReason enums ──────────────────────────────────────── @@ -97,9 +94,7 @@ class AgentResult: Attributes: output: The agent's final answer as a dict. Always contains a - ``"result"`` key whose value is a string (or ``None``) for chat - agents, or a structured result for Jev agents. The latter - is also available through :attr:`jev`. + ``"result"`` key with text, structured data or ``None``. If ``output_type`` was set on the agent, this is a validated instance of that type instead. execution_id: The Conductor execution ID (for debugging in the UI). @@ -130,21 +125,6 @@ class AgentResult: events: List["AgentEvent"] = field(default_factory=list) sub_results: Dict[str, Any] = field(default_factory=dict) - @property - def jev(self) -> Optional[JevResult]: - """Typed view of a successful Jev-shaped result; raw data stays in output.""" - from conductor.ai.agents.jev import JevResult - - if ( - self.is_success - and isinstance(self.output, dict) - and isinstance(self.output.get("result"), dict) - ): - value = self.output["result"] - if isinstance(value.get("model"), str) and isinstance(value.get("answers"), dict): - return JevResult.from_dict(value) - return None - @property def is_success(self) -> bool: """Whether the agent completed successfully.""" @@ -731,6 +711,7 @@ class EventType(str, Enum): """Types of events emitted during agent execution.""" THINKING = "thinking" + JEV = "jev" TOOL_CALL = "tool_call" TOOL_RESULT = "tool_result" HANDOFF = "handoff" @@ -752,7 +733,7 @@ class AgentEvent: ``guardrail_pass``, ``guardrail_fail``). tool_name: Tool name (for ``tool_call``, ``tool_result``). args: Tool call arguments (for ``tool_call``). - result: Tool result (for ``tool_result``). + result: Structured result for ``tool_result`` or ``jev``. target: Target agent name (for ``handoff``). output: Final output (for ``done``). execution_id: The Conductor execution ID. diff --git a/src/conductor/ai/agents/runtime/runtime.py b/src/conductor/ai/agents/runtime/runtime.py index 0793c175..ad51a00e 100644 --- a/src/conductor/ai/agents/runtime/runtime.py +++ b/src/conductor/ai/agents/runtime/runtime.py @@ -23,7 +23,7 @@ from conductor.ai.agents.runtime.config import AgentConfig from conductor.client.configuration.configuration import Configuration -from conductor.ai.agents.agent import Agent +from conductor.ai.agents.agent import Agent, AgentDef, _resolve_agent from conductor.ai.agents.run_settings import RunSettings from conductor.ai.agents.result import ( AgentEvent, @@ -292,6 +292,7 @@ def _normalize_handoff_target(task_ref: str) -> str: _NON_TOOL_TASK_TYPES = frozenset( { "LLM_CHAT_COMPLETE", + "JEV_AGENT", "SWITCH", "DO_WHILE", "INLINE", @@ -440,6 +441,15 @@ def _task_events(task: Any, execution_id: str) -> Iterator[AgentEvent]: task_status = str(getattr(task, "status", "") or "").upper() output_data = getattr(task, "output_data", None) or {} + if task_type == "JEV_AGENT" and task_status == "COMPLETED": + yield AgentEvent( + type=EventType.JEV, + content=task_ref, + result=output_data, + execution_id=execution_id, + ) + return + # LLM task -> THINKING if "LLM_CHAT_COMPLETE" in task_type: yield AgentEvent( @@ -842,14 +852,6 @@ def _start_via_server( serializer = AgentConfigSerializer() config_json = serializer.serialize(agent) - if getattr(agent, "kind", None) == "jev" and agent.questions is None: - from conductor.ai.agents.jev import jev_questions - - context = { - **(context or {}), - "questions": jev_questions((context or {}).get("questions")), - } - # Per-run LLM overrides (model/temperature/…) mutate the serialized # agentConfig before compile+register+start, so they flow into the # LLM tasks without a new server field. @@ -914,14 +916,6 @@ async def _start_via_server_async( serializer = AgentConfigSerializer() config_json = serializer.serialize(agent) - if getattr(agent, "kind", None) == "jev" and agent.questions is None: - from conductor.ai.agents.jev import jev_questions - - context = { - **(context or {}), - "questions": jev_questions((context or {}).get("questions")), - } - # Per-run LLM overrides (see :meth:`_start_via_server`). rs = RunSettings.coerce(run_settings) if rs is not None: @@ -1121,7 +1115,6 @@ def prepare(self, agent: Any) -> None: handle = runtime.start(agent, prompt) """ from conductor.ai.agents.frameworks.serializer import detect_framework - from conductor.ai.agents.agent import AgentDef, _resolve_agent if isinstance(agent, AgentDef): agent = _resolve_agent(agent) @@ -2023,9 +2016,6 @@ def _resolve_prompt(self, prompt: Any) -> str: return "" if isinstance(prompt, str): return prompt - if isinstance(prompt, (dict, list)): - # Structured agent input must be JSON, not Python's repr with single quotes. - return json.dumps(prompt, separators=(",", ":"), allow_nan=False) if not isinstance(prompt, PromptTemplate): return str(prompt) @@ -2344,7 +2334,6 @@ def plan( ``requiredWorkers`` keys. """ from conductor.ai.agents.frameworks.serializer import detect_framework - from conductor.ai.agents.agent import AgentDef, _resolve_agent if isinstance(agent, AgentDef): agent = _resolve_agent(agent) @@ -2368,10 +2357,6 @@ def plan( if prompt is not None: payload["prompt"] = self._resolve_prompt(prompt) if context is not None: - if getattr(agent, "kind", None) == "jev" and agent.questions is None: - from conductor.ai.agents.jev import jev_questions - - context = {**context, "questions": jev_questions(context.get("questions"))} payload["context"] = context return self._agent_client.compile_agent(payload) @@ -2418,7 +2403,6 @@ def deploy( results = [] for agent in all_agents: from conductor.ai.agents.frameworks.serializer import detect_framework - from conductor.ai.agents.agent import AgentDef, _resolve_agent if isinstance(agent, AgentDef): agent = _resolve_agent(agent) @@ -2460,7 +2444,6 @@ async def deploy_async( results = [] for agent in all_agents: from conductor.ai.agents.frameworks.serializer import detect_framework - from conductor.ai.agents.agent import AgentDef, _resolve_agent if isinstance(agent, AgentDef): agent = _resolve_agent(agent) @@ -2596,7 +2579,6 @@ def serve( # Register local Python worker functions for each agent from conductor.ai.agents.frameworks.serializer import detect_framework - from conductor.ai.agents.agent import AgentDef, _resolve_agent has_new = False for agent in all_agents: @@ -2751,7 +2733,6 @@ def run( # Check for foreign framework agent from conductor.ai.agents.frameworks.serializer import detect_framework - from conductor.ai.agents.agent import AgentDef, _resolve_agent if isinstance(agent, AgentDef): agent = _resolve_agent(agent) @@ -3865,7 +3846,6 @@ def start( # Check for foreign framework agent from conductor.ai.agents.frameworks.serializer import detect_framework - from conductor.ai.agents.agent import AgentDef, _resolve_agent if isinstance(agent, AgentDef): agent = _resolve_agent(agent) @@ -4128,7 +4108,6 @@ async def run_async( # Foreign framework check from conductor.ai.agents.frameworks.serializer import detect_framework - from conductor.ai.agents.agent import AgentDef, _resolve_agent if isinstance(agent, AgentDef): agent = _resolve_agent(agent) @@ -4303,7 +4282,6 @@ async def start_async( ) from conductor.ai.agents.frameworks.serializer import detect_framework - from conductor.ai.agents.agent import AgentDef, _resolve_agent if isinstance(agent, AgentDef): agent = _resolve_agent(agent) diff --git a/tests/unit/ai/test_agent_schema_contract.py b/tests/unit/ai/test_agent_schema_contract.py index a2d31dc4..dd481167 100644 --- a/tests/unit/ai/test_agent_schema_contract.py +++ b/tests/unit/ai/test_agent_schema_contract.py @@ -52,19 +52,14 @@ def test_agent_schema_rejects_unknown_root_fields(): def test_agent_schema_accepts_jev_definition(): - from conductor.ai.agents import JevAgent, ChoiceQuestion, ScoreQuestion, BooleanQuestion + from conductor.ai.agents import JevAgent - definition = JevAgent( - "jev", - model="jev-1.13", - questions={ - "team": ChoiceQuestion("Choose", {"billing": "Payments", "technical": "Bugs"}), - "priority": ScoreQuestion("Score", ["low", "high"]), - "ready": BooleanQuestion("Ready?"), - }, - ) + questions = { + "team": {"type": "choice", "instructions": "Choose", "choices": {"a": "A", "b": "B"}}, + "priority": {"type": "score", "instructions": "Score", "scale": ["low", "high"]}, + "ready": {"type": "boolean", "instructions": "Ready?"}, + } schema = json.loads(SCHEMA_PATH.read_text()) - jsonschema.validate(AgentConfigSerializer().serialize(definition), schema) - jsonschema.validate( - AgentConfigSerializer().serialize(JevAgent("dynamic", model="jev-1.13")), schema - ) + for fixed_questions in (questions, None): + agent = JevAgent("jev", model="jev-1.13", questions=fixed_questions) + jsonschema.validate(AgentConfigSerializer().serialize(agent), schema) diff --git a/tests/unit/ai/test_jev_agent.py b/tests/unit/ai/test_jev_agent.py index 87946e11..ac72b67d 100644 --- a/tests/unit/ai/test_jev_agent.py +++ b/tests/unit/ai/test_jev_agent.py @@ -1,99 +1,17 @@ -import json +from unittest.mock import AsyncMock, patch +import asyncio import pytest -from conductor.ai.agents import ( - Agent, - AgentResult, - BooleanQuestion, - ChoiceQuestion, - JevAgent, - ScoreQuestion, -) +from conductor.ai.agents import AgentDef, AgentRuntime, JevAgent from conductor.ai.agents.config_serializer import AgentConfigSerializer -from conductor.ai.agents.jev import jev_questions -from conductor.ai.agents.runtime.runtime import AgentRuntime - - -def test_jev_agent_serializes_as_an_agent_and_as_a_child(): - definition = JevAgent( - "choose", - model="jev-1.13", - questions={ - "action": ChoiceQuestion("Choose an action", {"go": "Proceed", "wait": "Wait"}), - "quality": ScoreQuestion("Evaluate quality", ["low", "high"]), - "ready": BooleanQuestion("Ready to proceed?"), - }, - ) - assert isinstance(definition, Agent) - config = AgentConfigSerializer().serialize(definition) - assert config["kind"] == "jev" - assert "decisionProvider" not in config - assert config["questions"]["action"]["choices"] == {"go": "Proceed", "wait": "Wait"} - assert "tools" not in config - assert "credentials" not in config - parent = Agent("parent", agents=[definition], strategy="sequential") - assert AgentConfigSerializer().serialize(parent)["agents"][0]["kind"] == "jev" - assert "kind" not in AgentConfigSerializer().serialize(Agent("chat", model="openai/model")) - - -def test_dynamic_questions_and_json_state(): - agent = JevAgent("choose", model="jev-1.13") - assert "questions" not in AgentConfigSerializer().serialize(agent) - runtime = AgentRuntime.__new__(AgentRuntime) - assert json.loads(runtime._resolve_prompt({"ready": True, "value": None})) == { - "ready": True, - "value": None, - } - # No attempt to register Jev models as chat providers. - runtime._ensure_models_for_agent(agent) - assert jev_questions({"ready": BooleanQuestion("Ready?")})["ready"]["type"] == "boolean" - - -@pytest.mark.parametrize( - "question", - [ - {"type": "choice", "instructions": "Choose", "choices": {"only": "Only"}}, - {"type": "score", "instructions": "Score", "scale": []}, - {"type": "boolean", "instructions": "", "choices": {}}, - ], -) -def test_invalid_question_contract_is_rejected(question): - with pytest.raises(ValueError, match=r"requires|invalid"): - JevAgent("bad", model="jev-1.13", questions={"q": question}) - - -def test_typed_result_preserves_provider_details_without_chat_wrapping(): - raw = { - "model": "jev-1.13", - "answers": { - "action": {"type": "choice", "choice": "go", "confidence": 0.9}, - "score": {"type": "score", "score": 1.2}, - "ready": {"type": "boolean", "probability": 0.8}, - }, - "usage": {"inputTokens": 10, "outputTokens": 3, "cost": 0.001, "currency": "USD"}, - "requestId": "req", - "latencyMs": 42, - } - result = AgentResult(output={"result": raw}) - assert result.jev.answers["action"].choice == "go" - assert result.jev.answers["score"].score == raw["answers"]["score"]["score"] - assert result.jev.answers["ready"].probability == raw["answers"]["ready"]["probability"] - assert result.jev.usage == raw["usage"] - assert result.jev.latency_ms == raw["latencyMs"] - assert result.jev.request_id == "req" - assert AgentResult(output={"result": "chat"}).jev is None +from conductor.ai.agents.runtime.config import AgentConfig - -def test_failed_jev_has_no_typed_answer(): - result = AgentResult(status="FAILED", output={"agentKind": "jev", "result": {}}) - assert result.jev is None +QUESTIONS = {"ready": {"type": "boolean", "instructions": "Ready?"}} @pytest.fixture def runtime(): - from unittest.mock import patch - from conductor.ai.agents.runtime.config import AgentConfig with patch("conductor.client.orkes_clients.OrkesClients"): with AgentRuntime(settings=AgentConfig(auto_start_workers=False)) as runtime: @@ -101,9 +19,8 @@ def runtime(): def test_agent_def_and_convenience_class_serialize_identically(runtime): - from conductor.ai.agents import AgentDef - questions = {"ready": BooleanQuestion("Ready?")} + questions = QUESTIONS definition = AgentDef(name="ready", kind="jev", model="jev-1.13", questions=questions) expected = { "name": "ready", @@ -121,12 +38,11 @@ def test_agent_def_and_convenience_class_serialize_identically(runtime): {"agentConfig": expected, "prompt": "Ready to ship"} ) runtime._agent_client.start_agent.assert_not_called() + runtime._ensure_models_for_agent(JevAgent("ready", model="jev-1.13")) @pytest.mark.parametrize("status", ["COMPLETED", "FAILED", "TIMED_OUT", "TERMINATED"]) def test_start_and_poll_preserves_structured_result_and_failure_reason(runtime, status): - from unittest.mock import patch - from conductor.ai.agents import AgentDef definition = AgentDef(name="ready", kind="jev", model="jev-1.13") raw = { @@ -149,9 +65,7 @@ def test_start_and_poll_preserves_structured_result_and_failure_reason(runtime, "reasonForIncompletion": "provider failure" if status != "COMPLETED" else None, }, ] - handle = runtime.start( - definition, "Ready?", context={"questions": {"ready": BooleanQuestion("Ready?")}} - ) + handle = runtime.start(definition, "Ready?", context={"questions": QUESTIONS}) assert handle.execution_id == "execution-1" payload = runtime._agent_client.start_agent.call_args.args[0] assert payload["agentConfig"] == {"name": "ready", "kind": "jev", "model": "jev-1.13"} @@ -165,19 +79,10 @@ def test_start_and_poll_preserves_structured_result_and_failure_reason(runtime, assert result.error == (None if status == "COMPLETED" else "provider failure") -def test_dynamic_questions_required_before_start(runtime): - with pytest.raises(ValueError, match="questions"): - runtime.start(JevAgent("ready", model="jev-1.13"), "Ready?") - runtime._agent_client.start_agent.assert_not_called() - - def test_async_start_and_dynamic_compile(runtime): - import asyncio - from unittest.mock import AsyncMock - from conductor.ai.agents import AgentDef definition = AgentDef(name="ready", kind="jev", model="jev-1.13") - context = {"questions": {"ready": BooleanQuestion("Ready?")}} + context = {"questions": QUESTIONS} runtime.plan(definition, "Ready?", context=context) assert ( runtime._agent_client.compile_agent.call_args.args[0]["context"]["questions"]["ready"][ @@ -198,9 +103,28 @@ def test_async_start_and_dynamic_compile(runtime): ) -def test_jev_has_no_tool_or_internal_task_exports(): - import conductor.ai.agents as agents - - assert not hasattr(agents, "DecisionModelTool") - assert not hasattr(agents, "DecisionAgent") - assert not hasattr(agents, "JEV_AGENT") +def test_nested_jev_example_routes_to_named_agents(monkeypatch): + import runpy + from pathlib import Path + + examples = Path(__file__).resolve().parents[3] / "examples" / "agents" + monkeypatch.syspath_prepend(str(examples)) + nested = runpy.run_path(str(examples / "jev_nested_triage.py"))["triage_agent"]() + serializer = AgentConfigSerializer() + config = serializer.serialize(nested) + assert config["external"] is False + assert not config.get("model") + assert len(config["agents"]) == 3 + assert sum(len(team["agents"]) for team in config["agents"]) == 10 + for team in [config, *config["agents"]]: + assert team["router"]["kind"] == "jev" + assert set(team["router"]["questions"]["agent"]["choices"]) == { + child["name"] for child in team["agents"] + } + luna = runpy.run_path(str(examples / "luna_jev_triage.py"))["triage_agent"]("configured/luna-6") + config = serializer.serialize(luna) + assert config["router"]["model"] == "configured/luna-6" + assert len(config["agents"]) == 10 + assert all(child["kind"] == "jev" for child in config["agents"]) + assert config["maxTurns"] == 1 + assert config["synthesize"] is False diff --git a/tests/unit/ai/test_tool_extraction.py b/tests/unit/ai/test_tool_extraction.py index 80b73588..2588513d 100644 --- a/tests/unit/ai/test_tool_extraction.py +++ b/tests/unit/ai/test_tool_extraction.py @@ -587,3 +587,34 @@ def test_survives_an_unreachable_execution(self, runtime): runtime._workflow_client.get_workflow.side_effect = RuntimeError("boom") assert runtime._extract_tool_calls_and_events("exec-1") == ([], []) + + +@pytest.mark.parametrize("status", ["COMPLETED", "IN_PROGRESS", "FAILED"]) +def test_jev_is_inference_not_a_tool(runtime, status): + output = { + "model": "jev-1.13", + "answers": {"department": {"type": "choice", "choice": "billing"}}, + "usage": {"inputTokens": 10, "outputTokens": 2}, + "latencyMs": 42, + "requestId": "request-1", + } + task = FakeTask( + task_type="JEV_AGENT", + task_def_name="JEV_AGENT", + reference_task_name="support_jev", + output_data=output, + status=status, + ) + workflow = FakeWorkflowRun([task], status=status) + assert runtime._extract_tool_calls(workflow) == [] + events = runtime._extract_events(workflow, "wf-1") + assert not any(e.type in (EventType.TOOL_CALL, EventType.TOOL_RESULT) for e in events) + if status == "COMPLETED": + sse = runtime._sse_to_agent_event( + {"event": "jev", "data": {"content": "support_jev", "result": output}}, "wf-1" + ) + assert events[0] == sse + assert events[0].type == EventType.JEV + assert events[0].result == output + elif status == "FAILED": + assert any(e.type == EventType.ERROR for e in events) From 9f639f6b7a442e71e3ecf4defdfc47180fa64d8c Mon Sep 17 00:00:00 2001 From: nicholascole Date: Fri, 25 Sep 2026 14:29:15 -0700 Subject: [PATCH 6/7] feat(workflow): add AI_DECISION task and Jev routing example --- examples/agentic_workflows/README.md | 10 ++- .../agentic_workflows/ai_decision_routing.py | 72 +++++++++++++++++++ .../client/workflow/task/ai_decision_task.py | 35 +++++++++ .../client/workflow/task/task_type.py | 1 + tests/unit/workflow/test_ai_decision_task.py | 56 +++++++++++++++ 5 files changed, 172 insertions(+), 2 deletions(-) create mode 100644 examples/agentic_workflows/ai_decision_routing.py create mode 100644 src/conductor/client/workflow/task/ai_decision_task.py create mode 100644 tests/unit/workflow/test_ai_decision_task.py diff --git a/examples/agentic_workflows/README.md b/examples/agentic_workflows/README.md index b1b7fed5..fb15443f 100644 --- a/examples/agentic_workflows/README.md +++ b/examples/agentic_workflows/README.md @@ -1,11 +1,13 @@ # Agentic Workflow Examples -AI/LLM workflow examples using Conductor's built-in system tasks (`LLM_CHAT_COMPLETE`, `LLM_INDEX_TEXT`, `LLM_SEARCH_INDEX`, MCP tools) combined with Python workers. +AI workflow examples using Conductor's built-in system tasks, with Python workers where needed. -All examples use **inline ChatMessage objects** for system prompts -- no named prompt templates or AIOrchestrator required. They work with OSS Conductor with AI/LLM support. +Chat examples use inline ChatMessage objects for system prompts. No named prompt templates or AIOrchestrator required. ## Prerequisites +The AI decision example requires server-side `AI_DECISION` support and Jev credentials. It needs no chat model or Python worker. The other examples require: + - Conductor server with AI/LLM support running (e.g., `http://localhost:7001/api`) - LLM provider named `openai` configured with a valid API key - `export CONDUCTOR_SERVER_URL=http://localhost:7001/api` @@ -14,6 +16,7 @@ All examples use **inline ChatMessage objects** for system prompts -- no named p | Example | Description | Interactive? | Pattern | |---------|-------------|:------------:|---------| +| [ai_decision_routing.py](ai_decision_routing.py) | Route requests to billing or technical support | No | AI_DECISION + SwitchTask + SetVariableTask | | [llm_chat.py](llm_chat.py) | Automated multi-turn science Q&A between two LLMs | No | LoopTask + LLM_CHAT_COMPLETE + worker for history | | [llm_chat_human_in_loop.py](llm_chat_human_in_loop.py) | Interactive chat with WAIT task pauses for user input | Yes | LoopTask + WaitTask + LLM_CHAT_COMPLETE | | [multiagent_chat.py](multiagent_chat.py) | Multi-agent debate with moderator routing between panelists | No | LoopTask + SwitchTask + SetVariableTask + JavaScript routing | @@ -23,6 +26,9 @@ All examples use **inline ChatMessage objects** for system prompts -- no named p ## Quick Start ```bash +# Jev request routing with server-managed credentials +CONDUCTOR_SERVER_URL=http://localhost:8080/api python -m examples.agentic_workflows.ai_decision_routing + # Automated multi-turn chat (no interaction needed) python examples/agentic_workflows/llm_chat.py diff --git a/examples/agentic_workflows/ai_decision_routing.py b/examples/agentic_workflows/ai_decision_routing.py new file mode 100644 index 00000000..283edcc3 --- /dev/null +++ b/examples/agentic_workflows/ai_decision_routing.py @@ -0,0 +1,72 @@ +"""Route a request with AI_DECISION → SWITCH → SET_VARIABLE. + +Requires server-side AI_DECISION support and Jev credentials. No worker needed. +Run: CONDUCTOR_SERVER_URL=http://localhost:8080/api python -m examples.agentic_workflows.ai_decision_routing +""" + +import argparse +import json +import time + +from conductor.client.configuration.configuration import Configuration +from conductor.client.orkes_clients import OrkesClients +from conductor.client.workflow.conductor_workflow import ConductorWorkflow +from conductor.client.workflow.task.ai_decision_task import AiDecisionTask +from conductor.client.workflow.task.set_variable_task import SetVariableTask +from conductor.client.workflow.task.switch_task import SwitchTask + + +def create_workflow(executor) -> ConductorWorkflow: + workflow = ConductorWorkflow(executor=executor, name="ai_decision_routing", version=1) + decision = AiDecisionTask( + task_ref_name="decision", + model="jev-1.13", + state=workflow.input("request"), + questions={ + "route": { + "type": "choice", + "instructions": "Choose the team best suited to handle this request.", + "choices": { + "billing": "Payments, invoices, refunds, or subscriptions.", + "technical": "Errors, outages, or product troubleshooting.", + }, + } + }, + ) + route = SwitchTask("route_request", decision.output("selectedCase")) + route.switch_case("billing", [ + SetVariableTask("assign_billing").input_parameter("team", "billing"), + ]) + route.switch_case("technical", [ + SetVariableTask("assign_technical").input_parameter("team", "technical"), + ]) + workflow >> decision >> route + workflow.output_parameters({ + "decision": decision.output(), + "team": "${workflow.variables.team}", + }) + return workflow + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--request", default="I was charged twice on my latest invoice.") + args = parser.parse_args() + clients = OrkesClients(configuration=Configuration()) + workflow = create_workflow(clients.get_workflow_executor()) + workflow.register(overwrite=True) + workflow_id = workflow.start_workflow_with_input({"request": args.request}) + print(f"Workflow: {workflow_id}") + client = clients.get_workflow_client() + while True: + result = client.get_workflow(workflow_id=workflow_id, include_tasks=False) + if result.is_completed(): + break + time.sleep(1) + if result.status != "COMPLETED": + raise SystemExit(f"{result.status}: {result.reason_for_incompletion}") + print(json.dumps(result.output, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/src/conductor/client/workflow/task/ai_decision_task.py b/src/conductor/client/workflow/task/ai_decision_task.py new file mode 100644 index 00000000..c1d54bdd --- /dev/null +++ b/src/conductor/client/workflow/task/ai_decision_task.py @@ -0,0 +1,35 @@ +from typing import Any, Dict, Optional + +from conductor.client.workflow.task.task import TaskInterface +from conductor.client.workflow.task.task_type import TaskType + + +class AiDecisionTask(TaskInterface): + """Run a Jev choice decision using server-managed credentials. + + State and question instructions must be nonempty strings. Questions must + contain exactly one choice question with 2 to 255 choices. The server validates + inputs after resolving workflow references. + + Output retains model, answers, usage, latencyMs and optional requestId. + Use output("selectedCase") with a separate SwitchTask to route the result. + """ + + def __init__( + self, + task_ref_name: str, + model: str, + state: str, + questions: Dict[str, Any], + task_name: Optional[str] = None, + ) -> None: + super().__init__( + task_reference_name=task_ref_name, + task_type=TaskType.AI_DECISION, + task_name=task_name or "ai_decision", + input_parameters={ + "model": model, + "state": state, + "questions": questions, + }, + ) diff --git a/src/conductor/client/workflow/task/task_type.py b/src/conductor/client/workflow/task/task_type.py index 36108f72..8628ffb3 100644 --- a/src/conductor/client/workflow/task/task_type.py +++ b/src/conductor/client/workflow/task/task_type.py @@ -26,6 +26,7 @@ class TaskType(str, Enum): JSON_JQ_TRANSFORM = "JSON_JQ_TRANSFORM" SET_VARIABLE = "SET_VARIABLE" GET_DOCUMENT = "GET_DOCUMENT" + AI_DECISION = "AI_DECISION" LLM_GENERATE_EMBEDDINGS = "LLM_GENERATE_EMBEDDINGS" LLM_GET_EMBEDDINGS = "LLM_GET_EMBEDDINGS" LLM_TEXT_COMPLETE = "LLM_TEXT_COMPLETE" diff --git a/tests/unit/workflow/test_ai_decision_task.py b/tests/unit/workflow/test_ai_decision_task.py new file mode 100644 index 00000000..c29969ad --- /dev/null +++ b/tests/unit/workflow/test_ai_decision_task.py @@ -0,0 +1,56 @@ +from conductor.client.http.api_client import ApiClient +from conductor.client.workflow.task.ai_decision_task import AiDecisionTask +from conductor.client.workflow.task.task_type import TaskType +from examples.agentic_workflows.ai_decision_routing import create_workflow + + +def test_ai_decision_references_and_serialization(): + questions = { + "route": { + "type": "choice", + "instructions": "Choose a team.", + "choices": {"billing": "Payments", "technical": "Errors"}, + } + } + task = AiDecisionTask("decision", "jev-1.13", "${workflow.input.request}", questions) + serialized = ApiClient().sanitize_for_serialization(task.to_workflow_task()) + assert task.task_type == TaskType.AI_DECISION + assert serialized["type"] == "AI_DECISION" + assert serialized["name"] == "ai_decision" + assert serialized["taskReferenceName"] == "decision" + assert serialized["inputParameters"] == { + "model": "jev-1.13", + "state": "${workflow.input.request}", + "questions": questions, + } + assert task.input("state") == "${decision.input.state}" + assert task.output() == "${decision.output}" + assert task.output("answers.route.choice") == "${decision.output.answers.route.choice}" + assert task.output("selectedCase") == "${decision.output.selectedCase}" + + +def test_custom_name_and_input_reference(): + task = AiDecisionTask("decision", "jev-1.13", "request", {}, task_name="choose_team") + task.input_parameter("questions", "${workflow.input.questions}") + definition = task.to_workflow_task() + assert definition.name == "choose_team" + assert definition.input_parameters["questions"] == "${workflow.input.questions}" + + +def test_routing_example_preserves_decision_output(): + definition = ApiClient().sanitize_for_serialization(create_workflow(None).to_workflow_def()) + decision, switch = definition["tasks"] + assert decision["type"] == "AI_DECISION" + assert decision["inputParameters"]["state"] == "${workflow.input.request}" + assert switch["type"] == "SWITCH" + assert switch["evaluatorType"] == "value-param" + assert switch["inputParameters"]["switchCaseValue"] == "${decision.output.selectedCase}" + assert set(switch["decisionCases"]) == {"billing", "technical"} + for team, tasks in switch["decisionCases"].items(): + assert len(tasks) == 1 + assert tasks[0]["type"] == "SET_VARIABLE" + assert tasks[0]["inputParameters"] == {"team": team} + assert definition["outputParameters"] == { + "decision": "${decision.output}", + "team": "${workflow.variables.team}", + } From ca11744557949762cfb50f51e5a63578df25ea6b Mon Sep 17 00:00:00 2001 From: nicholascole Date: Fri, 25 Sep 2026 14:39:16 -0700 Subject: [PATCH 7/7] fix(examples): return selected Jev routing branch result --- examples/agentic_workflows/README.md | 2 +- .../agentic_workflows/ai_decision_routing.py | 33 +++++++++++++------ tests/unit/workflow/test_ai_decision_task.py | 12 ++++--- 3 files changed, 32 insertions(+), 15 deletions(-) diff --git a/examples/agentic_workflows/README.md b/examples/agentic_workflows/README.md index fb15443f..d8244c4a 100644 --- a/examples/agentic_workflows/README.md +++ b/examples/agentic_workflows/README.md @@ -16,7 +16,7 @@ The AI decision example requires server-side `AI_DECISION` support and Jev crede | Example | Description | Interactive? | Pattern | |---------|-------------|:------------:|---------| -| [ai_decision_routing.py](ai_decision_routing.py) | Route requests to billing or technical support | No | AI_DECISION + SwitchTask + SetVariableTask | +| [ai_decision_routing.py](ai_decision_routing.py) | Route requests and return the selected branch's result | No | AI_DECISION + SwitchTask + InlineTask | | [llm_chat.py](llm_chat.py) | Automated multi-turn science Q&A between two LLMs | No | LoopTask + LLM_CHAT_COMPLETE + worker for history | | [llm_chat_human_in_loop.py](llm_chat_human_in_loop.py) | Interactive chat with WAIT task pauses for user input | Yes | LoopTask + WaitTask + LLM_CHAT_COMPLETE | | [multiagent_chat.py](multiagent_chat.py) | Multi-agent debate with moderator routing between panelists | No | LoopTask + SwitchTask + SetVariableTask + JavaScript routing | diff --git a/examples/agentic_workflows/ai_decision_routing.py b/examples/agentic_workflows/ai_decision_routing.py index 283edcc3..e283dc3d 100644 --- a/examples/agentic_workflows/ai_decision_routing.py +++ b/examples/agentic_workflows/ai_decision_routing.py @@ -1,4 +1,4 @@ -"""Route a request with AI_DECISION → SWITCH → SET_VARIABLE. +"""Route a request with AI_DECISION → SWITCH → INLINE branch → INLINE result. Requires server-side AI_DECISION support and Jev credentials. No worker needed. Run: CONDUCTOR_SERVER_URL=http://localhost:8080/api python -m examples.agentic_workflows.ai_decision_routing @@ -12,7 +12,7 @@ from conductor.client.orkes_clients import OrkesClients from conductor.client.workflow.conductor_workflow import ConductorWorkflow from conductor.client.workflow.task.ai_decision_task import AiDecisionTask -from conductor.client.workflow.task.set_variable_task import SetVariableTask +from conductor.client.workflow.task.inline import InlineTask from conductor.client.workflow.task.switch_task import SwitchTask @@ -33,17 +33,30 @@ def create_workflow(executor) -> ConductorWorkflow: } }, ) + billing = InlineTask( + "handle_billing", + script='({team: "billing", nextAction: "review_invoice", ' + 'message: "Check invoice line items and payment records.", request: $.request})', + bindings={"request": workflow.input("request")}, + ) + technical = InlineTask( + "handle_technical", + script='({team: "technical", nextAction: "collect_diagnostics", ' + 'message: "Collect error logs and steps to reproduce.", request: $.request})', + bindings={"request": workflow.input("request")}, + ) route = SwitchTask("route_request", decision.output("selectedCase")) - route.switch_case("billing", [ - SetVariableTask("assign_billing").input_parameter("team", "billing"), - ]) - route.switch_case("technical", [ - SetVariableTask("assign_technical").input_parameter("team", "technical"), - ]) - workflow >> decision >> route + route.switch_case("billing", [billing]) + route.switch_case("technical", [technical]) + result = InlineTask( + "selected_result", + script="$.billing || $.technical", + bindings={"billing": billing.output("result"), "technical": technical.output("result")}, + ) + workflow >> decision >> route >> result workflow.output_parameters({ "decision": decision.output(), - "team": "${workflow.variables.team}", + "result": result.output("result"), }) return workflow diff --git a/tests/unit/workflow/test_ai_decision_task.py b/tests/unit/workflow/test_ai_decision_task.py index c29969ad..f56e2e13 100644 --- a/tests/unit/workflow/test_ai_decision_task.py +++ b/tests/unit/workflow/test_ai_decision_task.py @@ -39,7 +39,7 @@ def test_custom_name_and_input_reference(): def test_routing_example_preserves_decision_output(): definition = ApiClient().sanitize_for_serialization(create_workflow(None).to_workflow_def()) - decision, switch = definition["tasks"] + decision, switch, result = definition["tasks"] assert decision["type"] == "AI_DECISION" assert decision["inputParameters"]["state"] == "${workflow.input.request}" assert switch["type"] == "SWITCH" @@ -48,9 +48,13 @@ def test_routing_example_preserves_decision_output(): assert set(switch["decisionCases"]) == {"billing", "technical"} for team, tasks in switch["decisionCases"].items(): assert len(tasks) == 1 - assert tasks[0]["type"] == "SET_VARIABLE" - assert tasks[0]["inputParameters"] == {"team": team} + assert tasks[0]["type"] == "INLINE" + assert tasks[0]["taskReferenceName"] == f"handle_{team}" + assert tasks[0]["inputParameters"]["request"] == "${workflow.input.request}" + assert result["type"] == "INLINE" + assert result["inputParameters"]["billing"] == "${handle_billing.output.result}" + assert result["inputParameters"]["technical"] == "${handle_technical.output.result}" assert definition["outputParameters"] == { "decision": "${decision.output}", - "team": "${workflow.variables.team}", + "result": "${selected_result.output.result}", }