Class: LanguageOperator::Agent::TaskExecutor

Inherits:
Object
  • Object
show all
Includes:
Instrumentation, Instrumentation::TaskTracer, Loggable
Defined in:
lib/language_operator/agent/task_executor.rb

Overview

Task Executor for DSL v1 organic functions

Executes both neural (LLM-based) and symbolic (code-based) tasks. Provides the execute_task method that MainDefinition blocks use to invoke tasks transparently regardless of implementation type.

Examples:

Executing a task

executor = TaskExecutor.new(agent, tasks_registry)
result = executor.execute_task(:fetch_data, inputs: { user_id: 123 })

In a main block

main do |inputs|
  data = execute_task(:fetch_data, inputs: inputs)
  execute_task(:process_data, inputs: data)
end

Constant Summary collapse

RETRYABLE_ERRORS =

Error types that should be retried

[
  Timeout::Error,
  Errno::ECONNREFUSED,
  Errno::ECONNRESET,
  Errno::ETIMEDOUT,
  SocketError
].freeze
ERROR_CATEGORIES =

Error categories for logging and operator integration

{
  validation: 'VALIDATION',
  execution: 'EXECUTION',
  timeout: 'TIMEOUT',
  network: 'NETWORK',
  system: 'SYSTEM'
}.freeze

Constants included from Instrumentation::TaskTracer

Instrumentation::TaskTracer::MAX_CAPTURED_LENGTH

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from Loggable

#logger

Constructor Details

#initialize(agent, tasks = {}, config = {}) ⇒ TaskExecutor

Initialize the task executor

Parameters:

  • agent (LanguageOperator::Agent::Base) —

    The agent instance (provides LLM client, tools)

  • tasks (Hash<Symbol, TaskDefinition>) (defaults to: {}) —

    Registry of task definitions

  • config (Hash) (defaults to: {}) —

    Execution configuration



76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
 
# File 'lib/language_operator/agent/task_executor.rb', line 76

def initialize(agent, tasks = {}, config = {})
  @agent = agent
  @tasks = tasks
  @config = default_config.merge(config)

  # Pre-cache task lookup and timeout information for performance
  @task_cache = build_task_cache
  @task_timeouts = build_timeout_cache

  logger.debug('TaskExecutor initialized',
               task_count: @tasks.size,
               timeout_symbolic: @config[:timeout_symbolic],
               timeout_neural: @config[:timeout_neural],
               timeout_hybrid: @config[:timeout_hybrid],
               max_retries: @config[:max_retries])
end

Instance Attribute Details

#agent ⇒ Object (readonly)

Returns the value of attribute agent.



69
70
71
 
# File 'lib/language_operator/agent/task_executor.rb', line 69

def agent
  @agent
end

#config ⇒ Object (readonly)

Returns the value of attribute config.



69
70
71
 
# File 'lib/language_operator/agent/task_executor.rb', line 69

def config
  @config
end

#tasks ⇒ Object (readonly)

Returns the value of attribute tasks.



69
70
71
 
# File 'lib/language_operator/agent/task_executor.rb', line 69

def tasks
  @tasks
end

Instance Method Details

#execute_llm(prompt) ⇒ String

Helper method for symbolic tasks to call LLM directly

Parameters:

  • prompt (String) —

    Prompt to send to LLM

Returns:

  • (String) —

    LLM response



319
320
321
322
 
# File 'lib/language_operator/agent/task_executor.rb', line 319

def execute_llm(prompt)
  response = @agent.send_message(prompt)
  response.is_a?(String) ? response : response.content
end

#execute_neural(task, inputs) ⇒ Hash

Execute a neural task (instructions-based, LLM-driven)

Parameters:

  • task (TaskDefinition) —

    The task definition

  • inputs (Hash) —

    Input parameters

Returns:

  • (Hash) —

    Validated outputs

Raises:

  • (StandardError) —

    If LLM execution fails or output validation fails



168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
 
# File 'lib/language_operator/agent/task_executor.rb', line 168

