class ProgrammedAgentInstructionResult(BaseModel):
Result of a programmable agent instruction execution.Contains the messages exchanged during execution and the computed value.
The value type is specified by the generic type parameter T.Parameters:
user_message (BaseMessage): The message sent by the user.
agent_message (BaseMessage): The message sent by the agent.
Abstract class for a programmable agent.A programmable agent is an agent that can be programmed to perform a
specific function or task. This class defines the interface for a
programmable agent.These methods should be implemented in order to ensure the agent supports
the necessary guarantees to enable a programming interface while
maintaining compatibility in a multi-agent system.A programmable agent is responsible for providing and maintaining a
programming interface for its functionality.
Run an atomic operation on the agent.An atomic operation is an operation that is guaranteed to
be executed without interruption by any other operation.Parameters:
callback (Callable[[], ProgrammedAgentInstructionResult[T]]): The operation to execute atomically.
Returns:ProgrammedAgentInstructionResult[T]: The result of the operation.
Repair the state of the agent.Agents may have other non-atomic interfaces, such as a user interface,
or chat between other agents. This method should restore the agent to
a state where it can perform operations according to the specified
requirement.Parameters:
requirement (ProgrammableAgentRequirement): The requirement to repair the state for.
Decorator for programmable agent capabilities.This decorator ensures that the decorated method is executed atomically
and maintains the agent’s state guarantees.Parameters:
func (Callable[…, ProgrammedAgentInstructionResult[T]]): The method to decorate.
Returns:Callable[…, ProgrammedAgentInstructionResult[T]]: The decorated
method that ensures atomic execution.
class ProgrammableChatAgent(ChatAgent, AbstractProgrammableAgent):
A chat agent that can be programmed to perform specific tasks.Provides a default implementation of atomic execution using threading locks
and basic state tracking for message roles. Implementing classes need to
provide specific repair logic for their use cases.Parameters:
_operation_lock (threading.Lock): Lock for ensuring atomic operations.
_last_message_role (Optional[str]): Role of the last message in the conversation.