diff --git a/docs/agents/reference/agent-schema.json b/docs/agents/reference/agent-schema.json index 12656b47..e5f0d358 100644 --- a/docs/agents/reference/agent-schema.json +++ b/docs/agents/reference/agent-schema.json @@ -40,6 +40,9 @@ "properties": { "name": { "type": "string", "pattern": "^[a-zA-Z_][a-zA-Z0-9_-]*$" }, "model": { "type": ["string", "null"] }, + "kind": { "const": "decision" }, + "provider": { "type": "string" }, + "questions": { "type": "object", "additionalProperties": { "type": "object" } }, "baseUrl": { "type": ["string", "null"] }, "strategy": { "type": ["string", "null"] }, "maxTurns": { "type": ["integer", "null"], "minimum": 0 }, diff --git a/examples/agentic_workflows/README.md b/examples/agentic_workflows/README.md index b1b7fed5..27fe5e23 100644 --- a/examples/agentic_workflows/README.md +++ b/examples/agentic_workflows/README.md @@ -2,10 +2,13 @@ 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. -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. They work with OSS Conductor with AI/LLM support. ## Prerequisites +The decision routing example requires server-side decision-evaluator support and +Decision credentials. No chat model or Python worker is needed. 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 +17,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 and return the selected branch's result | No | Decision-backed 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 | @@ -23,6 +27,9 @@ All examples use **inline ChatMessage objects** for system prompts -- no named p ## Quick Start ```bash +# Decision 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 @@ -38,6 +45,19 @@ python examples/agentic_workflows/function_calling_example.py ## Key Patterns +### AI decision routing + +Set `SwitchTask.evaluator_type = "decision"`. Its input parameters contain +`model`, optional `provider`, `state`, and exactly one choice question in +`questions`; its `expression` names that question. The choice keys must match the +task's `decisionCases` keys. The server performs inference, writes the full +response and `selectedCase` to the SWITCH output, and runs the selected branch. +Evaluation runs synchronously during task mapping. Provider failures fail the +workflow without task retries. + +Agent definitions with a decision `ToolDef` in `router=` compile the selector and +selected child workflow into this same decision-backed SWITCH shape automatically. + ### Passing dynamic messages to LLM_CHAT_COMPLETE When passing a workflow reference as `messages`, set it via `input_parameters` AFTER construction: diff --git a/examples/agentic_workflows/ai_decision_routing.py b/examples/agentic_workflows/ai_decision_routing.py new file mode 100644 index 00000000..3f90093a --- /dev/null +++ b/examples/agentic_workflows/ai_decision_routing.py @@ -0,0 +1,100 @@ +"""Route a request with one decision-backed SWITCH and an inline branch. + +The server evaluates the SWITCH synchronously and selects its branch. Provider +failures fail the workflow without task retries. No Python worker is 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.inline import InlineTask +from conductor.client.workflow.task.switch_task import EvaluatorType, SwitchTask + + +def create_workflow(executor, model="jev-1.13", provider=None) -> ConductorWorkflow: + workflow = ConductorWorkflow(executor=executor, name="ai_decision_routing", version=1) + decision = SwitchTask( + task_ref_name="route_request", + case_expression="route", + ) + decision.evaluator_type = EvaluatorType.DECISION.value + decision.input_parameters = { + "model": model, + **({"provider": provider} if provider is not None else {}), + "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.", + }, + } + }, + } + 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")}, + ) + decision.switch_case("billing", [billing]) + decision.switch_case("technical", [technical]) + result = InlineTask( + "selected_result", + script="$.billing || $.technical", + bindings={ + "billing": billing.output("result"), + "technical": technical.output("result"), + }, + ) + workflow >> decision >> result + workflow.output_parameters( + { + "selectedCase": decision.output("selectedCase"), + "answers": decision.output("answers"), + "usage": decision.output("usage"), + "cost": decision.output("usage.cost"), + "latencyMs": decision.output("latencyMs"), + "result": result.output("result"), + } + ) + return workflow + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--request", default="I was charged twice on my latest invoice.") + parser.add_argument("--model", default="jev-1.13") + parser.add_argument("--provider") + args = parser.parse_args() + clients = OrkesClients(configuration=Configuration()) + workflow = create_workflow(clients.get_workflow_executor(), args.model, args.provider) + 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/examples/agents/README.md b/examples/agents/README.md index 41bae16e..40ec3d48 100644 --- a/examples/agents/README.md +++ b/examples/agents/README.md @@ -26,3 +26,15 @@ 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. + +Decision: [decision_router.py](decision_router.py) uses a decision `ToolDef` as a +router selector, with normal chat agents as executable children. Deployment +compiles it into one decision-backed `SWITCH`; the tool is not deployed as a +standalone agent. It compiles by default; pass `--run` for inference. + +`decision_tool.py` shows the other agent pattern: a normal chat agent calls +Decision inference as a server-side tool. The server maps the tool call to a +branchless decision-backed `SWITCH`; no tool worker is needed. + +For direct use in a traditional workflow, see +[`ai_decision_routing.py`](../agentic_workflows/ai_decision_routing.py). diff --git a/examples/agents/decision_router.py b/examples/agents/decision_router.py new file mode 100644 index 00000000..12d4c3ca --- /dev/null +++ b/examples/agents/decision_router.py @@ -0,0 +1,86 @@ +"""Route to one of two chat agents with a decision tool. + +When a decision ToolDef is supplied as ``router=``, deployment compiles it into +one decision-backed SWITCH. It is a selector configuration, not a standalone +agent, and needs no Python worker. Pass --run for inference. +""" + +import argparse +import json +import os + +from conductor.ai.agents import Agent, AgentRuntime, Strategy, ToolDef +from conductor.client.configuration.configuration import Configuration + +PROMPT = "The customer reports a duplicate charge on the latest invoice." +CHAT_MODEL = os.getenv("CONDUCTOR_AGENT_LLM_MODEL", "openai/gpt-4o-mini") + + +def support_router(): + billing = Agent( + name="billing", + model=CHAT_MODEL, + instructions="Help the customer with billing and invoice questions.", + ) + technical = Agent( + name="technical", + model=CHAT_MODEL, + instructions="Help the customer troubleshoot product issues.", + ) + selector = ToolDef( + name="support_selector", + description="Choose the support agent that should handle the request.", + tool_type="decision", + config={ + "model": "jev-1.13", + **( + {"provider": os.environ["CONDUCTOR_DECISION_PROVIDER"]} + if "CONDUCTOR_DECISION_PROVIDER" in os.environ + else {} + ), + "questions": { + "agent": { + "type": "choice", + "instructions": "Choose the right support team.", + "choices": { + "billing": "Payments and invoices", + "technical": "Product errors and troubleshooting", + }, + } + }, + }, + ) + return Agent( + name="decision_support_router", + strategy=Strategy.ROUTER, + router=selector, + agents=[billing, technical], + max_turns=1, + synthesize=False, + ) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--run", action="store_true") + 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_router() + if not args.run: + print(json.dumps(runtime.plan(agent), indent=2)) + return + + runtime.deploy(agent) + 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}") + print(json.dumps(result.output, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/examples/agents/decision_tool.py b/examples/agents/decision_tool.py new file mode 100644 index 00000000..34e0e93b --- /dev/null +++ b/examples/agents/decision_tool.py @@ -0,0 +1,66 @@ +"""Use Decision inference as a tool inside a normal chat agent. + +The server executes the tool as a branchless decision-backed SWITCH, so no Python +worker is needed. +""" + +import os + +from conductor.ai.agents import Agent, AgentRuntime, ToolDef +from conductor.client.configuration.configuration import Configuration + +decision_config = { + "model": "jev-1.13", + **( + {"provider": os.environ["CONDUCTOR_DECISION_PROVIDER"]} + if "CONDUCTOR_DECISION_PROVIDER" in os.environ + else {} + ), + "questions": { + "department": { + "type": "choice", + "instructions": "Choose the correct support department.", + "choices": { + "billing": "Payments and invoices", + "technical": "Product errors and troubleshooting", + }, + } + }, +} + +classify_request = ToolDef( + name="classify_request", + description="Classify the customer request before answering.", + tool_type="decision", + input_schema={ + "type": "object", + "properties": {"state": {"type": "string"}}, + "required": ["state"], + }, + config=decision_config, +) + +support_agent = Agent( + name="decision_tool_support", + model=os.getenv("CONDUCTOR_AGENT_LLM_MODEL", "openai/gpt-4o-mini"), + instructions="Call classify_request before answering the customer.", + tools=[classify_request], +) + + +def main(): + config = Configuration( + server_api_url=os.getenv("CONDUCTOR_SERVER_URL", "http://localhost:8080/api") + ) + with AgentRuntime(config) as runtime: + result = runtime.run( + support_agent, + "I was charged twice on my latest invoice.", + ) + if not result.is_success: + raise RuntimeError(result.error) + print(result.output) + + +if __name__ == "__main__": + main() diff --git a/src/conductor/ai/agents/agent.py b/src/conductor/ai/agents/agent.py index b199aa11..40ef6382 100644 --- a/src/conductor/ai/agents/agent.py +++ b/src/conductor/ai/agents/agent.py @@ -15,6 +15,7 @@ from typing import Any, Callable, Dict, List, Optional, Union from conductor.ai.agents.claude_code import ClaudeCode +from conductor.ai.agents.tool import ToolDef _VALID_NAME_RE = re.compile(r"^[a-zA-Z_][a-zA-Z0-9_-]*$") @@ -496,8 +497,8 @@ class Agent: ``"handoff"``). Valid values: ``handoff``, ``sequential``, ``parallel``, ``router``, ``round_robin``, ``random``, ``swarm``, ``manual``. - router: For ``strategy="router"``, an :class:`Agent` or callable that - selects which sub-agent runs each turn. + router: For ``strategy="router"``, an :class:`Agent`, callable, or + decision :class:`ToolDef` that selects which sub-agent runs each turn. output_type: A Pydantic model or dataclass for structured output. guardrails: List of :class:`Guardrail` instances for input/output validation. memory: Optional :class:`ConversationMemory` for session management. @@ -555,7 +556,7 @@ def __init__( tools: Optional[List[Any]] = None, agents: Optional[List[Any]] = None, strategy: Union[str, Strategy] = Strategy.HANDOFF, - router: Optional[Union["Agent", Callable[..., Any]]] = None, + router: Optional[Union["Agent", Callable[..., Any], "ToolDef"]] = None, output_type: Optional[type] = None, guardrails: Optional[List[Any]] = None, memory: Optional[Any] = None, @@ -913,11 +914,14 @@ 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 decision tool, which the server compiles inline. """ - return not self.model + return not self.model and not ( + self.strategy == Strategy.ROUTER + and isinstance(self.router, ToolDef) + and self.router.tool_type == "decision" + ) # ── Instance-method resolution ────────────────────────────────────── diff --git a/src/conductor/ai/agents/config_serializer.py b/src/conductor/ai/agents/config_serializer.py index fcc8ed67..0c57619b 100644 --- a/src/conductor/ai/agents/config_serializer.py +++ b/src/conductor/ai/agents/config_serializer.py @@ -11,6 +11,8 @@ import logging from typing import TYPE_CHECKING, Any, Dict +from conductor.ai.agents.tool import ToolDef + if TYPE_CHECKING: from conductor.ai.agents.agent import Agent @@ -455,8 +457,22 @@ def _serialize_handoff(self, handoff: Any, agent_name: str) -> dict: def _serialize_router(self, agent: "Agent") -> Any: """Serialize a router to either an AgentConfig or a WorkerRef.""" from conductor.ai.agents.agent import Agent as AgentClass - router = agent.router + if isinstance(router, ToolDef): + if router.tool_type != "decision": + raise ValueError("Only a decision ToolDef can be used as a router") + config = router.config + if not config.get("model"): + raise ValueError("Decision router tool config requires 'model'") + if not config.get("questions"): + raise ValueError("Decision router tool config requires 'questions'") + return { + "name": router.name, + "kind": "decision", + "model": config["model"], + "questions": config["questions"], + **({"provider": config["provider"]} if config.get("provider") is not None else {}), + } if isinstance(router, AgentClass) or (hasattr(router, "model") and router.model): return self._serialize_agent(router) elif callable(router): diff --git a/src/conductor/client/workflow/task/switch_task.py b/src/conductor/client/workflow/task/switch_task.py index a85fce91..8267a001 100644 --- a/src/conductor/client/workflow/task/switch_task.py +++ b/src/conductor/client/workflow/task/switch_task.py @@ -13,6 +13,7 @@ class EvaluatorType(str, Enum): JAVASCRIPT = "javascript", ECMASCRIPT = "graaljs", VALUE_PARAM = "value-param" + DECISION = "decision" class SwitchTask(TaskInterface): @@ -42,7 +43,9 @@ def default_case(self, tasks: List[TaskInterface]) -> Self: def to_workflow_task(self) -> WorkflowTask: workflow = super().to_workflow_task() - if self._use_javascript: + if self.evaluator_type is not None: + workflow.expression = self._expression + elif self._use_javascript: workflow.evaluator_type = EvaluatorType.ECMASCRIPT workflow.expression = self._expression else: diff --git a/tests/unit/ai/test_agent_schema_contract.py b/tests/unit/ai/test_agent_schema_contract.py index eecd1db3..d19397ae 100644 --- a/tests/unit/ai/test_agent_schema_contract.py +++ b/tests/unit/ai/test_agent_schema_contract.py @@ -12,7 +12,7 @@ from conductor.ai.agents.config_serializer import AgentConfigSerializer from conductor.ai.agents.guardrail import RegexGuardrail from conductor.ai.agents.termination import TextMentionTermination -from conductor.ai.agents.tool import tool +from conductor.ai.agents.tool import ToolDef, tool ROOT = Path(__file__).resolve().parents[3] @@ -49,3 +49,19 @@ 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_decision_tool_router(): + questions = {"team": {"type": "choice", "choices": {"a": "A", "b": "B"}}} + schema = json.loads(SCHEMA_PATH.read_text()) + agent = Agent( + name="router", + strategy="router", + router=ToolDef( + name="selector", + tool_type="decision", + config={"model": "jev-1.13", "questions": questions}, + ), + agents=[Agent(name="a", model="openai/gpt-4o-mini")], + ) + jsonschema.validate(AgentConfigSerializer().serialize(agent), schema) diff --git a/tests/unit/ai/test_decision_tool.py b/tests/unit/ai/test_decision_tool.py new file mode 100644 index 00000000..9833fb5b --- /dev/null +++ b/tests/unit/ai/test_decision_tool.py @@ -0,0 +1,47 @@ +import pytest + +from conductor.ai.agents import Agent, Strategy, ToolDef +from conductor.ai.agents.config_serializer import AgentConfigSerializer +from examples.agents.decision_router import support_router +from examples.agents.decision_tool import support_agent + + +def test_decision_router_example_serializes_inline_selector(monkeypatch): + monkeypatch.setenv("CONDUCTOR_DECISION_PROVIDER", "typesafe") + config = AgentConfigSerializer().serialize(support_router()) + selector = config["router"] + + assert config["external"] is False + assert selector["kind"] == "decision" + assert selector["model"] == "jev-1.13" + assert selector["provider"] == "typesafe" + assert set(selector["questions"]["agent"]["choices"]) == { + child["name"] for child in config["agents"] + } + assert Agent(name="existing_workflow").external is True + + +@pytest.mark.parametrize( + ("tool_type", "config", "message"), + [ + ("http", {}, "Only a decision ToolDef"), + ("decision", {"questions": {}}, "requires 'model'"), + ("decision", {"model": "jev-1.13"}, "requires 'questions'"), + ], +) +def test_decision_router_rejects_invalid_selector(tool_type, config, message): + agent = Agent( + name="router", + strategy=Strategy.ROUTER, + router=ToolDef(name="selector", tool_type=tool_type, config=config), + ) + with pytest.raises(ValueError, match=message): + AgentConfigSerializer().serialize(agent) + + +def test_decision_tool_example_uses_existing_tool_serialization(): + config = AgentConfigSerializer().serialize(support_agent) + tool = config["tools"][0] + assert tool["toolType"] == "decision" + assert tool["config"]["model"] == "jev-1.13" + assert set(tool["config"]["questions"]["department"]["choices"]) == {"billing", "technical"} diff --git a/tests/unit/workflow/test_decision_switch_task.py b/tests/unit/workflow/test_decision_switch_task.py new file mode 100644 index 00000000..d37423e2 --- /dev/null +++ b/tests/unit/workflow/test_decision_switch_task.py @@ -0,0 +1,54 @@ +import pytest + +from conductor.client.http.api_client import ApiClient +from conductor.client.workflow.task.switch_task import SwitchTask +from examples.agentic_workflows.ai_decision_routing import create_workflow + + +def test_direct_example_serializes_decision_switch_and_branches(): + definition = ApiClient().sanitize_for_serialization( + create_workflow(None, provider="typesafe").to_workflow_def() + ) + decision = definition["tasks"][0] + inputs = decision["inputParameters"] + assert (decision["type"], decision["evaluatorType"], decision["expression"]) == ( + "SWITCH", + "decision", + "route", + ) + assert inputs["model"] == "jev-1.13" + assert inputs["provider"] == "typesafe" + assert inputs["state"] == "${workflow.input.request}" + assert set(inputs["questions"]["route"]["choices"]) == set(decision["decisionCases"]) + assert not {"retryCount", "asyncComplete"}.intersection(decision) + assert definition["outputParameters"]["selectedCase"] == "${route_request.output.selectedCase}" + assert definition["outputParameters"]["result"] == "${selected_result.output.result}" + + +@pytest.mark.parametrize( + ("javascript", "expression", "inputs", "evaluator", "serialized_expression"), + [ + ( + False, + "${workflow.input.route}", + {"switchCaseValue": "${workflow.input.route}"}, + "value-param", + "switchCaseValue", + ), + (True, "return $.route;", {}, "graaljs", "return $.route;"), + ], +) +def test_existing_switch_serialization( + javascript, expression, inputs, evaluator, serialized_expression +): + task = SwitchTask("route", expression, use_javascript=javascript) + assert ApiClient().sanitize_for_serialization(task.to_workflow_task()) == { + "name": "route", + "taskReferenceName": "route", + "inputParameters": inputs, + "type": "SWITCH", + "decisionCases": {}, + "defaultCase": [], + "evaluatorType": evaluator, + "expression": serialized_expression, + }