def execute_neural(task, inputs)
  # Validate inputs first
  validated_inputs = task.validate_inputs(inputs)

  logger.debug('Executing neural task',
               task: task.name,
               instructions: task.instructions_text,
               inputs: validated_inputs)

  # Build prompt for LLM
  prompt = build_neural_prompt(task, validated_inputs)

  logger.info('Sending prompt to LLM',
              task: task.name,
              prompt_length: prompt.length,
              available_tools: @agent.respond_to?(:tools) ? @agent.tools.map(&:name) : 'N/A')

  # Execute LLM call within traced span
  outputs = tracer.in_span('gen_ai.chat', attributes: neural_task_attributes(task, prompt, validated_inputs)) do |span|
    # Call LLM with full tool access
    logger.debug('Calling LLM with prompt', task: task.name, prompt_preview: prompt[0..200])
    response = @agent.send_message(prompt)

    # Check for tool calls and log details
    has_tool_calls = response.respond_to?(:tool_calls) && response.tool_calls&.any?
    tool_call_count = has_tool_calls ? response.tool_calls.length : 0

    logger.info('LLM response received, extracting content',
                task: task.name,
                response_class: response.class.name,
                has_tool_calls: has_tool_calls,
                tool_call_count: tool_call_count)

    response_text = response.is_a?(String) ? response : response.content

    logger.info('Neural task response received',
                task: task.name,
                response_length: response_text.length)

    # Record token usage and response metadata
    record_token_usage(response, span)

    # Record tool calls if available
    record_tool_calls(response, span)

    logger.info('Parsing neural task response',
                task: task.name)

    # Parse response within child span with retry logic
    parsed_outputs = tracer.in_span('task_executor.parse_response') do |parse_span|
      record_parse_metadata(response_text, parse_span)

      begin
        parse_neural_response(response_text, task)
      rescue RuntimeError => e
        # If parsing fails and this is a JSON parsing error, try one more time with clarified prompt
        raise e unless e.message.include?('returned invalid JSON') && !@parsing_retry_attempted

        @parsing_retry_attempted = true

        logger.warn('JSON parsing failed, retrying with clarified prompt',
                    task: task.name,
                    original_error: e.message,
                    response_preview: response_text[0..300])

        # Build retry prompt with clearer instructions
        retry_prompt = build_parsing_retry_prompt(task, validated_inputs, response_text, e.message)

        logger.info('Retrying LLM call with clarified prompt',
                    task: task.name,
                    retry_prompt_length: retry_prompt.length)

        # Retry LLM call
        retry_response = @agent.send_message(retry_prompt)
        retry_response_text = retry_response.is_a?(String) ? retry_response : retry_response.content

        logger.info('Parsing retry response',
                    task: task.name,
                    retry_response_length: retry_response_text.length)

        # Try parsing the retry response
        parse_neural_response(retry_response_text, task)
      end
    end

    logger.info('Response parsed successfully',
                task: task.name,
                output_keys: parsed_outputs.keys)

    # Record output metadata
    record_output_metadata(parsed_outputs, span)

    parsed_outputs
  end

  logger.info('Validating task outputs',
              task: task.name)

  # Validate outputs against schema
  task.validate_outputs(outputs)
end

#execute_parallel(tasks, in_threads: 4) ⇒ Array

Execute multiple tasks in parallel

Provides explicit parallelism for task execution. Users specify which tasks should run in parallel, and this method handles the concurrent execution.

Examples:

Execute multiple independent tasks

results = execute_parallel([
  { name: :fetch_source1 },
  { name: :fetch_source2 }
])
# => [result1, result2]

With inputs

results = execute_parallel([
  { name: :process, inputs: { data: data1 } },
  { name: :analyze, inputs: { data: data2 } }
])

Parameters:

  • tasks (Array<Hash>) —

    Array of task specifications

  • in_threads (Integer) (defaults to: 4) —

    Number of threads to use (default: 4)

Returns:

  • (Array) —

    Results from all tasks in the same order as input

Raises:

  • (RuntimeError) —

    If any task fails



347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
 
# File 'lib/language_operator/agent/task_executor.rb', line 347

def execute_parallel(tasks, in_threads: 4)
  require 'parallel'

  # Capture current OpenTelemetry context before parallel execution
  current_context = OpenTelemetry::Context.current

  logger.info('Executing tasks in parallel', count: tasks.size, threads: in_threads)

  results = Parallel.map(tasks, in_threads: in_threads) do |task_spec|
    # Restore OpenTelemetry context in worker thread
    OpenTelemetry::Context.with_current(current_context) do
      task_name = task_spec[:name]
      task_inputs = task_spec[:inputs] || {}

      execute_task(task_name, inputs: task_inputs)
    end
  end

  logger.info('Parallel execution complete', results_count: results.size)
  results
rescue Parallel::DeadWorker => e
  logger.error('Parallel execution failed - worker died', error: e.message)
  raise "Parallel task execution failed: #{e.message}"
rescue StandardError => e
  logger.error('Parallel execution failed', error: e.class.name, message: e.message)
  raise
end

#execute_task(task_name, inputs: {}, timeout: nil, max_retries: nil) ⇒ Hash

Execute a task by name with given inputs

