Jido.AI.Reasoning.GraphOfThoughts.Machine (Jido AI v2.2.0)

Copy Markdown View Source

Pure state machine for the Graph-of-Thoughts (GoT) reasoning pattern.

This module implements state transitions for a GoT agent without any side effects. It uses Fsmx for state machine management and returns directives that describe what external effects should be performed.

Overview

Graph-of-Thoughts extends Tree-of-Thoughts by allowing nodes to have multiple parents and supporting merging/aggregation of thoughts. This enables more complex reasoning patterns like combining insights from different branches.

States

  • :idle - Initial state, waiting for a prompt
  • :generating - Generating a thought for the current context
  • :connecting - Finding connections between nodes
  • :aggregating - Aggregating multiple nodes into one
  • :completed - Final state, solution found
  • :error - Error state

Graph Structure

The graph is stored as nodes and edges:

%{
  nodes: %{
    "node_1" => %{
      id: "node_1",
      content: "Initial problem...",
      score: nil,
      depth: 0,
      metadata: %{}
    },
    ...
  },
  edges: [
    %{from: "node_1", to: "node_2", type: :generates},
    %{from: "node_2", to: "node_3", type: :refines},
    ...
  ]
}

Edge Types

  • :generates - Parent thought generates child thought
  • :refines - Refinement of existing thought
  • :aggregates - Multiple thoughts aggregated into one
  • :connects - Conceptual connection between thoughts

Usage

The machine is used by the GoT strategy:

machine = Machine.new()
{machine, directives} = Machine.update(machine, {:start, prompt, call_id}, env)

All state transitions are pure - side effects are described in directives.

Status Type Boundary

Internal (Machine struct): Status is stored as strings ("idle", "completed") due to Fsmx library requirements.

External (Strategy state, Snapshots): Status is converted to atoms (:idle, :completed) via to_map/1 before storage in agent state.

Never compare machine.status directly with atoms - use Machine.to_map/1 first.

Summary

Types

External status (atom) - used in strategy state after to_map/1 conversion

Internal machine status (string) - required by Fsmx library

t()

Functions

Adds an edge between two nodes.

Adds a node to the graph.

Returns the default prompt for aggregation.

Returns the default prompt for finding connections.

Returns the default system prompt for thought generation.

Finds the best leaf node by score.

Finds all leaf nodes (nodes with no outgoing edges).

Creates a machine from a map (e.g., from strategy state storage).

Generates a unique call ID for LLM requests.

Generates a unique node ID.

Gets all ancestor node IDs (parents, grandparents, etc.).

Gets all child node IDs for a node.

Gets all descendant node IDs (children, grandchildren, etc.).

Gets all edges to a node.

Gets a node by ID.

Gets all nodes in the graph.

Gets all edges from a node.

Gets all parent node IDs for a node.

Detects if there's a cycle in the graph.

Creates a new machine with default state.

Returns the current status as an atom.

Converts the machine state to a map suitable for strategy state storage.

Traces the path from root to a node.

Updates the machine state based on an incoming message.

Types

aggregation_strategy()

@type aggregation_strategy() :: :voting | :weighted | :synthesis

edge()

@type edge() :: %{from: String.t(), to: String.t(), type: edge_type()}

edge_type()

@type edge_type() :: :generates | :refines | :aggregates | :connects

external_status()

@type external_status() ::
  :idle | :generating | :connecting | :aggregating | :completed | :error

External status (atom) - used in strategy state after to_map/1 conversion

internal_status()

@type internal_status() :: String.t()

Internal machine status (string) - required by Fsmx library

t()

@type t() :: %Jido.AI.Reasoning.GraphOfThoughts.Machine{
  aggregation_strategy: aggregation_strategy(),
  current_call_id: String.t() | nil,
  current_node_id: String.t() | nil,
  edges: [edge()],
  generation_count: non_neg_integer(),
  max_depth: pos_integer(),
  max_nodes: pos_integer(),
  nodes: %{required(String.t()) => thought_node()},
  pending_node_ids: [String.t()],
  pending_operation: atom() | nil,
  prompt: String.t() | nil,
  result: term(),
  root_id: String.t() | nil,
  started_at: integer() | nil,
  status: internal_status(),
  streaming_text: String.t(),
  termination_reason: termination_reason(),
  usage: usage()
}

