praval.decorators

Decorator-based Agent API for Praval Framework.

This module provides a Pythonic decorator interface for creating agents that automatically handle reef communication and coordination.

Example:

from praval import agent, chat, broadcast, start_agents, get_reef

@agent("explorer", responds_to=["concept_request"])
def explore_concepts(spore):
    concepts = chat("Find concepts related to: " + spore.knowledge.get("concept",
    ""))
    broadcast({"type": "discovery", "discovered": concepts.split(",")})
    return {"discovered": concepts}

# Start the agent system
start_agents(explore_concepts, initial_data={"type": "concept_request", "concept":
"AI"})
get_reef().wait_for_completion()
get_reef().shutdown()

Functions

achat(message[, timeout])	Async version of chat function for use within async agent handlers.
broadcast(data[, channel, message_type])	Quick broadcast function that uses the current agent's communication.
chat(message[, timeout])	Quick chat function that uses the current agent's LLM with timeout support.
get_agent_info(agent_func)	Get information about an @agent decorated function.

praval.decorators.chat(message, timeout=10.0)

Quick chat function that uses the current agent’s LLM with timeout support. Can only be used within @agent decorated functions.

Parameters:

• message (str) – Message to send to the LLM

• timeout (float) – Maximum time to wait for response in seconds

Return type:

str

Returns:

LLM response as string

Raises:

• RuntimeError – If called outside of an @agent function

• TimeoutError – If LLM call exceeds timeout

async praval.decorators.achat(message, timeout=10.0)

Async version of chat function for use within async agent handlers.

Parameters:

• message (str) – Message to send to the LLM

• timeout (float) – Maximum time to wait for response in seconds

Return type:

str

Returns:

LLM response as string

Raises:

• RuntimeError – If called outside of an @agent function

• TimeoutError – If LLM call exceeds timeout

praval.decorators.broadcast(data, channel=None, message_type=None)

Quick broadcast function that uses the current agent’s communication. Can only be used within @agent decorated functions.

Parameters:

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

• channel (Optional[str]) – Channel to broadcast to. Defaults to the channel set by start_agents(), or reef’s default channel if not in a start_agents() context.

• message_type (Optional[str]) – Message type to set (automatically added to data)

Return type:

str

Returns:

Spore ID of the broadcast message

Raises:

RuntimeError – If called outside of an @agent function

Example

# Broadcast to all agents on the same channel (set by start_agents) broadcast({“type”: “analysis_request”, “data”: findings})

# Broadcast to a specific channel broadcast({“type”: “alert”}, channel=”urgent_alerts”)

praval.decorators.get_agent_info(agent_func)

Get information about an @agent decorated function.

Parameters:

agent_func (Callable) – Function decorated with @agent

Return type:

Dict[str, Any]

Returns:

Dictionary with agent metadata