This is the main entry point called from MainDefinition blocks. Routes to neural or symbolic execution based on task implementation. Includes timeout, retry logic, and comprehensive error handling.

Parameters:

  • task_name (Symbol) —

    Name of the task to execute

  • inputs (Hash) (defaults to: {}) —

    Input parameters for the task

  • timeout (Numeric) (defaults to: nil) —

    Override timeout for this task (seconds)

  • max_retries (Integer) (defaults to: nil) —

    Override max retries for this task

Returns:

  • (Hash) —

    Validated output from the task

Raises:

  • (ArgumentError) —

    If task not found or inputs invalid

  • (TaskExecutionError) —

    If task execution fails after retries



106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
 
# File 'lib/language_operator/agent/task_executor.rb', line 106

def execute_task(task_name, inputs: {}, timeout: nil, max_retries: nil)
  execution_start = Time.now
  max_retries ||= @config[:max_retries]

  # Reset JSON parsing retry flag for this task
  @parsing_retry_attempted = false

  with_span('task_executor.execute_task', attributes: build_task_execution_attributes(task_name, inputs, max_retries)) do
    # Fast task lookup using pre-built cache
    task_name_sym = task_name.to_sym
    task_info = @task_cache[task_name_sym]
    raise ArgumentError, "Task not found: #{task_name}. Available tasks: #{@tasks.keys.join(', ')}" unless task_info

    task = task_info[:definition]
    task_type = task_info[:type]

    # Use cached timeout if not explicitly provided
    timeout ||= @task_timeouts[task_name_sym]

    # Optimize logging - only log if debug level enabled or log_executions is true
    if logger.logger.level <= ::Logger::DEBUG || @config[:log_executions]
      logger.info('Executing task',
                  task: task_name,
                  type: task_type,
                  timeout: timeout,
                  max_retries: max_retries,
                  inputs: summarize_values(inputs))
    end

    # Add timeout to span attributes after it's determined
    OpenTelemetry::Trace.current_span&.set_attribute('task.timeout', timeout)

    # Execute with retry logic
    result = execute_with_retry(task, task_name, inputs, timeout, max_retries, execution_start)

    # Add task outputs to span for learning system (if enabled)
    current_span = OpenTelemetry::Trace.current_span
    current_span&.set_attribute('task.outputs', result.to_json) if current_span && capture_enabled?(:outputs)


    result
  end
rescue ArgumentError => e
  # Validation errors should not be retried - re-raise immediately
  log_task_error(task_name, e, :validation, execution_start)
  raise TaskValidationError.new(task_name, e.message, e)
rescue TaskValidationError => e
  # TaskValidationError from validate_inputs should be logged as :validation
  log_task_error(task_name, e, :validation, execution_start)
  raise e
rescue StandardError => e
  # Catch any unexpected errors that escaped retry logic
  log_task_error(task_name, e, :system, execution_start)
  raise create_appropriate_error(task_name, e)
end

#execute_tool(tool_name, params = {}) ⇒ Object

Helper method for symbolic tasks to execute tools

Executes an MCP tool directly through the agent's MCP clients.

Parameters:

  • tool_name (Symbol, String) —

    Name of the tool to execute

  • params (Hash) (defaults to: {}) —

    Tool parameters

Returns:

  • (Object) —

    Tool response (parsed from tool result)



277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
 
# File 'lib/language_operator/agent/task_executor.rb', line 277

def execute_tool(tool_name, params = {})
  tool_name_str = tool_name.to_s

  logger.info('Tool call initiated by symbolic task',
              tool: tool_name_str,
              params: summarize_values(params))

  # Find the tool across all MCP clients
  tool = @agent.tools.find { |t| t.name == tool_name_str }
  raise ArgumentError, "Tool '#{tool_name_str}' not found" unless tool

  # Execute the tool (it's a Proc/lambda wrapped by RubyLLM)
  result = tool.call(**params)

  # Extract text from MCP Content objects
  text_result = if result.is_a?(RubyLLM::MCP::Content)
                  result.text
                elsif result.respond_to?(:map) && result.first.is_a?(RubyLLM::MCP::Content)
                  result.map(&:text).join
                else
                  result
                end

  logger.debug('Tool call completed',
               tool: tool_name_str,
               result_preview: text_result.is_a?(String) ? text_result[0..200] : text_result.class.name)

  # Try to parse JSON response if it looks like JSON
  if text_result.is_a?(String) && (text_result.strip.start_with?('{') || text_result.strip.start_with?('['))
    JSON.parse(text_result, symbolize_names: true)
  else
    text_result
  end
rescue JSON::ParserError
  # Not JSON, return as-is
  text_result
end