fenn.agents

class fenn.agents.BaseNode

Bases: object

__init__()

exec(prep_res)

Execute the node logic.

post(shared, prep_res, exec_res)

Post-process execution results.

prep(shared)

Prepare data for execution.

run(shared)

set_params(params)

class fenn.agents.Flow(start=None)

Bases: BaseNode

A directed-graph orchestrator for chaining Node executions.

Flow manages the traversal of a directed graph of nodes. Each node returns an action string after execution, and Flow uses that action to look up the next node via the successor mappings created with connect(). The graph can branch, terminate explicitly, or warn on missing transitions.

Parameters:

start (Node or None) – The first node to execute when the flow is run. Can be set later via start().

__init__(start=None)

Initialize the flow with an optional start node.

Parameters:

start (Node or None) – The first node to execute, or None if start() will be called later.

connect(src, dst, action='default')

Wire a transition from src to dst triggered by action.

When src returns action from its lifecycle methods, the flow will move to dst. Passing None for dst marks the path as an explicit terminal transition.

Parameters:

• src (Node) – The source node whose return value selects the destination.

• dst (Node or None) – The next node to execute, or None to terminate the flow.

• action (str) – The action string returned by the source node that triggers this transition. Default: "default".

Returns:

self, enabling method chaining.

Return type:

Flow

get_next_node(curr, action)

Resolve the next node given the current node and action.

Parameters:

• curr (Node) – The currently executing node.

• action (str or None) – The action returned by the current node’s lifecycle.

Returns:

The next node to execute, or None if the flow should terminate.

Return type:

Node or None

post(shared, prep_res, exec_res)

Post-process execution results.

Returns the result of the final node’s execution unchanged. Override this method when subclasses need to inspect or transform the flow-level result.

Parameters:

• shared (dict) – Mutable state shared across all nodes in the flow.

• prep_res (Any) – The value returned by prep().

• exec_res (Any) – The action string returned by the final node’s lifecycle.

Returns:

The final action from the last executed node.

Return type:

Any

start(start)

Set the start node for the flow.

Parameters:

start (Node) – The node that will be executed first.

Returns:

The start node, enabling method chaining.

Return type:

Node

class fenn.agents.LLMClient(provider=None, model=None, api_key=None, api_key_env=None, base_url=None)

Bases: object

Unified LLM client supporting all major providers via an OpenAI-compatible API.

Parameters:

• provider (str, optional) – Provider name (e.g. “openai”, “anthropic”, “openrouter”, “ollama”). Auto-detected from model name or base_url when omitted.

• model (str, optional) – Model identifier. Defaults to the provider’s recommended default.

• api_key (str, optional) – API key. Takes priority over api_key_env and environment lookup.

• api_key_env (str, optional) – Environment variable name to read the API key from. Overrides the provider’s default env var (e.g. OPENROUTER_API_KEY).

• base_url (str, optional) – Custom API base URL. Overrides the provider’s default endpoint.

__init__(provider=None, model=None, api_key=None, api_key_env=None, base_url=None)

Parameters:

• provider (str | None)

• model (str | None)

• api_key (str | None)

• api_key_env (str | None)

• base_url (str | None)

Return type:

None

ask(prompt, schema=None, retries=3)

Send a single prompt and return the response.

Parameters:

• prompt (str) – The user message to send.

• schema (pydantic.BaseModel, optional) – If provided, validates the response against this schema.

• retries (int) – Retry attempts on rate limit errors.

Return type:

str or pydantic.BaseModel

chat_complete(messages, schema=None, retries=3)

Call the chat completions API with a list of message dicts.

Parameters:

• messages (list of dict) – Messages in OpenAI format: [{“role”: “user”, “content”: “…”}].

• schema (pydantic.BaseModel, optional) – If provided, instructs the model to return JSON matching this schema.

• retries (int) – Number of retry attempts on rate limit errors.

Return type:

str or pydantic.BaseModel

stream(prompt)

Send a prompt and yield response tokens one by one.

Parameters:

prompt (str) – The user message to send.

Yields:

str – Individual tokens from the LLM response.

Return type:

Iterator[str]

