From a9806150fe204c5490e86ae05724386229282ec1 Mon Sep 17 00:00:00 2001 From: nicholascole Date: Mon, 19 Oct 2026 15:49:52 -0700 Subject: [PATCH] Add JEV examples --- README.md | 3 + .../agentic_workflows/ai_decision_routing.rb | 85 +++++++++++++++++ examples/agents/decision_router.rb | 54 +++++++++++ examples/agents/decision_tool.rb | 46 ++++++++++ lib/conductor/agents/agent.rb | 5 +- lib/conductor/agents/config_serializer.rb | 13 ++- lib/conductor/agents/runtime/tool_registry.rb | 2 +- lib/conductor/agents/tool.rb | 3 +- lib/conductor/workflow/dsl/task_ref.rb | 2 +- spec/conductor/agents/decision_spec.rb | 91 +++++++++++++++++++ spec/fixtures/agents/agent-schema.json | 3 + 11 files changed, 301 insertions(+), 6 deletions(-) create mode 100644 examples/agentic_workflows/ai_decision_routing.rb create mode 100644 examples/agents/decision_router.rb create mode 100644 examples/agents/decision_tool.rb create mode 100644 spec/conductor/agents/decision_spec.rb diff --git a/README.md b/README.md index 8f6c97d..28718a7 100644 --- a/README.md +++ b/README.md @@ -355,6 +355,9 @@ The `examples/` directory contains comprehensive examples: | [`dynamic_workflow.rb`](examples/dynamic_workflow.rb) | Create and execute workflows at runtime | | [`workflow_ops.rb`](examples/workflow_ops.rb) | Lifecycle operations: pause, resume, restart, retry | | [`agentic_workflows/`](examples/agentic_workflows/) | LLM chat and AI workflow examples | +| [`ai_decision_routing.rb`](examples/agentic_workflows/ai_decision_routing.rb) | Route billing or technical requests with a synchronous decision SWITCH | +| [`decision_router.rb`](examples/agents/decision_router.rb) | Print a decision router configuration; add `--run` to select and run a chat agent | +| [`decision_tool.rb`](examples/agents/decision_tool.rb) | Run a chat agent that calls a server-side decision tool | Run examples: diff --git a/examples/agentic_workflows/ai_decision_routing.rb b/examples/agentic_workflows/ai_decision_routing.rb new file mode 100644 index 0000000..26207d6 --- /dev/null +++ b/examples/agentic_workflows/ai_decision_routing.rb @@ -0,0 +1,85 @@ +# frozen_string_literal: true + +# Port of Python PR #511: the server evaluates one synchronous decision SWITCH. +# No Ruby worker or client-side inference is needed. The server needs a JEV provider. +# Run: CONDUCTOR_SERVER_URL=http://localhost:8081/api CONDUCTOR_DECISION_PROVIDER=openrouter \ +# bundle exec ruby -Ilib examples/agentic_workflows/ai_decision_routing.rb +require 'conductor' +require 'optparse' + +module ExampleAiDecisionRouting + def self.build(model: 'jev-1.13', provider: ENV['CONDUCTOR_DECISION_PROVIDER']) + models = Conductor::Http::Models + inputs = { + 'model' => model, + '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.' + } + } + } + } + inputs['provider'] = provider unless provider.nil? + billing = models::WorkflowTask.new( + name: 'handle_billing', task_reference_name: 'handle_billing', type: 'INLINE', + input_parameters: { 'evaluatorType' => 'graaljs', 'expression' => '({team: "billing", nextAction: "review_invoice", ' \ + 'message: "Check invoice line items and payment records.", request: $.request})', + 'request' => '${workflow.input.request}' } + ) + technical = models::WorkflowTask.new( + name: 'handle_technical', task_reference_name: 'handle_technical', type: 'INLINE', + input_parameters: { 'evaluatorType' => 'graaljs', 'expression' => '({team: "technical", nextAction: "collect_diagnostics", ' \ + 'message: "Collect error logs and steps to reproduce.", request: $.request})', + 'request' => '${workflow.input.request}' } + ) + decision = Conductor::Workflow::Dsl::TaskRef.new( + ref_name: 'route_request', task_name: 'route_request', task_type: 'SWITCH', input_parameters: inputs, + options: { evaluator_type: 'decision', expression: 'route', + decision_cases: { 'billing' => [billing], 'technical' => [technical] }, default_case: [] } + ) + result = models::WorkflowTask.new( + name: 'selected_result', task_reference_name: 'selected_result', type: 'INLINE', + input_parameters: { 'evaluatorType' => 'graaljs', 'expression' => '$.billing || $.technical', + 'billing' => '${handle_billing.output.result}', 'technical' => '${handle_technical.output.result}' } + ) + models::WorkflowDef.new( + name: 'ai_decision_routing_ruby', version: 1, tasks: [decision.to_workflow_task, result], + input_parameters: ['request'], + output_parameters: { + 'selectedCase' => '${route_request.output.selectedCase}', 'answers' => '${route_request.output.answers}', + 'usage' => '${route_request.output.usage}', 'cost' => '${route_request.output.usage.cost}', + 'latencyMs' => '${route_request.output.latencyMs}', 'result' => '${selected_result.output.result}' + } + ) + end + + def self.run(request: 'I was charged twice on my latest invoice.', **options) + executor = Conductor::Workflow::WorkflowExecutor.new + workflow = build(**options) + executor.register_workflow(workflow, overwrite: true) + id = executor.start_workflow(Conductor::Http::Models::StartWorkflowRequest.new( + name: workflow.name, version: 1, input: { 'request' => request } + )) + puts "Workflow: #{id}" + result = executor.wait_for_workflow(id, timeout_seconds: 120) + raise "#{result.status}: #{result.reason_for_incompletion}" unless result.status == 'COMPLETED' + + puts JSON.pretty_generate(result.output) + result + end +end + +if $PROGRAM_NAME == __FILE__ + options = {} + OptionParser.new do |parser| + parser.on('--request REQUEST') { |value| options[:request] = value } + parser.on('--model MODEL') { |value| options[:model] = value } + parser.on('--provider PROVIDER') { |value| options[:provider] = value } + end.parse! + ExampleAiDecisionRouting.run(**options) +end diff --git a/examples/agents/decision_router.rb b/examples/agents/decision_router.rb new file mode 100644 index 0000000..3214d70 --- /dev/null +++ b/examples/agents/decision_router.rb @@ -0,0 +1,54 @@ +# frozen_string_literal: true + +# Port of Python PR #511: a decision Tool selects one of two chat agents. +# The selector compiles to a synchronous server-side SWITCH; no Ruby worker is needed. +# Run without --run to print the configuration without contacting the server. +# Run: CONDUCTOR_SERVER_URL=http://localhost:8085/api CONDUCTOR_DECISION_PROVIDER=openrouter \ +# bundle exec ruby -Ilib examples/agents/decision_router.rb --run +require 'conductor/agents' + +module ExampleDecisionRouter + include Conductor::Agents + + def self.build(model: ENV.fetch('CONDUCTOR_AGENT_LLM_MODEL', 'openai/gpt-4o-mini')) + billing = Agent.new(name: 'billing_ruby', model: model, + instructions: 'Help the customer with billing and invoice questions.') + technical = Agent.new(name: 'technical_ruby', model: model, + instructions: 'Help the customer troubleshoot product issues.') + config = { + 'model' => 'jev-1.13', + 'questions' => { + 'agent' => { + 'type' => 'choice', 'instructions' => 'Choose the right support team.', + 'choices' => { 'billing_ruby' => 'Payments and invoices', 'technical_ruby' => 'Product errors and troubleshooting' } + } + } + } + config['provider'] = ENV['CONDUCTOR_DECISION_PROVIDER'] if ENV.key?('CONDUCTOR_DECISION_PROVIDER') + selector = Tool.new(name: 'support_selector_ruby', tool_type: ToolType::DECISION, + description: 'Choose the support agent that should handle the request.', config: config) + Agent.new(name: 'decision_support_router_ruby', strategy: Strategy::ROUTER, router: selector, + agents: [billing, technical], max_turns: 1, synthesize: false) + end + + def self.run(runtime: Conductor::Agents.runtime) + agent = build + runtime.deploy(agent) + execution = runtime.call_async(agent, 'The customer reports a duplicate charge on the latest invoice.') + puts "Execution: #{execution.execution_id}" + puts execution.result(timeout: 120) + execution + end +end + +if $PROGRAM_NAME == __FILE__ + if ARGV.delete('--run') + begin + ExampleDecisionRouter.run + ensure + Conductor::Agents.shutdown + end + else + puts JSON.pretty_generate(Conductor::Agents::ConfigSerializer.serialize(ExampleDecisionRouter.build)) + end +end diff --git a/examples/agents/decision_tool.rb b/examples/agents/decision_tool.rb new file mode 100644 index 0000000..582d076 --- /dev/null +++ b/examples/agents/decision_tool.rb @@ -0,0 +1,46 @@ +# frozen_string_literal: true + +# Port of Python PR #511: a normal chat agent calls a server-side decision tool. +# The tool compiles to a branchless decision SWITCH and needs no Ruby worker. +# Run: CONDUCTOR_SERVER_URL=http://localhost:8085/api CONDUCTOR_DECISION_PROVIDER=openrouter \ +# bundle exec ruby -Ilib examples/agents/decision_tool.rb +require 'conductor/agents' + +module ExampleDecisionTool + include Conductor::Agents + + def self.build(model: ENV.fetch('CONDUCTOR_AGENT_LLM_MODEL', 'openai/gpt-4o-mini')) + config = { + 'model' => 'jev-1.13', + 'questions' => { + 'department' => { + 'type' => 'choice', 'instructions' => 'Choose the correct support department.', + 'choices' => { 'billing' => 'Payments and invoices', 'technical' => 'Product errors and troubleshooting' } + } + } + } + config['provider'] = ENV['CONDUCTOR_DECISION_PROVIDER'] if ENV.key?('CONDUCTOR_DECISION_PROVIDER') + classify = Tool.new( + name: 'classify_request_ruby', description: 'Classify the customer request before answering.', tool_type: ToolType::DECISION, + input_schema: { 'type' => 'object', 'properties' => { 'state' => { 'type' => 'string' } }, 'required' => ['state'] }, + config: config + ) + Agent.new(name: 'decision_tool_support_ruby', model: model, + instructions: 'Call classify_request_ruby before answering the customer.', tools: [classify]) + end + + def self.run(runtime: Conductor::Agents.runtime) + execution = runtime.call_async(build, 'I was charged twice on my latest invoice.') + puts "Execution: #{execution.execution_id}" + puts execution.result(timeout: 120) + execution + end +end + +if $PROGRAM_NAME == __FILE__ + begin + ExampleDecisionTool.run + ensure + Conductor::Agents.shutdown + end +end diff --git a/lib/conductor/agents/agent.rb b/lib/conductor/agents/agent.rb index 3563294..39f67c0 100644 --- a/lib/conductor/agents/agent.rb +++ b/lib/conductor/agents/agent.rb @@ -47,7 +47,7 @@ class Agent attr_accessor :model, :instructions, :router, :output_type, :memory, :termination, :max_turns, :max_tokens, :timeout_seconds, :temperature, :stateful, :metadata, :description, :external, :base_url, :prefill_tools, :approval_handler, - :planner, :fallback, :fallback_max_turns, :planner_context + :planner, :fallback, :fallback_max_turns, :planner_context, :synthesize # @param name [String] ^[a-zA-Z_][a-zA-Z0-9_-]*$ # @param model [String, nil] "provider/model"; the left side is the server integration name @@ -57,7 +57,7 @@ def initialize(name:, model: nil, instructions: '', tools: [], agents: [], strat output_type: nil, guardrails: [], memory: nil, termination: nil, handoffs: [], callbacks: [], credentials: [], max_turns: 25, max_tokens: nil, timeout_seconds: 0, temperature: nil, stateful: false, metadata: nil, description: nil, external: false, base_url: nil, - prefill_tools: [], planner: nil, fallback: nil, fallback_max_turns: nil, planner_context: []) + prefill_tools: [], planner: nil, fallback: nil, fallback_max_turns: nil, planner_context: [], synthesize: true) @name = name.to_s raise ConfigurationError, "invalid agent name #{name.inspect}: must match #{NAME_PATTERN.source}" unless NAME_PATTERN.match?(@name) raise ConfigurationError, 'max_turns must be >= 1' unless max_turns.is_a?(Integer) && max_turns >= 1 @@ -91,6 +91,7 @@ def initialize(name:, model: nil, instructions: '', tools: [], agents: [], strat @fallback = fallback @fallback_max_turns = fallback_max_turns @planner_context = Array(planner_context) + @synthesize = synthesize [planner, fallback].compact.each do |child| raise ConfigurationError, 'planner and fallback must be Agents' unless child.is_a?(Agent) end diff --git a/lib/conductor/agents/config_serializer.rb b/lib/conductor/agents/config_serializer.rb index 52fb455..7334897 100644 --- a/lib/conductor/agents/config_serializer.rb +++ b/lib/conductor/agents/config_serializer.rb @@ -74,6 +74,7 @@ def serialize_plan(agent) def serialize_extras(agent) extras = {} + extras['synthesize'] = false unless agent.synthesize extras['metadata'] = agent.metadata if agent.metadata && !agent.metadata.empty? callbacks = agent.callback_positions.map { |p| { 'position' => p, 'taskName' => "#{agent.name}_#{p}" } } extras['callbacks'] = callbacks unless callbacks.empty? @@ -84,7 +85,7 @@ def serialize_extras(agent) def effective_model(agent) return agent.model if agent.model && !agent.model.to_s.empty? - return nil if agent.external + return nil if agent.external || (agent.strategy == Strategy::ROUTER && agent.router.is_a?(Tool) && agent.router.tool_type == ToolType::DECISION) inherited = (agent.agents + [agent.planner, agent.fallback].compact).map { |a| effective_model(a) }.compact.first return inherited if inherited @@ -212,6 +213,16 @@ def serialize_handoff(handoff, agent_name) def serialize_router(agent) router = agent.router + if router.is_a?(Tool) + raise ConfigurationError, 'Only a decision Tool can be used as a router' unless router.tool_type == ToolType::DECISION + + config = router.config.transform_keys(&:to_s) + raise ConfigurationError, "Decision router tool config requires 'model'" if config['model'].nil? || config['model'].empty? + raise ConfigurationError, "Decision router tool config requires 'questions'" if config['questions'].nil? || config['questions'].empty? + + return { 'name' => router.name, 'kind' => 'decision', 'model' => config['model'], + 'questions' => config['questions'], 'provider' => config['provider'] }.compact + end return serialize_agent(router) if router.is_a?(Agent) return { 'taskName' => "#{agent.name}_router_fn" } if router.respond_to?(:call) diff --git a/lib/conductor/agents/runtime/tool_registry.rb b/lib/conductor/agents/runtime/tool_registry.rb index 12169e6..f2e194e 100644 --- a/lib/conductor/agents/runtime/tool_registry.rb +++ b/lib/conductor/agents/runtime/tool_registry.rb @@ -66,7 +66,7 @@ def system_workers(agent, required, domain: nil) workers << build_system_worker(g.name, SystemWorkers.guardrail(g, logger: @logger), domain, a) if wanted?(required, g.name) end - if a.router.respond_to?(:call) + if !a.router.is_a?(Tool) && a.router.respond_to?(:call) name = "#{a.name}_router_fn" workers << build_system_worker(name, SystemWorkers.router(a.router, a.agents.map(&:name), logger: @logger), domain, a) if wanted?(required, name) end diff --git a/lib/conductor/agents/tool.rb b/lib/conductor/agents/tool.rb index 3255180..e02abb2 100644 --- a/lib/conductor/agents/tool.rb +++ b/lib/conductor/agents/tool.rb @@ -22,11 +22,12 @@ module ToolType RAG_SEARCH = 'rag_search' PULL_WORKFLOW_MESSAGES = 'pull_workflow_messages' CLI = 'cli' + DECISION = 'decision' MEDIA = [GENERATE_IMAGE, GENERATE_AUDIO, GENERATE_VIDEO, GENERATE_PDF].freeze RAG = [RAG_INDEX, RAG_SEARCH].freeze LOCAL = [WORKER, CLI].freeze - ALL = [WORKER, HTTP, API, MCP, HUMAN, AGENT_TOOL, *MEDIA, *RAG, PULL_WORKFLOW_MESSAGES, CLI].freeze + ALL = [WORKER, HTTP, API, MCP, HUMAN, AGENT_TOOL, *MEDIA, *RAG, PULL_WORKFLOW_MESSAGES, CLI, DECISION].freeze def self.valid?(type) ALL.include?(type.to_s) diff --git a/lib/conductor/workflow/dsl/task_ref.rb b/lib/conductor/workflow/dsl/task_ref.rb index 7e26e5f..2f67224 100644 --- a/lib/conductor/workflow/dsl/task_ref.rb +++ b/lib/conductor/workflow/dsl/task_ref.rb @@ -108,7 +108,7 @@ def apply_task_specific_fields(wf_task) end def apply_switch_fields(wf_task) - wf_task.evaluator_type = 'value-param' + wf_task.evaluator_type = @options[:evaluator_type] || 'value-param' wf_task.expression = @options[:expression] # Convert decision_cases from TaskRef arrays to WorkflowTask arrays diff --git a/spec/conductor/agents/decision_spec.rb b/spec/conductor/agents/decision_spec.rb new file mode 100644 index 0000000..c9aa9e8 --- /dev/null +++ b/spec/conductor/agents/decision_spec.rb @@ -0,0 +1,91 @@ +# frozen_string_literal: true + +require 'spec_helper' +require 'conductor/agents' +require 'json_schemer' +require_relative '../../../examples/agentic_workflows/ai_decision_routing' +require_relative '../../../examples/agents/decision_router' +require_relative '../../../examples/agents/decision_tool' + +RSpec.describe 'Decision routing' do + let(:agents) { Conductor::Agents } + let(:serializer) { agents::ConfigSerializer } + + it 'keeps a traditional decision as one SWITCH with branches and metadata outputs' do + workflow = ExampleAiDecisionRouting.build(provider: 'openrouter').to_h + decision = workflow['tasks'].first + expect(workflow['tasks'].map { |task| task['type'] }).to eq(%w[SWITCH INLINE]) + expect(decision).to include('evaluatorType' => 'decision', 'expression' => 'route') + expect(decision['inputParameters']).to include('model' => 'jev-1.13', 'provider' => 'openrouter', + 'state' => '${workflow.input.request}') + expect(decision['inputParameters']['questions']['route']['choices'].keys).to eq(%w[billing technical]) + expect(decision['decisionCases'].keys).to eq(%w[billing technical]) + expect(workflow['outputParameters'].keys).to eq(%w[selectedCase answers usage cost latencyMs result]) + expect(ExampleAiDecisionRouting.build(provider: nil).tasks.first.input_parameters).not_to have_key('provider') + end + + it 'preserves the default value-param evaluator for existing switches' do + task = Conductor::Workflow::Dsl::TaskRef.new(ref_name: 'route', task_name: 'route', task_type: 'SWITCH', + options: { expression: '${workflow.input.route}' }).to_workflow_task + expect(task.evaluator_type).to eq('value-param') + expect(task.expression).to eq('${workflow.input.route}') + end + + it 'preserves JavaScript evaluators and expressions on existing switches' do + task = Conductor::Workflow::Dsl::TaskRef.new(ref_name: 'route', task_name: 'route', task_type: 'SWITCH', + options: { expression: '$.route', evaluator_type: 'graaljs' }).to_workflow_task + expect(task.evaluator_type).to eq('graaljs') + expect(task.expression).to eq('$.route') + end + + it 'serializes a decision selector without making the model-less parent external or starting a worker' do + agent = ExampleDecisionRouter.build + agent.router.config['provider'] = 'openrouter' + config = serializer.serialize(agent) + expect(config).to include('external' => false, 'synthesize' => false, 'strategy' => 'router') + expect(config).not_to have_key('model') + expect(config['router']).to eq('name' => 'support_selector_ruby', 'kind' => 'decision', 'model' => 'jev-1.13', + 'provider' => 'openrouter', 'questions' => agent.router.config['questions']) + expect(config['agents'].map { |child| child['name'] }).to eq(%w[billing_ruby technical_ruby]) + registry = agents::ToolRegistry.new(agents::AgentConfig.new) + expect(registry.workers_for(agent)).to be_empty + schema = JSONSchemer.schema(JSON.parse(File.read(File.expand_path('../../fixtures/agents/agent-schema.json', __dir__)))) + expect(schema.validate(config).to_a).to be_empty + end + + it 'omits an absent router provider and accepts symbol configuration keys' do + agent = ExampleDecisionRouter.build + agent.router.config = { model: 'jev-1.13', questions: { agent: { type: 'choice' } } } + expect(serializer.serialize(agent)['router']).to eq('name' => 'support_selector_ruby', 'kind' => 'decision', + 'model' => 'jev-1.13', 'questions' => { agent: { type: 'choice' } }) + end + + it 'rejects a non-decision tool as a router' do + agent = ExampleDecisionRouter.build + agent.router = agents::Tool.http('fetch', 'http://localhost') + expect { serializer.serialize(agent) }.to raise_error(agents::ConfigurationError, /Only a decision Tool/) + end + + it 'requires decision router model and questions' do + agent = ExampleDecisionRouter.build + [nil, ''].each do |model| + agent.router.config['model'] = model + expect { serializer.serialize(agent) }.to raise_error(agents::ConfigurationError, /requires 'model'/) + end + agent.router.config['model'] = 'jev-1.13' + [nil, {}].each do |questions| + agent.router.config['questions'] = questions + expect { serializer.serialize(agent) }.to raise_error(agents::ConfigurationError, /requires 'questions'/) + end + end + + it 'uses the ordinary server-side tool serialization and worker path for a decision tool' do + agent = ExampleDecisionTool.build + tool = serializer.serialize(agent)['tools'].first + expect(tool).to include('name' => 'classify_request_ruby', 'toolType' => 'decision', 'config' => agent.tools.first.config) + expect(tool['inputSchema']['required']).to eq(['state']) + expect(agent.tools.first).to be_server_side + registry = agents::ToolRegistry.new(agents::AgentConfig.new) + expect(registry.workers_for(agent)).to be_empty + end +end diff --git a/spec/fixtures/agents/agent-schema.json b/spec/fixtures/agents/agent-schema.json index 12656b4..e5f0d35 100644 --- a/spec/fixtures/agents/agent-schema.json +++ b/spec/fixtures/agents/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 },