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, model, ...])	A simple, composable LLM agent.
`AgentConfig`([provider, model, base_url, ...])	Configuration for Agent behavior and LLM parameters.

class praval.core.agent.AgentConfig(provider=None, model=None, base_url=None, api_key_env=None, temperature=0.7, max_tokens=1000, max_output_tokens=None, system_message=None, timeout=None, retries=2, stream=False, response_schema=None, reasoning=None, store=False, cache=None, strict_tools=False, provider_options=None, stream_options=None)[source]

Bases: object

Configuration for Agent behavior and LLM parameters.

Parameters:

provider (str | None)
model (str | None)
base_url (str | None)
api_key_env (str | None)
temperature (float)
max_tokens (int)
max_output_tokens (int | None)
system_message (str | None)
timeout (float | None)
retries (int)
stream (bool)
response_schema (Dict[str, Any] | None)
reasoning (Dict[str, Any] | None)
store (bool)
cache (Dict[str, Any] | None)
strict_tools (bool)
provider_options (Dict[str, Any] | None)
stream_options (Dict[str, Any] | None)

provider: str | None = None

model: str | None = None

base_url: str | None = None

api_key_env: str | None = None

temperature: float = 0.7

max_tokens: int = 1000

max_output_tokens: int | None = None

system_message: str | None = None

timeout: float | None = None

retries: int = 2

stream: bool = False

response_schema: Dict[str, Any] | None = None

reasoning: Dict[str, Any] | None = None

store: bool = False

cache: Dict[str, Any] | None = None

strict_tools: bool = False

provider_options: Dict[str, Any] | None = None

stream_options: Dict[str, Any] | None = None

__post_init__()[source]: Validate configuration parameters.

__init__(provider=None, model=None, base_url=None, api_key_env=None, temperature=0.7, max_tokens=1000, max_output_tokens=None, system_message=None, timeout=None, retries=2, stream=False, response_schema=None, reasoning=None, store=False, cache=None, strict_tools=False, provider_options=None, stream_options=None)

Parameters:

provider (str | None)
model (str | None)
base_url (str | None)
api_key_env (str | None)
temperature (float)
max_tokens (int)
max_output_tokens (int | None)
system_message (str | None)
timeout (float | None)
retries (int)
stream (bool)
response_schema (Dict[str, Any] | None)
reasoning (Dict[str, Any] | None)
store (bool)
cache (Dict[str, Any] | None)
strict_tools (bool)
provider_options (Dict[str, Any] | None)
stream_options (Dict[str, Any] | None)

Return type:

None

class praval.core.agent.Agent(name, provider=None, model=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)[source]

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)
model (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, model=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)[source]

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)
model (str | None)
hitl_enabled (bool)
hitl_db_path (str | None)

Raises:

ValueError – If name is empty or configuration is invalid
ProviderError – If provider setup fails

tools: Dict[str, Dict[str, Any]]

conversation_history: List[Dict[str, Any]]

chat(message)[source]

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

generate(message, **kwargs)[source]

Generate a provider-neutral model response.

This is the richer counterpart to chat(); chat() remains the compatibility API that returns only text.

Return type:

Any

Parameters:

message (Any)
kwargs (Any)

transcribe(audio, *, model=None, filename=None, mime_type=None, language=None, prompt=None, response_format='json', temperature=None, provider_options=None, timeout=None, metadata=None)[source]

Transcribe request-based audio without changing chat history.

Return type:

str

Parameters:

audio (Any)
model (str | None)
filename (str | None)
mime_type (str | None)
language (str | None)
prompt (str | None)
response_format (str)
temperature (float | None)
provider_options (Dict[str, Any] | None)
timeout (float | None)
metadata (Dict[str, Any] | None)

speak(text, *, model=None, voice='alloy', response_format='mp3', speed=1.0, instructions=None, provider_options=None, timeout=None, metadata=None)[source]

Synthesize request-based speech without changing chat history.

Return type:

bytes

Parameters:

text (str)
model (str | None)
voice (str)
response_format (str)
speed (float)
instructions (str | None)
provider_options (Dict[str, Any] | None)
timeout (float | None)
metadata (Dict[str, Any] | None)

async agenerate(message, **kwargs)[source]

Async wrapper for generate().

Return type:

Any

Parameters:

message (Any)
kwargs (Any)

stream(message, **kwargs)[source]

Stream provider-neutral model events.

Return type:

Any

Parameters:

message (Any)
kwargs (Any)

async astream(message, **kwargs)[source]

Asynchronously stream provider-neutral model events.

Return type:

Any

Parameters:

message (Any)
kwargs (Any)

configure_hitl(*, enabled=True, db_path=None)[source]

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)[source]

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)[source]

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')[source]

Reject an intervention for this agent.

Return type:

Any

Parameters:

intervention_id (str)
reason (str)
reviewer (str)

resume_run(run_id)[source]

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

async aresume_run(run_id)[source]

Asynchronously resume a suspended run containing async-only tools.

Return type:: str
Parameters:: run_id (str)

tool(func)[source]

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

add_tool_spec(spec, handler, *, async_only=False)[source]

Register an externally described JSON-schema tool on this agent.

Parameters:

spec (ToolSpec) – Provider-neutral tool declaration.
handler (Callable[..., Any]) – Callable invoked with the model-supplied keyword arguments.
async_only (bool) – Whether the tool may only run through async Agent APIs.

Raises:

TypeError – If spec or handler has the wrong type.
ValueError – If the name or schema is invalid or already registered.

Return type:

None

send_knowledge(to_agent, knowledge, channel='main')[source]

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')[source]

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)[source]

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)[source]

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)[source]

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)[source]

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)[source]

Set a custom spore handler for this agent.

Parameters:: handler (Callable) – Function that takes a Spore object and handles it
Return type:: None

close()[source]

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__()[source]

Context manager entry - returns the agent.

Return type:: Agent

__exit__(exc_type, exc_val, exc_tb)[source]

Context manager exit - ensures cleanup.

Return type:: None

__del__()[source]: 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')[source]

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)[source]

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)[source]

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

Return type:: List
Parameters:: memory_id (str)

get_conversation_context(turns=10)[source]

Get recent conversation context

Return type:: List
Parameters:: turns (int)

create_knowledge_reference(content, importance=0.8)[source]

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)[source]

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')[source]

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