Agent Module#

class isek.agent.abstract_agent.AbstractAgent(persona: Persona | None = None, model: AbstractModel | None = None, tools: List[Callable] | None = None, deepthink_enabled: bool = False, **kwargs)[source]#

Bases: ABC

An abstract base class for creating intelligent agents.

This class provides a foundational structure for agents, including persona management, interaction with language models, memory, and tool usage.

abstract build(daemon=False)[source]#

Abstract method to build or set up the agent.

This method should be implemented by subclasses to perform any necessary initialization or setup before the agent starts running.

Parameters:: daemon (bool) -- If True, the agent might be set up to run as a daemon process. Defaults to False.

decompose_task(command: str) → str[source]#

Decomposes a given command or task into a list of subtasks using the LLM.

This method prompts the language model to break down a complex task into a list of high-level subtasks. The subtasks are expected to be returned by the model as a Python list string.

Parameters:: command (str) -- The command or task to be decomposed.
Returns:: A string representation of a list of subtasks (e.g., '["Subtask 1", "Subtask 2"]').
Return type:: str

hearbeat()[source]#

Executes a heartbeat cycle for the agent.

This function is intended to be called periodically to keep the agent active or to perform routine tasks when no explicit input is given. It calls the run() method with no input.

response(input: str) → str[source]#

Generates a response to the given input.

This method orchestrates the agent's core logic:

Logs the trigger (input or heartbeat).
Stores the user input in memory (if provided).
Builds prompt templates incorporating persona, system instructions, tools, and conversation history.
Optionally performs a "deep think" step to decompose the task.
Enters a loop to interact with the language model: a. Sends the current messages and available tools to the model. b. Appends the model's response to the message history. c. If the response contains content, stores it in memory. d. If there are no tool calls, returns the content. e. If there are tool calls, executes each tool, appends the result to messages, and continues the loop.

Parameters:: input (str) -- User instructions or environment signals, such as text. The method internally handles cases where input might be None or empty, though its type hint is str.
Returns:: The agent's textual response to the input.
Return type:: str

run(input: str | None = None) → None[source]#

Runs a single cycle of the agent's operation.

The agent processes the input, generates a response, and logs the interaction.

Note

The type hint for the return value is None, but the implementation returns a string response. The docstring reflects the implementation. Consider updating the type hint to -> str for consistency.

Parameters:: input (Optional[str]) -- User instructions or environment signals, such as text. Defaults to None.
Returns:: The agent's response to the input.
Return type:: str

run_cli()[source]#

Runs the agent in a command-line interface (CLI) mode.

This method continuously prompts the user for input and processes it using the run() method until the user types "exit".

class isek.agent.distributed_agent.DistributedAgent(**kwargs: Any)[source]#

Bases: AbstractAgent, Node

Represents an agent that operates within a distributed network of nodes.

This class extends AbstractAgent with capabilities from Node, allowing it to communicate and interact with other nodes in a peer-to-peer or distributed fashion. It uses its persona to define its identity and introduction within the network.

build(daemon: bool = False) → None[source]#

Builds and starts the server component of the distributed agent.

If daemon is False, the server is built and run in the current thread, potentially blocking further execution. If daemon is True, the server is started in a separate daemon thread, allowing the main program to continue.

This method typically calls self.build_server() from the Node base class.

Parameters:: daemon (bool) -- If True, run the server in a daemon thread. Defaults to False.

build_node_id() → str[source]#

Generates the unique identifier for this node in the distributed network.

The node ID is derived from the agent's persona name. This method likely overrides a method from the Node base class.

Returns:: The unique node identifier.
Return type:: str

metadata() → Dict[str, str][source]#

Provides metadata about this agent/node.

This metadata can be shared with other nodes in the network to identify and describe this agent. It includes the agent's name and introduction. This method likely overrides a method from the Node base class.

Returns:: A dictionary containing metadata, specifically "name" and "intro".
Return type:: Dict[str, str]

