Method Execute

flow/flow.go:216–302

Execute runs the flow once with the given input data. This is called automatically on each broker event, but can also be invoked directly for testing or one-shot use.

(ctx context.Context, data string)

// called automatically on each broker event, but can also be
// invoked directly for testing or one-shot use.
func (f *Flow) Execute(ctx context.Context, data string) error {
	ctx, cancel := f.withTimeout(ctx)
	defer cancel()

	// Stepped flows run the ordered, checkpointed step loop.
	if len(f.opts.Steps) > 0 {
		_, err := f.startRun(ctx, data)
		return err
	}

	runID := uuid.New().String()
	info, _ := ai.RunInfoFrom(ctx)
	info.RunID = runID
	info.Flow = f.name
	ctx = ai.WithRunInfo(ctx, info)

	start := time.Now()

	prompt := data
	if f.tmpl != nil {
		var buf bytes.Buffer
		_ = f.tmpl.Execute(&buf, map[string]string{"Data": data})
		prompt = buf.String()
	}

	result := Result{
		FlowName:  f.name,
		Trigger:   f.opts.TriggerTopic,
		Prompt:    prompt,
		Timestamp: start,
	}

	// Flow triggers, Agent reasons: hand the event to the named agent.
	if f.opts.Agent != "" {
		reply, err := f.callAgent(ctx, f.opts.Agent, prompt)
		result.Duration = time.Since(start).Seconds()
		if err != nil {
			result.Error = err.Error()
			result.ErrorKind = string(ai.ClassifyError(err))
			f.record(result)
			return err
		}
		result.Reply = reply
		f.record(result)
		f.log.Logf(logger.InfoLevel, "Flow %s dispatched to agent %s in %.1fs",
			f.name, f.opts.Agent, result.Duration)
		return nil
	}

	// Otherwise run a single augmented-LLM step with the services as tools.
	discovered, err := f.toolSet.Discover()
	if err != nil {
		result.Duration = time.Since(start).Seconds()
		result.Error = err.Error()
		result.ErrorKind = string(ai.ClassifyError(err))
		f.record(result)
		return fmt.Errorf("discover tools: %w", err)
	}