class fenn.agents.Node(max_retries=1, wait=0)

Bases: BaseNode

A retryable unit of work in a Flow.

Wraps the prep/exec/post lifecycle from BaseNode with automatic retry support around the exec step. If all retry attempts are exhausted, execution falls back to exec_fallback instead of propagating the exception.

Parameters:

• max_retries (int) – Maximum number of attempts to run exec before giving up and calling exec_fallback. Default: 1 (no retries).

• wait (int or float) – Seconds to sleep between failed attempts. Default: 0.

__init__(max_retries=1, wait=0)

Initialize the node’s retry configuration.

Parameters:

• max_retries (int) – Maximum number of attempts to run exec before giving up. Default: 1.

• wait (int or float) – Seconds to sleep between failed attempts. Default: 0.

exec_fallback(prep_res, exc)

Handle the final exec failure after all retries are exhausted.

Default behavior is to re-raise the exception. Override this to return a fallback result instead of failing.

Parameters:

• prep_res (Any) – The result returned by prep, passed through to exec.

• exc (Exception) – The exception raised by the final failed exec attempt.

Returns:

Fallback result to use in place of a successful exec call.

Return type:

Any

Raises:

Exception – Re-raises exc by default.

class fenn.agents.RAGNode(sources=None, query_key='query', context_key='rag_context', chunks_key='rag_chunks', top_k=5, next_action='default', faiss=False, embedding_provider='local', embedding_model='all-MiniLM-L6-v2', embedding_api_key=None, chunk_mode='smart', persist_path=None)

Bases: Node

Flow node that retrieves relevant context from indexed sources.

Loads and indexes all sources once at construction time, then per run queries the index using shared[query_key] and writes the results into shared[chunks_key] and shared[context_key].

Parameters:

• sources (str or list of str, optional) – File paths, folder paths, or URLs to load and index on init. Additional sources can be indexed later with add_source().

• query_key (str) – Key in shared that holds the user query. Default: "query".

• context_key (str) – Key written into shared with the concatenated chunk text. Default: "rag_context".

• chunks_key (str) – Key written into shared with the raw list of chunks. Default: "rag_chunks".

• top_k (int) – Maximum number of chunks to retrieve. Default: 5.

• next_action (str) – Action string returned by post(), used by Flow.get_next_node(). Default: "default".

• faiss (bool) – Use FAISS semantic search instead of BM25. Default: False.

• embedding_provider (str) – Embedding provider (only used when faiss=True). Default: "local".

• embedding_model (str) – Embedding model (only used when faiss=True). Default: "all-MiniLM-L6-v2".

• embedding_api_key (str, optional) – API key for the embedding provider.

• chunk_mode (str) – Document chunking strategy. One of "smart", "paragraphs", "sentences", "fixed". Default: "smart".

• persist_path (str or Path, optional) – Directory to save/load the FAISS index. Only used when faiss=True.

__init__(sources=None, query_key='query', context_key='rag_context', chunks_key='rag_chunks', top_k=5, next_action='default', faiss=False, embedding_provider='local', embedding_model='all-MiniLM-L6-v2', embedding_api_key=None, chunk_mode='smart', persist_path=None)

Initialize the node’s retry configuration.

Parameters:

• max_retries (int) – Maximum number of attempts to run exec before giving up. Default: 1.

• wait (int or float) – Seconds to sleep between failed attempts. Default: 0.

• sources (str | list[str] | None)

• query_key (str)

• context_key (str)

• chunks_key (str)

• top_k (int)

• next_action (str)

• faiss (bool)

• embedding_provider (str)

• embedding_model (str)

• embedding_api_key (str | None)

• chunk_mode (str)

• persist_path (str | None)

Return type:

None

add_source(source)

Index an additional source. Returns self for chaining.

Parameters:

source (str)

Return type:

RAGNode

exec(query)

Execute the node logic.

Parameters:

query (str)

Return type:

list[str]

post(shared, query, chunks)

Post-process execution results.

Parameters:

• shared (dict[str, Any])

• query (str)

• chunks (list[str])

Return type:

str

prep(shared)

Prepare data for execution.

Parameters:

shared (dict[str, Any])

Return type:

str