Current section
Files
Jump to
Current section
Files
README.md
# Hayata
[](https://hex.pm/packages/hayata)
[](https://github.com/zgbjgg/hayata/actions/workflows/erlang.yml)
[](https://codecov.io/gh/zgbjgg/hayata)
[](https://raw.githubusercontent.com/zgbjgg/hayata/master/LICENSE)
[](https://hex.pm/packages/hayata)
[](https://hex.pm/packages/hayata)
Hayata is an Erlang/OTP framework for building AI agents around a
graph-based execution model. It uses OTP processes, supervision,
monitoring, and message passing to coordinate agent execution, with
explicit ownership of process state and durable recovery checkpoints.
Hayata is designed to be usable from Erlang and Elixir applications. Its
graph-oriented model is inspired by agent orchestration frameworks such
as LangGraph, while exposing an Erlang/OTP-native runtime rather than a
feature-for-feature clone of another API.
## Highlights
- Graph-based agent execution with a configurable decision policy.
- Sequential and parallel node execution.
- OTP supervision and process monitoring for agents, nodes, edges,
reducers, and edge watchers.
- Asynchronous invocation APIs and result notifications.
- Invocation checkpoints and task-result persistence in Mnesia.
- Recovery of interrupted invocations, including parallel transitions.
- Conservative handling of uncertain tasks: a task marked as running
without a durable result is not silently re-executed.
- Telemetry lifecycle events for agents, nodes, edges, and reducers.
- Provider-agnostic RAG retrieval with normalized document results,
usable directly from nodes or through LLM tool calling.
## Architecture
- **Agent** --- coordinates invocations, applies the decision policy,
tracks invocation state, and initiates recovery.
- **Node** --- a supervised `gen_server` that owns local execution
state, coordinates outgoing edges, tracks executions, and monitors
edges and reducers.
- **Edge** --- an anonymous `gen_server` that runs an edge task and
reports its terminal result.
- **Edge watcher** --- monitors the node and its edge. If the node
terminates unexpectedly, the watcher terminates the edge so it does
not remain orphaned.
- **Reducer** --- a temporary process that collects edge results for
an execution and produces a consolidated outcome.
- **Task store** --- persists invocation checkpoints and task outcomes
in Mnesia.
- **Node supervisor** --- supervises node processes and restarts them
after abnormal termination.
## Requirements
- Erlang/OTP 29.1.1 (target runtime)
- Rebar3
## Build and test
``` sh
rebar3 compile
rebar3 eunit
```
## Quick start
The following example mirrors a basic graph-agent pattern: define a
node, define control flow, start the agent's supervision tree, and
invoke it. This example uses a mock node; it does not call an LLM.
``` erlang
%% hello_hayata.erl
-module(hello_hayata).
-export([run/0]).
run() ->
AgentId = hello_agent,
MockNode = fun(State) ->
Message = maps:get(input, State, <<"Hello">>),
#{
messages => [
#{role => ai, content => <<"Hello, world!">>},
#{role => user, content => Message}
]
}
end,
Nodes = #{mock_node => MockNode},
DecisionPolicy = fun(Context) ->
case maps:get(step, Context) of
0 ->
{next, mock_node, <<"Hello">>};
1 ->
{finish, maps:get(state, Context)}
end
end,
Config = #{
id => AgentId,
state => #{},
nodes => Nodes,
decision_policy => DecisionPolicy,
max_steps => 3,
execution_timeout => 5000
},
case hayata_sup:start_agent(Config) of
{ok, AgentSupPid} ->
try
run_agent(AgentSupPid)
after
hayata_sup:stop_agent(AgentSupPid)
end;
{error, Reason} ->
{error, {agent_start_failed, Reason}}
end.
run_agent(AgentSupPid) ->
case hayata_agent_sup:agent(AgentSupPid) of
{ok, AgentPid} ->
case hayata_agent:run(AgentPid, #{input => <<"Hello">>}, #{subscriber => self()}) of
{ok, InvocationId} ->
await_result(InvocationId, 100);
{error, Reason} ->
{error, Reason}
end;
{error, Reason} ->
{error, Reason}
end.
await_result(_InvocationId, 0) ->
{error, notification_timeout};
await_result(InvocationId, Attempts) ->
receive
{hayata_agent_result, InvocationId, {ok, Result}} ->
{ok, Result};
{hayata_agent_result, InvocationId, {error, Reason}} ->
{error, Reason}
after 100 ->
await_result(InvocationId, Attempts - 1)
end.
```
Save the module as `hello_hayata.erl` in a location included in your
Erlang code path, compile it, and call:
``` erlang
hello_hayata:run().
```
## Agent Execution Contract
Hayata uses two different function contracts during agent execution:
- **Node functions** receive the node execution state and return a map of state updates.
- **Decision policies** receive the agent execution context and decide which node executes next, whether multiple nodes should execute in parallel, or whether the invocation should finish.
Both functions receive a single map argument, but their available fields and responsibilities are different.
### Node Function
Every node is defined as a one-argument function:
```erlang
Node = fun(State) ->
%% Read the current state.
Input = maps:get(input, State),
%% Return state updates.
#{
answer => Input,
status => completed
}
end.
```
The `State` argument contains the agent's current shared state, with the current node input added under the `input` key.
Its structure is:
```erlang
#{
%% All fields from the agent's current shared state.
input => NodeInput,
%% Other application-defined state fields.
messages => [],
customer => undefined
}
```
The following fields are available:
| Field | Description |
|---|---|
| `input` | Input explicitly passed to the node by the decision policy. |
| Application-defined fields | All fields currently present in the agent's shared state. |
| `hayata_streamer` | Optional function for publishing partial node results. Available only when invocation streaming is enabled and a subscriber is configured. |
| `hayata_llm` | Optional function for invoking the configured LLM without streaming. Available when an LLM configuration is provided to the agent. |
| `hayata_llm_stream` | Optional function for invoking the configured LLM with streaming. Available when an LLM configuration is provided. Streaming requires invocation streaming to be enabled. |
| `hayata_tool_loop` | Optional function for executing an LLM tool-calling loop. Available when an LLM configuration is provided. |
| `hayata_rag` | Optional function for retrieving relevant documents from a configured RAG provider. Available when a RAG configuration is provided to the agent. |
#### Returning node updates
A node must return a map containing the state updates it wants to publish:
```erlang
Node = fun(State) ->
Input = maps:get(input, State),
#{
response => Input,
processed => true
}
end.
```
Hayata merges the returned map into the current shared agent state. Existing keys with the same name are replaced by the returned values; keys not included in the update remain unchanged.
```erlang
NewState = maps:merge(CurrentState, NodeUpdates).
```
Nodes do not directly mutate the shared agent state. They return updates, and Hayata applies them after successful execution.
#### Node input and state
The input passed to a node is determined by the decision policy:
```erlang
{next, NodeId, NodeInput}
```
Hayata adds this value to the current shared state under the `input` key before invoking the node.
For parallel execution, every selected node receives the same input and the same current shared state. Their results are collected independently and are not merged into shared state automatically.
#### Partial results
When invocation streaming is enabled, a node can publish partial results using `hayata_streamer`:
```erlang
Node = fun(State) ->
Streamer = maps:get(hayata_streamer, State),
ok = Streamer(#{chunk => <<"First part">>}),
ok = Streamer(#{chunk => <<"Second part">>}),
#{completed => true}
end.
```
Partial results are emitted as `node_partial_result` events. They are notifications and do not modify shared agent state.
### Decision Policy
The decision policy is also a one-argument function:
```erlang
DecisionPolicy = fun(Context) ->
case maps:get(step, Context) of
0 ->
{next, first_node, <<"Hello">>};
1 ->
{finish, maps:get(state, Context)}
end
end.
```
The `Context` map provides information about the current agent invocation and its execution progress.
Its structure is:
```erlang
#{
agent_id => AgentId,
state => CurrentSharedState,
input => OriginalInvocationInput,
step => CurrentStep,
max_steps => MaxSteps,
parallel_results => ParallelResults
}
```
The following fields are available:
| Field | Description |
|---|---|
| `agent_id` | Identifier of the running agent. |
| `state` | Current shared agent state, including updates from previously completed nodes. |
| `input` | Original input supplied when starting the invocation with `hayata_agent:run/2` or `run/3`. |
| `step` | Current execution step. Starts at `0` and increases when node executions are scheduled. |
| `max_steps` | Maximum number of node executions allowed for the invocation. |
| `parallel_results` | Ordered list of results from the most recently completed parallel transition. It is `undefined` when no parallel results are available. |
#### Policy decisions
The decision policy must return one of the following values:
**Execute one node**
```erlang
{next, NodeId, Input}
```
Schedules a single node and passes `Input` to its function.
**Execute multiple nodes in parallel**
```erlang
{next, [NodeA, NodeB], Input}
```
Schedules multiple distinct nodes concurrently. All nodes receive the same input and shared state.
When every node completes successfully, their returned update maps become available through `parallel_results`, preserving the order of the requested node identifiers.
**Finish the invocation**
```erlang
{finish, Result}
```
Completes the invocation and publishes `Result` as its final outcome.
**Return an error**
```erlang
{error, Reason}
```
Fails the current invocation with the provided reason.
Any other return value is considered an invalid policy decision.
#### Parallel execution example
```erlang
DecisionPolicy = fun(Context) ->
case maps:get(step, Context) of
0 ->
{next, [customer_node, orders_node], maps:get(input, Context)};
2 ->
[Customer, Orders] = maps:get(parallel_results, Context),
{next, aggregate_node, #{
customer => Customer,
orders => Orders
}};
3 ->
{finish, maps:get(state, Context)}
end
end.
```
The `parallel_results` list follows the order in which the nodes were requested, regardless of their completion order.
### Important distinctions
| Concept | Node function | Decision policy |
|---|---|---|
| Argument | Node execution state | Agent execution context |
| Receives current shared state | Yes | Yes |
| Receives original invocation input | Only through shared state if retained by the application | Yes, through `input` |
| Receives node-specific input | Yes, through `input` | Selects it when returning `next` |
| Receives execution step | No | Yes |
| Receives parallel results | No, unless explicitly passed as node input or included in shared state | Yes |
| Returns | State update map | Execution decision tuple |
This separation keeps node execution focused on processing data and returning state updates, while the decision policy remains responsible for orchestration and control flow.
## Conversational memory
Agents can optionally preserve conversation history between successive
invocations. Enable this behavior with `conversation_memory => true`
in the agent configuration:
```erlang
Config = #{
id => support_agent,
state => #{messages => []},
nodes => Nodes,
decision_policy => DecisionPolicy,
conversation_memory => true
}.
```
When conversational memory is enabled, Hayata carries the `messages`
entry from the agent's current state into the next invocation. This lets
a node build each LLM request from the previous conversation instead of
starting with an empty message history every time.
For example, if a node appends each new user message to the `messages`
list, successive invocations can build a history like
`[FirstMessage, SecondMessage]`. The application is responsible for
appending messages in the format expected by its LLM provider and for
deciding which messages belong in the conversation.
The option defaults to `false`. With the default behavior, every
invocation starts from the configured initial state, so message history
from a previous invocation is not retained. When the option is enabled,
only the `messages` entry is carried over; other state fields are reset
to their configured initial values at the beginning of each invocation.
This feature preserves history in the lifetime of the running agent. It
is not a vector or semantic memory system, and it does not by itself
provide a durable conversation store or guarantee that conversation
history survives an agent or VM restart.
## Invocation API
Hayata supports asynchronous agent invocations with optional result
notifications and event streaming.
### Starting an invocation
Start an invocation without a subscriber:
```erlang
{ok, InvocationId} = hayata_agent:run(AgentPid, Input).
```
The invocation runs asynchronously. Use `hayata_agent:get_result/1`
to retrieve the final outcome when it is available.
To receive a terminal result notification, provide a subscriber PID:
```erlang
{ok, InvocationId} = hayata_agent:run(
AgentPid,
Input,
#{subscriber => self()}
).
```
The subscriber receives one of these messages:
```erlang
{hayata_agent_result, InvocationId, {ok, Result}}
{hayata_agent_result, InvocationId, {error, Reason}}
```
`InvocationId` identifies the invocation. If the invocation is rejected
before it starts, `run/3` returns an error instead of sending a terminal
result notification for that request.
### Streaming events
Enable streaming by setting `stream => true` and providing a subscriber:
```erlang
{ok, InvocationId} = hayata_agent:run(
AgentPid,
Input,
#{
subscriber => self(),
stream => true
}
).
```
The subscriber receives streaming events in this format:
```erlang
{hayata_agent_event, InvocationId, Event}
```
Each `Event` is a map containing:
- `type` — the event type.
- `sequence` — a monotonically increasing sequence number within the
invocation.
- `data` — event-specific information.
The event types are:
| Event type | Meaning | Data |
|---|---|---|
| `invocation_started` | The invocation has started. | Invocation input. |
| `node_started` | A node execution has started. | `node_id`, `execution_id`. |
| `node_partial_result` | A node has emitted a partial result. | `node_id`, `execution_id`, `edge_id`, `result`. |
| `node_completed` | A node execution has completed. | `node_id`, `execution_id`. |
| `invocation_completed` | The invocation completed successfully. | `result`. |
| `invocation_failed` | The invocation failed. | `reason`. |
Sequence numbers start at `1` for each invocation. In parallel
transitions, lifecycle and partial-result events from different nodes
may interleave. Use `node_id` and `execution_id` to associate events
with their corresponding executions.
Streaming events do not replace the terminal result notification.
When streaming is enabled, the subscriber receives both
`hayata_agent_event` messages and the traditional
`hayata_agent_result` notification.
### Emitting partial results from a node
Node functions receive a context map as their single argument. When
streaming is enabled, Hayata injects a `hayata_streamer` function into
that context.
The streamer accepts a map and emits a `node_partial_result` event:
```erlang
Node = fun(Context) ->
Streamer = maps:get(hayata_streamer, Context),
ok = Streamer(#{chunk => first}),
ok = Streamer(#{chunk => second}),
#{answer => 42}
end.
```
A partial-result event has this shape:
```erlang
#{
type => node_partial_result,
sequence => 3,
data => #{
node_id => answer_node,
execution_id => ExecutionId,
edge_id => node_result,
result => #{chunk => first}
}
}
```
The example sequence number is illustrative. Actual sequence numbers
depend on the other events emitted during the invocation.
The streamer accepts maps only. Passing a non-map returns
`{error, invalid_partial_result}`. A partial result from an execution
that is no longer active is rejected rather than emitted as a valid
event.
Partial results are notifications; they do not independently modify
the agent's shared state or replace the node's final result. The node
must still return its final result map as usual.
The streamer synchronously submits the partial result to the agent.
This allows the agent to publish the partial event before processing
the node's terminal completion notification.
### Streaming disabled
Streaming is disabled by default. Existing invocations can continue
to use `run/2` or subscribe to terminal results without enabling
streaming.
When streaming is disabled, `hayata_streamer` is not injected into the
node context and no `hayata_agent_event` messages are emitted.
### Run options
`hayata_agent:run/3` accepts a map with these supported keys:
| Option | Type | Default | Description |
|---|---|---|---|
| `subscriber` | PID | `undefined` | Process that receives terminal notifications and, when enabled, stream events. |
| `stream` | Boolean | `false` | Enables invocation events and partial-result streaming. |
Streaming requires a subscriber. Unknown options, an invalid subscriber,
or an invalid `stream` value return an error.
## Parallel node execution
A decision policy can schedule multiple nodes in parallel by returning
`{next, NodeIds, Input}`, where `NodeIds` is a non-empty list of
distinct node identifiers.
All branch nodes receive the same input and current shared agent state.
Their returned maps are collected independently and do not directly
modify the shared state.
When every branch completes successfully, the policy receives the
results through the `parallel_results` field in its context. Results are
a list of maps in the same order as the requested node identifiers,
regardless of completion order.
``` erlang
DecisionPolicy = fun(Context) ->
case maps:get(step, Context) of
0 ->
{next, [customer, orders], undefined};
2 ->
[Customer, Orders] = maps:get(parallel_results, Context),
{next, aggregate, #{
customer => Customer,
orders => Orders
}};
3 ->
{finish, maps:get(state, Context)}
end
end.
```
The `parallel_results` field is `undefined` before a parallel transition
completes. After all branches succeed, it contains their ordered result
maps and remains available to the policy that chooses the next
transition. When the policy schedules another node or parallel
transition, the previous results are cleared. A new invocation starts
with `parallel_results` set to `undefined`.
If a branch fails or the invocation times out, the transition fails and
pending branches are cancelled; incomplete parallel results are not
published as a successful transition result.
## Durable checkpoints and recovery
Hayata uses Mnesia to persist invocation checkpoints and task outcomes.
The task store uses disk-backed Mnesia tables and synchronous
transactions; completed task results and invocation checkpoints are
flushed to the Mnesia log after writes.
A checkpoint captures the information required to identify and resume an
invocation, including the invocation ID, current state and step, active
execution, parallel results, and terminal result/error fields. When an
agent starts and finds an unfinished invocation for its ID, it restores
that invocation and reports `recovery_required` rather than accepting a
new invocation on top of it.
Call `hayata_agent:recover(AgentPid)` to initiate recovery of the
restored invocation.
Recovery classifies each task using its durable record:
- **Completed:** reuse the persisted task result instead of executing
that task again.
- **Not started:** schedule the task for execution during recovery.
- **Uncertain:** if a task record indicates that execution started but
no completed result exists, recovery returns
`{error, task_execution_uncertain}` rather than silently repeating
the operation.
For parallel execution, Hayata can combine completed branch results with
branches that have not started, execute only the missing branches, and
persist the final invocation result. Results retain the configured node
order.
### Recovery guarantees and limitations
Recovery is designed to avoid silently repeating a task whose completion
is uncertain. It does not provide exactly-once semantics for arbitrary
external side effects. If a node may have called an external service
before the VM stopped but its result was not durably saved, Hayata
cannot know whether that side effect occurred. Applications should use
idempotency keys, external deduplication, or other application-level
safeguards where retry safety matters.
Durable recovery depends on preserving the configured Mnesia data
directory across VM restarts. If that directory is removed or replaced,
the persisted checkpoints are unavailable.
### Reproducing VM restart scenarios
The repository includes [`VM_RECOVERY_TESTS.md`](VM_RECOVERY_TESTS.md),
a step-by-step manual verification guide with complete local commands
for:
1. A completed task result surviving an abrupt VM shutdown.
2. A task left running without a result being classified as uncertain
after restart.
3. A running invocation checkpoint being restored by a new agent
process.
4. A parallel invocation reusing a completed branch and executing a
not-started branch after an abrupt VM shutdown.
These manual integration checks complement the EUnit suite; run both
against the exact commit intended for release.
## Execution lifecycle
1. The node validates an execution request, edge tasks, and timeout.
2. A reducer is started and activated when result consolidation is
required.
3. The node starts and monitors edge processes.
4. Each edge runs its task and sends its result to the reducer and
node.
5. The reducer consolidates the edge results and reports the final
result.
6. The node updates execution status, notifies any configured
recipient, and tracks process termination.
Logical edge IDs and execution IDs correlate messages. Stale
notifications must not overwrite terminal execution state or trigger
duplicate notifications.
## Telemetry
Hayata uses [Telemetry](https://hexdocs.pm/telemetry/) to expose
lifecycle events from agents, nodes, edges, and reducers. Applications
can attach handlers to collect metrics, trace execution activity, or
integrate with monitoring systems.
| Component | Event | Meaning |
|---|---|---|
| Agent | `[hayata, agent, invocation, start]` | An invocation starts. |
| Agent | `[hayata, agent, invocation, stop]` | An invocation completes or fails normally. |
| Agent | `[hayata, agent, invocation, exception]` | An invocation encounters a caught policy exception. |
| Node | `[hayata, node, execution, start]` | A node execution starts. |
| Node | `[hayata, node, execution, stop]` | A node execution reaches a terminal state, such as completion or cancellation. |
| Node | `[hayata, node, execution, exception]` | A node execution ends with a reducer recovery failure. |
| Edge | `[hayata, edge, start]` | An edge task starts. |
| Edge | `[hayata, edge, stop]` | An edge task ends without a caught task exception. |
| Edge | `[hayata, edge, exception]` | An edge task raises a caught exception. |
| Reducer | `[hayata, reducer, start]` | A reducer is activated. |
| Reducer | `[hayata, reducer, stop]` | A reducer completes or times out. |
A `stop` event indicates a terminal outcome, not necessarily success.
Where present, use the `status` metadata to distinguish completed,
failed, cancelled, and timed-out executions.
Start events include `system_time`. Terminal events include `duration`,
calculated with `erlang:monotonic_time/0`. Both use Erlang native time
units; convert durations when exporting to systems that expect
milliseconds.
Common metadata includes `agent_id`, `node_id`, `execution_id`,
`edge_id`, `status`, and, for caught exceptions, `class`. Reducer events
may also include `expected_edges`, `result_count`, `error_count`, and
`pending_edges`. Available fields vary by event.
Hayata's telemetry metadata avoids including full results, exception
reasons, or stacktraces. Applications should also review their own
handlers and exporters to avoid exposing sensitive inputs, state,
prompts, or outputs.
### Attaching a handler
``` erlang
-module(my_hayata_telemetry).
-export([attach/0, detach/0, handle_event/4]).
-define(HANDLER_ID, {?MODULE, agent_invocations}).
attach() ->
telemetry:attach_many(
?HANDLER_ID,
[
[hayata, agent, invocation, start],
[hayata, agent, invocation, stop],
[hayata, agent, invocation, exception]
],
fun ?MODULE:handle_event/4,
#{}
).
detach() ->
telemetry:detach(?HANDLER_ID).
handle_event(Event, Measurements, Metadata, _Config) ->
logger:info(
"Hayata telemetry event",
#{
event => Event,
measurements => Measurements,
metadata => Metadata
}
).
```
Call `my_hayata_telemetry:attach/0` during application startup and
`my_hayata_telemetry:detach/0` when the handler is no longer needed.
Keep handlers lightweight; forward expensive processing to a dedicated
process or metrics backend.
## Fault handling and process supervision
The edge watcher monitors the node independently. If the node dies, including through an untrappable `kill`, the watcher terminates the edge so it does not remain orphaned. Node processes are supervised and restarted after abnormal termination.
Edge tasks execute in linked worker processes, keeping the edge GenServer responsive while a task is running. If the edge terminates abnormally, its linked worker is terminated as well. However, terminating a process cannot undo external side effects that have already occurred.
If a task may have affected an external service before its result was durably saved, recovery reports `task_execution_uncertain` and does not automatically re-execute the task. Applications should use idempotency keys, external deduplication, or other application-level safeguards where retry safety matters.
## Design principles
- Use OTP processes and monitoring for concurrency and failure
detection.
- Keep process responsibilities and state ownership explicit.
- Treat logical IDs separately from process PIDs.
- Keep cancellation and recovery scoped to the affected execution.
- Persist task outcomes before treating them as completed.
- Reject uncertain execution states rather than silently repeating
potentially non-idempotent work.
- Validate behavior incrementally with EUnit and explicit VM-restart
checks.
## Retrieval-Augmented Generation (RAG)
Hayata provides a provider-agnostic retrieval interface through the
`hayata_rag` module. It allows nodes to retrieve relevant documents
from an application-defined retrieval provider without coupling the
framework to a particular vector database, embedding service, or
search engine.
Hayata does not implement document indexing, chunking, embedding
generation, or vector storage. Applications integrate their own
retrieval infrastructure and provide a retriever function that
implements the Hayata contract.
### Configuring a retriever
Configure RAG using the `rag_config` key in the agent configuration.
The `retrieve` entry accepts either an anonymous function with arity
two or an MFA tuple pointing to an exported function with arity two.
The retriever receives a query and an options map. It must return
`{ok, #{documents := Documents}}` or `{error, Reason}`.
```erlang
Retriever = fun(Query, Options) ->
%% Call your own retrieval infrastructure.
my_retriever:search(Query, Options)
end,
Config = #{
id => knowledge_agent,
state => #{},
nodes => Nodes,
decision_policy => DecisionPolicy,
rag_config => #{
retrieve => Retriever
}
}.
```
An MFA retriever can also be configured:
```erlang
rag_config => #{
retrieve => {my_retriever, search}
}
```
The provider is responsible for translating its native response into
the normalized Hayata document contract.
### Normalized document contract
Every retrieved document must contain:
| Field | Type | Required | Description |
|---|---|---|---|
| `id` | Binary | Yes | Non-empty document or chunk identifier. |
| `content` | Binary | Yes | Retrieved text content. |
| `metadata` | Map | Yes | Application-defined document metadata. |
| `score` | Number | No | Retrieval relevance score, when available. |
A valid retriever response looks like this:
```erlang
{ok, #{
documents => [
#{
id => <<"chunk-001">>,
content => <<"Erlang uses lightweight processes.">>,
metadata => #{
source => <<"erlang-guide.pdf">>,
page => 12
},
score => 0.95
}
]
}}
```
The score is optional because retrieval providers do not necessarily
expose comparable relevance scores.
The `hayata_rag` module validates the retriever configuration and
normalizes its result contract. Invalid responses are rejected, while
provider errors are propagated.
### Using RAG directly from a node
When `rag_config` is configured, Hayata injects a `hayata_rag`
function into every node execution context.
The function accepts a binary query and an options map:
```erlang
Rag(Query, Options)
```
It returns either a normalized document result or an error.
```erlang
RetrieveNode = fun(Context) ->
Rag = maps:get(hayata_rag, Context),
case Rag(<<"Erlang process model">>, #{limit => 5}) of
{ok, #{documents := Documents}} ->
#{
retrieved_documents => Documents
};
{error, Reason} ->
#{
retrieval_error => Reason
}
end
end.
```
RAG retrieval is independent of LLM configuration. An agent can
retrieve and process documents without configuring an LLM provider.
### Using RAG through LLM tool calling
RAG can also be exposed as a tool to an LLM through
`hayata_tool_loop`. This allows the model to decide when retrieval is
necessary.
The tool handler delegates to the same injected `hayata_rag` function.
There is no separate retrieval implementation or RAG-specific tool
runtime.
```erlang
RagNode = fun(Context) ->
Rag = maps:get(hayata_rag, Context),
ToolLoop = maps:get(hayata_tool_loop, Context),
SearchTool = #{
name => <<"search_knowledge">>,
description =>
<<"Searches the knowledge base for relevant documents.">>,
parameters => #{
type => object,
properties => #{
query => #{
type => string,
description => <<"The information to search for.">>
}
},
required => [<<"query">>]
},
handler => fun(Arguments) ->
Rag(
maps:get(query, Arguments),
#{limit => 5}
)
end
},
Request = #{
messages => [
#{
role => user,
content => [
#{
type => text,
text => <<"Explain the system architecture.">>
}
]
}
]
},
case ToolLoop(Request, [SearchTool], 5) of
{ok, Result} ->
#{answer => maps:get(response, Result)};
{error, Reason} ->
#{error => Reason}
end
end.
```
This example requires both `rag_config` and an LLM configuration in
the agent. The `hayata_tool_loop` capability is available when an LLM
is configured.
### Design boundaries
Hayata deliberately keeps retrieval separate from generation:
- `hayata_rag` handles document retrieval and result validation.
- `hayata_llm` handles text generation.
- `hayata_tool_loop` coordinates LLM tool calling and execution.
- Applications own their retrieval infrastructure, indexing,
embeddings, and document lifecycle.
This separation allows applications to use different retrieval
technologies without requiring provider-specific adapters in the
Hayata core.
## LLM
Hayata's LLM integration is designed around a provider-neutral adapter contract. The shared contract defines the normalized request and response shapes, capability reporting, error codes, and optional streaming interface. Provider-specific API payloads and HTTP details stay inside each adapter.
The `hayata_tool_loop` coordinates model calls and tool execution independently of the provider: it validates normalized assistant responses, executes registered tools in order, adds tool results to the conversation history, and continues until the model returns a final response or the iteration limit is reached. Providers should return normalized tool calls; they should not execute Hayata tool handlers themselves.
### Available adapters
- **OpenAI** — [OpenAI Responses API adapter documentation](docs/llm/openai.md). Supports normalized text generation, tool calling, and streaming.
### Implementing another provider
Use the [Adding an LLM provider guide](docs/llm/adding-a-provider.md) as the implementation checklist. It covers the adapter contract, normalized errors, tool-call serialization, streaming guarantees, and the test coverage expected before a provider is listed here.
## License
Apache-2.0. See [LICENSE.md](LICENSE.md) for details.