on_message(sender: Any, message: str) → str | None[source]#

Handles incoming messages from other nodes in the network.

This method is a callback invoked by the Node infrastructure when a message is received. It logs the message and then processes it using the agent's main run() logic.

Parameters:

sender (Any) -- The identifier of the sending node. The exact type depends on the node implementation.
message (str) -- The content of the message received.

Returns:

The agent's response to the message, or None if the run method (or its overridden version) returns None.

Return type:

Optional[str]

search_partners(query: str) → str[source]#

Searches for suitable partner nodes in the network to help answer a query.

This method is used when the agent determines it might lack the necessary knowledge or capabilities to handle a query on its own. It first retrieves a list of potential nodes based on a vector search (using get_nodes_by_vector()) and then uses the language model to select the single best-fitting partner from this list based on the query and the partners' introductions.

Parameters:: query (str) -- The query or task for which partners are being sought.
Returns:: A string containing the name, node_id, and reason for selecting the best-fitting partner, formatted as "name: XXX, node_id: XXX, Reason: XXX". Returns an empty string or an error message if no suitable partner is found or if the LLM call fails.
Return type:: str

class isek.agent.memory.AgentMemory[source]#

Bases: object

Manages the complete cognitive state of an agent.

This includes working memory for recent interactions, a long-term knowledge store, a queue for scheduled tasks, and internal state variables that track the agent's status and goals.

create_task(task_description: str, priority: int = 1) → str[source]#

Creates a new task and adds it to the task queue.

Each task is assigned a unique ID and includes its description, creation timestamp, completion status, and priority.

Parameters:

task_description (str) -- A description of the task to be performed.
priority (int) -- The priority of the task (e.g., higher number for higher priority). Defaults to 1.

Returns:

The unique ID of the created task.

Return type:

str

generate_timestamp() → str[source]#

Generates a current timestamp string.

The format is "YYYY-MM-DD HH:MM:SS".

Returns:: The current timestamp as a formatted string.
Return type:: str

generate_unique_id() → str[source]#

Generates a short unique ID.

The ID is the first 8 characters of a UUID version 4.

Returns:: A unique 8-character string identifier.
Return type:: str

get_all_knowledge_topics() → List[str][source]#

Retrieves a list of all topics currently in the knowledge store.

Returns:: A list of all knowledge topics.
Return type:: List[str]

get_all_state_variables() → Dict[str, Any][source]#

Retrieves all current state variables and their values.

Returns:: A dictionary containing all state variables.
Return type:: Dict[str, Any]

get_all_tasks() → Dict[str, Dict[str, Any]][source]#

Retrieves all tasks from the queue, both pending and completed.

Returns:: A dictionary of all tasks, where keys are task IDs and values are dictionaries of task details.
Return type:: Dict[str, Dict[str, Any]]

get_pending_tasks() → Dict[str, Dict[str, Any]][source]#

Retrieves all tasks that are currently not marked as completed.

Returns:: A dictionary of pending tasks, where keys are task IDs and values are dictionaries of task details.
Return type:: Dict[str, Dict[str, Any]]

get_recent_memory_items(count: int = 4) → List[str][source]#

Retrieves the most recent items from working memory.

Parameters:: count (int) -- The maximum number of recent items to retrieve. Defaults to 4.
Returns:: A list of the most recent memory items, up to count.
Return type:: List[str]

get_state_variable(variable_name: str, default: T | None = None) → Any | T[source]#

Retrieves the value of a state variable.

Parameters:

variable_name (str) -- The name of the state variable to retrieve.
default (T) -- The default value to return if the variable_name is not found. Defaults to None.

Returns:

The value of the state variable, or the default value if not found.

Return type:

Union[Any, T]

knowledge_interaction(action: str, content: str) → str[source]#

Manages interactions with the agent's knowledge store.

Allows querying for existing knowledge or inserting new knowledge.

Parameters:

