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, max_tool_rounds=8, 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 | None)
max_tokens (int)
max_output_tokens (int | None)
system_message (str | None)
timeout (float | None)
retries (int)
max_tool_rounds (int)
stream (bool)
response_schema (Dict[str, Any] | None)
reasoning (str | Dict[str, Any] | ReasoningConfig | 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 | None = 0.7

max_tokens: int = 1000

max_output_tokens: int | None = None

system_message: str | None = None

timeout: float | None = None

retries: int = 2

max_tool_rounds: int = 8

stream: bool = False

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

reasoning: str | Dict[str, Any] | ReasoningConfig | 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, max_tool_rounds=8, 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 | None)
max_tokens (int)
max_output_tokens (int | None)
system_message (str | None)
timeout (float | None)
retries (int)
max_tool_rounds (int)
stream (bool)
response_schema (Dict[str, Any] | None)
reasoning (str | Dict[str, Any] | ReasoningConfig | 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, reasoning=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)
reasoning (str | Dict[str, Any] | ReasoningConfig | 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, reasoning=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
reasoning (Union[str, Dict[str, Any], ReasoningConfig, None]) – Portable level or provider-specific reasoning controls
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 non-system messages to retain, trimmed in whole user-turn units; the newest unit is always kept (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, **kwargs)[source]

Send a message to the agent and get a response.

Parameters:

message (Optional[str]) – User message to send to the agent
**kwargs (Any) – Per-call options, the same as for generate()

Return type:

str

Returns:

Agent’s response as a string

Raises:

ValueError – If message is empty or None, or a tool name is unknown
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.

Parameters:

message (Any) – User message to send to the agent
**kwargs (Any) – Per-call options: response_schema, reasoning, provider_options, timeout, metadata, stream_options, stream, max_tool_rounds, allowed_tool_names and additional_system_message. Unknown keywords are logged and ignored.

Raises:

ValueError – If message is empty or a tool name is unknown
PravalError – If response generation fails

Return type:

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.

The user turn enters history when the stream is created. The answer enters history when the final event is produced, before it is yielded; a failed or abandoned stream leaves only the user turn.

Return type:

Any

Parameters:

message (Any)
kwargs (Any)

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

Asynchronously stream provider-neutral model events.

History is updated as described for stream().

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

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, and removes the channel an @agent owns (<name>_channel) once it has no other subscribers
Unregisters the tools this agent added to the global tool registry
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[source]: 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