praval.core.agent

Core Agent class for the Praval framework.

The Agent class provides a simple, composable interface for LLM-based conversations with support for multiple providers, tools, and state persistence.

Classes

Agent(name[, provider, persist_state, ...])	A simple, composable LLM agent.
AgentConfig([provider, temperature, ...])	Configuration for Agent behavior and LLM parameters.

class praval.core.agent.AgentConfig(provider=None, temperature=0.7, max_tokens=1000, system_message=None)

Bases: object

Configuration for Agent behavior and LLM parameters.

Parameters:

• provider (str | None)

• temperature (float)

• max_tokens (int)

• system_message (str | None)

provider: Optional[str] = None

temperature: float = 0.7

max_tokens: int = 1000

system_message: Optional[str] = None

__post_init__()

Validate configuration parameters.

__init__(provider=None, temperature=0.7, max_tokens=1000, system_message=None)

Parameters:

• provider (str | None)

• temperature (float)

• max_tokens (int)

• system_message (str | None)

Return type:

None

class praval.core.agent.Agent(name, provider=None, persist_state=False, system_message=None, config=None, memory_enabled=False, memory_config=None, knowledge_base=None, max_history=100, hitl_enabled=False, hitl_db_path=None)

Bases: object

A simple, composable LLM agent.

The Agent class provides the core functionality for LLM-based conversations with support for multiple providers, conversation history, tools, and state persistence.

Examples

Basic usage: >>> agent = Agent(“assistant”) >>> response = agent.chat(“Hello!”)

With persistence: >>> agent = Agent(“my_agent”, persist_state=True) >>> agent.chat(“Remember this conversation”)

With tools: >>> agent = Agent(“calculator”) >>> @agent.tool >>> def add(x: int, y: int) -> int: … return x + y

Parameters:

• name (str)

• provider (str | None)

• persist_state (bool)

• system_message (str | None)

• config (Dict[str, Any] | None)

• memory_enabled (bool)

• memory_config (Dict[str, Any] | None)

• knowledge_base (str | None)

• max_history (int | None)

• hitl_enabled (bool)

• hitl_db_path (str | None)

__init__(name, provider=None, persist_state=False, system_message=None, config=None, memory_enabled=False, memory_config=None, knowledge_base=None, max_history=100, hitl_enabled=False, hitl_db_path=None)

Initialize a new Agent.

Parameters:

• name (str) – Unique identifier for this agent

• provider (Optional[str]) – LLM provider to use (openai, anthropic, cohere)

• persist_state (bool) – Whether to persist conversation state

• system_message (Optional[str]) – System message to set agent behavior

• config (Optional[Dict[str, Any]]) – Additional configuration parameters

• memory_enabled (bool) – Whether to enable vector memory capabilities

• memory_config (Optional[Dict[str, Any]]) – Configuration for memory system

• knowledge_base (Optional[str]) – Path to knowledge base files to auto-index

• max_history (Optional[int]) – Max conversation turns to retain (None for unbounded)

• hitl_enabled (bool)

• hitl_db_path (str | None)

Raises:

• ValueError – If name is empty or configuration is invalid

• ProviderError – If provider setup fails

chat(message)

Send a message to the agent and get a response.

Parameters:

message (Optional[str]) – User message to send to the agent

Return type:

str

Returns:

Agent’s response as a string

Raises:

• ValueError – If message is empty or None

• PravalError – If response generation fails

configure_hitl(*, enabled=True, db_path=None)

Configure HITL behavior for this agent.

Parameters:

• enabled (bool) – Whether HITL is enabled for this agent

• db_path (Optional[str]) – Optional SQLite path override for intervention storage

Return type:

None

get_pending_interventions(run_id=None, limit=100)

Get pending interventions filtered to this agent.

Return type:

List[Any]

Parameters:

• run_id (str | None)

• limit (int)

approve_intervention(intervention_id, *, reviewer='human', edited_args=None)

Approve or edit-approve an intervention for this agent.

Return type:

Any

Parameters:

• intervention_id (str)

• reviewer (str)

• edited_args (Dict[str, Any] | None)

reject_intervention(intervention_id, *, reason, reviewer='human')

Reject an intervention for this agent.

Return type:

Any

Parameters:

• intervention_id (str)

• reason (str)

• reviewer (str)

resume_run(run_id)

Resume a previously suspended HITL run after a decision.

Parameters:

run_id (str) – Suspended run identifier

Return type:

str

Returns:

Final model response for the resumed run

tool(func)

Decorator to register a function as a tool for the agent.

Parameters:

func (Callable) – Function to register as a tool

Return type:

Callable

Returns:

The original function (unchanged)

Raises:

ValueError – If function lacks proper type hints

send_knowledge(to_agent, knowledge, channel='main')

Send knowledge to another agent through the reef.

Parameters:

• to_agent (str) – Name of the target agent

• knowledge (Dict[str, Any]) – Knowledge data to send

• channel (str) – Reef channel to use (default: “main”)

Return type:

str

Returns:

Spore ID of the sent message

broadcast_knowledge(knowledge, channel='main')

Broadcast knowledge to all agents in the reef.

Parameters:

• knowledge (Dict[str, Any]) – Knowledge data to broadcast

• channel (str) – Reef channel to use (default: “main”)

Return type:

str

Returns:

Spore ID of the broadcast message

request_knowledge(from_agent, request, timeout=30)

Request knowledge from another agent with timeout.

Parameters:

• from_agent (str) – Name of the agent to request from

• request (Dict[str, Any]) – Request data

• timeout (int) – Timeout in seconds

Return type:

Optional[Dict[str, Any]]

Returns:

Response data or None if timeout

on_spore_received(spore)

Handle received spores from the reef.

This is a default implementation that can be overridden by subclasses for custom spore handling.

Parameters:

spore – The received Spore object

Return type:

None

subscribe_to_channel(channel_name)

Subscribe this agent to a reef channel.

Parameters:

channel_name (str) – Name of the channel to subscribe to

Return type:

None

unsubscribe_from_channel(channel_name)

Unsubscribe this agent from a reef channel.

Parameters:

channel_name (str) – Name of the channel to unsubscribe from

Return type:

None

property spore_handler: Callable | None

Get the current spore handler for this agent.

Returns:

The custom spore handler function, or None if not set

set_spore_handler(handler)

Set a custom spore handler for this agent.

Parameters:

handler (Callable) – Function that takes a Spore object and handles it

Return type:

None

close()

Release all resources held by the agent.

This method: - Unsubscribes from all reef channels - Shuts down the memory system - Clears conversation history

Safe to call multiple times. After calling close(), the agent should not be used for further operations.

Example:

agent = Agent("assistant")
try:
    response = agent.chat("Hello")
finally:
    agent.close()

# Or use as context manager:
with Agent("assistant") as agent:
    response = agent.chat("Hello")

Return type:

None

__enter__()

Context manager entry - returns the agent.

Return type:

Agent

__exit__(exc_type, exc_val, exc_tb)

Context manager exit - ensures cleanup.

Return type:

None

__del__()

Destructor - attempt cleanup if not already done.

property is_closed: bool

Check if the agent has been closed.

remember(content, importance=0.5, memory_type='short_term')

Store a memory

Parameters:

• content (str) – The content to remember

• importance (float) – Importance score (0.0 to 1.0)

• memory_type (str) – Type of memory (“short_term”, “semantic”, “episodic”)

Return type:

Optional[str]

Returns:

Memory ID if successful, None otherwise

recall(query, limit=5, similarity_threshold=0.1)

Recall memories based on a query

Parameters:

• query (str) – Search query

• limit (int) – Maximum number of results

• similarity_threshold (float) – Minimum similarity score

Return type:

List

Returns:

List of MemoryEntry objects

recall_by_id(memory_id)

Recall a specific memory by ID (for resolving spore references)

Return type:

List

Parameters:

memory_id (str)

get_conversation_context(turns=10)

Get recent conversation context

Return type:

List

Parameters:

turns (int)

create_knowledge_reference(content, importance=0.8)

Create knowledge references for lightweight spores

Parameters:

• content (str) – Knowledge content to store and reference

• importance (float) – Importance threshold

Return type:

List[str]

Returns:

List of knowledge reference IDs

resolve_spore_knowledge(spore)

Resolve knowledge references in a spore

Parameters:

spore – Spore object with potential knowledge references

Return type:

Dict[str, Any]

Returns:

Complete knowledge including resolved references

send_lightweight_knowledge(to_agent, large_content, summary, channel='main')

Send large knowledge as lightweight spore with references

Parameters:

• to_agent (str) – Target agent

• large_content (str) – Large content to reference

• summary (str) – Brief summary for the spore

• channel (str) – Communication channel

Return type:

str

Returns:

Spore ID