action (str) -- The action to perform: "query" or "insert".
content (str) -- For "query", this is the topic to search for. For "insert", this is the content to store (also used as topic currently).

Returns:

A string confirming the action or the result of the query.

Return type:

str

log_operation(operation_name: str, details: str) → None[source]#

Logs a memory operation using the configured logger.

Parameters:

operation_name (str) -- The name of the memory operation performed (e.g., "set_state_variable").
details (str) -- Specific details about the operation (e.g., variable name and value).

mark_task_completed(task_id: str) → bool[source]#

Marks a specified task in the queue as completed.

Sets the 'completed' flag to True and records the completion timestamp.

Parameters:: task_id (str) -- The ID of the task to mark as completed.
Returns:: True if the task was found and marked completed, False otherwise.
Return type:: bool

retrieve_knowledge(topic: str) → str | None[source]#

Retrieves knowledge content associated with a specific topic.

Searches the existing knowledge store for an item matching the given topic.

Note

The original docstring mentioned a generator_func parameter, which is not present in the method signature. This version reflects the current signature.

Parameters:: topic (str) -- The topic to retrieve knowledge about.
Returns:: The content of the knowledge item if found, otherwise None.
Return type:: Optional[str]

scheduling(option: str, id: str | None = None, task: str | None = None, finished: bool = False) → str[source]#

Manages the agent's task schedule.

Allows adding new tasks, marking tasks as complete, or fetching all tasks.

Warning

The finished parameter is present in the signature but not currently used. In the "complete" option, task_id is used in the return message but refers to the local id parameter.

Parameters:

option (str) -- The scheduling action: "add", "complete", or "fetch".
id (Optional[str]) -- The ID of the task (used for "complete"). Defaults to None.
task (Optional[str]) -- The description of the task (used for "add"). Defaults to None.
finished (bool) -- This parameter is currently unused. Defaults to False.

Returns:

A string confirming the action or providing the fetched schedule.

Return type:

str

set_state_variable(variable_name: str, value: Any) → str[source]#

Sets or updates a state variable in the agent's memory.

Parameters:

variable_name (str) -- The name of the state variable to set or update.
value (Any) -- The new value for the state variable.

Returns:

A confirmation message indicating the variable was set.

Return type:

str

state_interaction(action: str, title: str, value: Any) → str[source]#

Manages interactions with the agent's state variables.

This method allows for updating, inserting (which is equivalent to updating), or fetching the agent's state.

Parameters:

action (str) -- The action to perform: "update", "insert", or "fetch".
title (str) -- The name (key) of the state variable.
value (Any) -- The value to set for the state variable (used for "update" and "insert"). Ignored for "fetch".

Returns:

A string confirming the action taken or providing the fetched state.

Return type:

str

store_knowledge(topic: str, content: str) → str[source]#

Stores a piece of knowledge in the knowledge store.

Each knowledge item is associated with a topic and includes the content and a creation timestamp.

Parameters:

topic (str) -- The topic or title for the knowledge item.
content (str) -- The actual content of the knowledge item.

Returns:

The unique ID of the stored knowledge item.

Return type:

str

store_memory_item(content: str) → str[source]#

Adds an item to the agent's working memory.

Working memory typically stores recent interactions or temporary thoughts.

Parameters:: content (str) -- The string content to store in working memory.
Returns:: A confirmation message.
Return type:: str

class isek.agent.persona.Persona(name: str, bio: str, lore: str, knowledge: str, routine: str)[source]#

Bases: object

Represents the identity, characteristics, and background of an agent.

A persona defines how an agent behaves, its knowledge base, its mission, and its typical routines or operational procedures.

classmethod default() → Persona[source]#

Creates a default Persona instance.

This persona is a general-purpose "Helper" with predefined attributes.

Returns:: A default Persona instance.
Return type:: Persona

classmethod from_json(data: Dict[str, str]) → Persona[source]#

Creates a Persona instance from a dictionary of data.

