diff --git a/docs/agents/reference/agent-definition.md b/docs/agents/reference/agent-definition.md index fe630aa9..6b4a1acd 100644 --- a/docs/agents/reference/agent-definition.md +++ b/docs/agents/reference/agent-definition.md @@ -10,3 +10,24 @@ 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, 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}`. + +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. + +[Example](../../../examples/agents/jev_agent.py) + +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 12656b47..425d9019 100644 --- a/docs/agents/reference/agent-schema.json +++ b/docs/agents/reference/agent-schema.json @@ -40,6 +40,8 @@ "properties": { "name": { "type": "string", "pattern": "^[a-zA-Z_][a-zA-Z0-9_-]*$" }, "model": { "type": ["string", "null"] }, + "kind": { "const": "jev" }, + "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..d8244c4a 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 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 | @@ -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..e283dc3d --- /dev/null +++ b/examples/agentic_workflows/ai_decision_routing.py @@ -0,0 +1,85 @@ +"""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 +""" + +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.inline import InlineTask +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.", + }, + } + }, + ) + 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", [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(), + "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.") + 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/examples/agents/README.md b/examples/agents/README.md index 41bae16e..690f6f45 100644 --- a/examples/agents/README.md +++ b/examples/agents/README.md @@ -26,3 +26,11 @@ 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: [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 new file mode 100644 index 00000000..6923467d --- /dev/null +++ b/examples/agents/jev_agent.py @@ -0,0 +1,53 @@ +"""Compile a Jev agent. Pass --run for inference. Configure credentials on Conductor.""" + +import argparse +import json +import os + +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." + + +def support_agent(): + return JevAgent( + name="jev_support_agent", + model="jev-1.13", + questions={ + "department": { + "type": "choice", + "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: + print(json.dumps(runtime.plan(agent, PROMPT), indent=2)) + return + + 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["result"], indent=2)) + + +if __name__ == "__main__": + main() 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 fce5b308..b42b7934 100644 --- a/src/conductor/ai/agents/__init__.py +++ b/src/conductor/ai/agents/__init__.py @@ -50,6 +50,7 @@ def get_weather(city: str) -> str: LocalCodeExecutor, ServerlessCodeExecutor, ) +from conductor.ai.agents.jev import JevAgent # Exceptions from conductor.ai.agents.exceptions import ( @@ -232,6 +233,7 @@ def resolve_credentials(task: object, names: list) -> dict: from conductor.ai.agents.tracing import is_tracing_enabled __all__ = [ + "JevAgent", # OpenAI Agents SDK compatibility "Runner", "RunResult", diff --git a/src/conductor/ai/agents/agent.py b/src/conductor/ai/agents/agent.py index b199aa11..583b11bc 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. 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. 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,33 @@ 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 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}") # Handle ClaudeCode: don't inherit parent model for claude-code agents if isinstance(ad.model, ClaudeCode): resolved_model = ad.model @@ -229,7 +254,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 +274,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 ──────────────────────────────────── @@ -913,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 fcc8ed67..9800bb6e 100644 --- a/src/conductor/ai/agents/config_serializer.py +++ b/src/conductor/ai/agents/config_serializer.py @@ -36,7 +36,21 @@ 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": + 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"] = agent.questions + if agent.metadata: + config["metadata"] = agent.metadata + 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/jev.py b/src/conductor/ai/agents/jev.py new file mode 100644 index 00000000..f9d3e70f --- /dev/null +++ b/src/conductor/ai/agents/jev.py @@ -0,0 +1,23 @@ +"""Jev agent definitions. Inference and credentials stay on Conductor.""" + +from copy import deepcopy +from typing import Any, Dict, Optional + +from conductor.ai.agents.agent import Agent + + +class JevAgent(Agent): + """Supply questions here or through context.questions at runtime.""" + + kind = "jev" + + def __init__( + self, + name: str, + *, + model: str, + questions: Optional[Dict[str, Any]] = None, + metadata: Optional[Dict[str, Any]] = None, + ): + 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 29ccc92f..41e8b335 100644 --- a/src/conductor/ai/agents/result.py +++ b/src/conductor/ai/agents/result.py @@ -94,7 +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``). + ``"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). @@ -598,7 +598,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, ) @@ -711,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" @@ -732,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 e53b921a..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( @@ -1106,6 +1116,9 @@ def prepare(self, agent: Any) -> None: """ from conductor.ai.agents.frameworks.serializer import detect_framework + if isinstance(agent, AgentDef): + agent = _resolve_agent(agent) + if isinstance(agent, str): return # nothing to prepare for run-by-name @@ -2233,7 +2246,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,14 +2312,22 @@ 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 @@ -2314,6 +2335,9 @@ def plan(self, agent: Agent) -> Any: """ from conductor.ai.agents.frameworks.serializer import detect_framework + if isinstance(agent, AgentDef): + agent = _resolve_agent(agent) + framework = detect_framework(agent) if framework: from conductor.ai.agents.frameworks.serializer import serialize_agent @@ -2330,6 +2354,10 @@ 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: + payload["context"] = context return self._agent_client.compile_agent(payload) # ── Deploy (CI/CD) ───────────────────────────────────────────── @@ -2376,6 +2404,9 @@ def deploy( for agent in all_agents: from conductor.ai.agents.frameworks.serializer import detect_framework + if isinstance(agent, AgentDef): + agent = _resolve_agent(agent) + framework = detect_framework(agent) registered_name = self._deploy_via_server(agent, framework=framework) @@ -2414,6 +2445,9 @@ async def deploy_async( for agent in all_agents: from conductor.ai.agents.frameworks.serializer import detect_framework + if isinstance(agent, AgentDef): + agent = _resolve_agent(agent) + framework = detect_framework(agent) registered_name = await self._deploy_via_server_async(agent, framework=framework) @@ -2548,6 +2582,8 @@ def serve( 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 @@ -2698,6 +2734,9 @@ def run( # Check for foreign framework agent from conductor.ai.agents.frameworks.serializer import detect_framework + if isinstance(agent, AgentDef): + agent = _resolve_agent(agent) + framework = detect_framework(agent) if framework is not None: @@ -3808,6 +3847,9 @@ def start( # Check for foreign framework agent from conductor.ai.agents.frameworks.serializer import detect_framework + if isinstance(agent, AgentDef): + agent = _resolve_agent(agent) + framework = detect_framework(agent) if framework is not None: return self._start_framework( @@ -4067,6 +4109,9 @@ async def run_async( # Foreign framework check from conductor.ai.agents.frameworks.serializer import detect_framework + if isinstance(agent, AgentDef): + agent = _resolve_agent(agent) + framework = detect_framework(agent) if framework is not None: @@ -4238,6 +4283,9 @@ async def start_async( from conductor.ai.agents.frameworks.serializer import detect_framework + if isinstance(agent, AgentDef): + agent = _resolve_agent(agent) + framework = detect_framework(agent) if framework is not None: return await self._start_framework_async( 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/ai/test_agent_schema_contract.py b/tests/unit/ai/test_agent_schema_contract.py index eecd1db3..dd481167 100644 --- a/tests/unit/ai/test_agent_schema_contract.py +++ b/tests/unit/ai/test_agent_schema_contract.py @@ -49,3 +49,17 @@ 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 + + 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()) + 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 new file mode 100644 index 00000000..ac72b67d --- /dev/null +++ b/tests/unit/ai/test_jev_agent.py @@ -0,0 +1,130 @@ +from unittest.mock import AsyncMock, patch +import asyncio + +import pytest + +from conductor.ai.agents import AgentDef, AgentRuntime, JevAgent +from conductor.ai.agents.config_serializer import AgentConfigSerializer +from conductor.ai.agents.runtime.config import AgentConfig + +QUESTIONS = {"ready": {"type": "boolean", "instructions": "Ready?"}} + + +@pytest.fixture +def runtime(): + + 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): + + questions = QUESTIONS + 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() + 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): + + 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": 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"} + 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_async_start_and_dynamic_compile(runtime): + + definition = AgentDef(name="ready", kind="jev", model="jev-1.13") + context = {"questions": QUESTIONS} + 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_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) 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..f56e2e13 --- /dev/null +++ b/tests/unit/workflow/test_ai_decision_task.py @@ -0,0 +1,60 @@ +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, result = 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"] == "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}", + "result": "${selected_result.output.result}", + }