termination_reason()

@type termination_reason() :: :success | :error | :max_nodes | :max_depth | nil

thought_node()

@type thought_node() :: %{
  id: String.t(),
  content: String.t(),
  score: float() | nil,
  depth: non_neg_integer(),
  metadata: map()
}

usage()

@type usage() :: %{
  optional(:input_tokens) => non_neg_integer(),
  optional(:output_tokens) => non_neg_integer(),
  optional(:total_tokens) => non_neg_integer()
}

Functions

add_edge(machine, from_id, to_id, type)

@spec add_edge(t(), String.t(), String.t(), edge_type()) :: t()

Adds an edge between two nodes.

add_node(machine, node)

@spec add_node(t(), thought_node()) :: t()

Adds a node to the graph.

before_transition(struct, from, to, state_field)

default_aggregation_prompt()

@spec default_aggregation_prompt() :: String.t()

Returns the default prompt for aggregation.

default_connection_prompt()

@spec default_connection_prompt() :: String.t()

Returns the default prompt for finding connections.

default_generation_prompt()

@spec default_generation_prompt() :: String.t()

Returns the default system prompt for thought generation.

find_best_leaf(machine)

@spec find_best_leaf(t()) :: thought_node() | nil

Finds the best leaf node by score.

find_leaves(machine)

@spec find_leaves(t()) :: [thought_node()]

Finds all leaf nodes (nodes with no outgoing edges).

from_map(map)

@spec from_map(map()) :: t()

Creates a machine from a map (e.g., from strategy state storage).

generate_call_id()

@spec generate_call_id() :: String.t()

Generates a unique call ID for LLM requests.

generate_node_id()

@spec generate_node_id() :: String.t()

Generates a unique node ID.

get_ancestors(machine, node_id)

@spec get_ancestors(t(), String.t()) :: [String.t()]

Gets all ancestor node IDs (parents, grandparents, etc.).

get_children(machine, node_id)

@spec get_children(t(), String.t()) :: [String.t()]

Gets all child node IDs for a node.

get_descendants(machine, node_id)

@spec get_descendants(t(), String.t()) :: [String.t()]

Gets all descendant node IDs (children, grandchildren, etc.).

get_incoming_edges(machine, node_id)

@spec get_incoming_edges(t(), String.t()) :: [edge()]

Gets all edges to a node.

get_node(machine, node_id)

@spec get_node(t(), String.t()) :: thought_node() | nil

Gets a node by ID.

get_nodes(machine)

@spec get_nodes(t()) :: [thought_node()]

Gets all nodes in the graph.

get_outgoing_edges(machine, node_id)

@spec get_outgoing_edges(t(), String.t()) :: [edge()]

Gets all edges from a node.

get_parents(machine, node_id)

@spec get_parents(t(), String.t()) :: [String.t()]

Gets all parent node IDs for a node.

has_cycle?(machine)

@spec has_cycle?(t()) :: boolean()

Detects if there's a cycle in the graph.

new(opts \\ [])

@spec new(keyword()) :: t()

Creates a new machine with default state.

status(machine)

@spec status(t()) :: external_status()

Returns the current status as an atom.

to_map(machine)

@spec to_map(t()) :: map()

Converts the machine state to a map suitable for strategy state storage.

trace_path(machine, node_id)

@spec trace_path(t(), String.t()) :: [String.t()]

Traces the path from root to a node.

update(machine, message, env \\ %{})

@spec update(t(), term(), map()) :: {t(), list()}

Updates the machine state based on an incoming message.

Returns {updated_machine, directives} where directives describe external effects to perform.

Messages

  • {:start, prompt, call_id} - Start GoT reasoning
  • {:llm_result, call_id, result} - Handle LLM response
  • {:llm_partial, call_id, delta, chunk_type} - Handle streaming