The dictionary keys should correspond to the Persona's constructor parameters (name, bio, lore, knowledge, routine).

Parameters:: data (Dict[str, str]) -- A dictionary containing the persona attributes.
Returns:: A Persona instance created from the provided data.
Return type:: Persona
Raises:: KeyError -- If the data dictionary is missing one or more required keys (name, bio, lore, knowledge, routine).

classmethod load(path: str) → Persona[source]#

Loads Persona data from a JSON file and creates a Persona instance.

Parameters:

path (str) -- The file path to the JSON file containing persona data.

Returns:

A Persona instance created from the data in the file.

Return type:

Persona

Raises:

FileNotFoundError -- If the specified file path does not exist.
json.JSONDecodeError -- If the file content is not valid JSON.
KeyError -- If the JSON data is missing required fields for persona creation.

class isek.agent.single_agent.SingleAgent(**kwargs: Any)[source]#

Bases: AbstractAgent

A concrete implementation of an agent that runs as a single, standalone instance.

This agent type typically interacts via a command-line interface (CLI) and does not inherently possess distributed or networked capabilities unless explicitly added through its tools or persona logic. It builds upon the AbstractAgent.

build(daemon: bool = False) → None[source]#

Builds and starts the single agent, typically by launching its command-line interface.

If daemon is False (the default), the agent's CLI (run_cli()) is run in the current thread, which will block further execution in that thread until the CLI is exited.

If daemon is True, the CLI is started in a separate daemon thread, allowing the main program to continue executing while the agent runs in the background.

Parameters:: daemon (bool) -- If True, run the agent's CLI in a daemon thread. Defaults to False.

class isek.agent.toolbox.ToolBox(persona: Persona | None = None)[source]#

Bases: object

Manages a collection of tools (functions) that an agent can use.

The ToolBox handles the registration of tools, storage of their metadata (like descriptions and schemas for LLM interaction), and the execution of tool calls based on requests from a language model.

execute_tool_call(tool_call: Any, **extra_kwargs: Any) → str[source]#

Executes a tool call based on an object from an LLM response.

The tool_call object is expected to have a function.name attribute for the tool's name and a function.arguments attribute containing a JSON string of arguments for the tool.

Parameters:

tool_call (Any) -- The tool call object, typically from an LLM's response (e.g., OpenAI's tool_calls object). It should have function.name and function.arguments (as JSON string).
extra_kwargs (Any) -- Additional keyword arguments to be passed to the tool function, merged with arguments from tool_call.

Returns:

A string representation of the tool's execution result. Returns an error message string if the tool is not found or if an error occurs during execution.

Return type:

str

get_tool(name: str) → Callable[[...], Any] | None[source]#

Retrieves a registered tool function by its name.

Parameters:: name (str) -- The name of the tool to retrieve.
Returns:: The callable tool function if found, otherwise None.
Return type:: Optional[Callable[..., Any]]

get_tool_names() → List[str][source]#

Gets a list of names of all registered tools.

Returns:: A list of tool names.
Return type:: List[str]

get_tool_schemas() → List[Dict[str, Any]][source]#

Gets the LLM-compatible schemas for all registered tools.

Note

The original method had an optional category parameter which is not used in the current implementation as tools are not categorized. This docstring reflects the current signature.

Returns:: A list of tool schemas. Each schema is a dictionary.
Return type:: List[Dict[str, Any]]

register_tool(func: Callable[[...], Any]) → None[source]#

Registers a new callable function as a tool.

The function's name is used as the tool name. Its docstring is used as the description, and an LLM-compatible schema is generated based on its signature and type annotations.

Parameters:: func (Callable[..., Any]) -- The function to register as a tool. It should have type hints for its parameters and a docstring for its description.

register_tools(tools: List[Callable[[...], Any]]) → None[source]#

Registers multiple tools at once.

Iterates through the provided list of functions and calls register_tool() for each one.

Parameters:: tools (List[Callable[..., Any]]) -- A list of callable functions to register as tools.