pepperplus/pepperplus-cb
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge branch 'main' of https://git.science.uu.nl/ics/sp/2025/n25b/pepperplus-cb into feat/face-recognition

Browse Source
This commit is contained in:
JobvAlewijk
2026-01-29 16:21:55 +01:00
parent f6477e5325 d9e5de6e51
commit 00a13426a0
35 changed files with 2110 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

385
src/control_backend/agents/bdi/agentspeak_generator.py
View File

@@ -18,6 +18,7 @@ from control_backend.agents.bdi.agentspeak_ast import (
StatementType,
TriggerType,
)
from control_backend.core.config import settings
from control_backend.schemas.program import (
BaseGoal,
BasicNorm,
@@ -48,6 +49,15 @@ class AgentSpeakGenerator:
It handles the conversion of phases, norms, goals, and triggers into AgentSpeak rules and plans,
ensuring the robot follows the defined behavioral logic.
The generator follows a systematic approach:
1. Sets up initial phase and cycle notification rules
2. Adds keyword inference capabilities for natural language processing
3. Creates default plans for common operations
4. Processes each phase with its norms, goals, and triggers
5. Adds fallback plans for robust execution
:ivar _asp: The internal AgentSpeak program representation being built.
"""
_asp: AstProgram
@@ -56,6 +66,10 @@ class AgentSpeakGenerator:
"""
Translates a Program object into an AgentSpeak source string.
This is the main entry point for the code generation process. It initializes
the AgentSpeak program structure and orchestrates the conversion of all
program elements into their AgentSpeak representations.
:param program: The behavior program to translate.
:return: The generated AgentSpeak code as a string.
"""
@@ -78,6 +92,18 @@ class AgentSpeakGenerator:
return str(self._asp)
def _add_keyword_inference(self) -> None:
"""
Adds inference rules for keyword detection in user messages.
This method creates rules that allow the system to detect when specific
keywords are mentioned in user messages. It uses string operations to
check if a keyword is a substring of the user's message.
The generated rule has the form:
keyword_said(Keyword) :- user_said(Message) & .substring(Keyword, Message, Pos) & Pos >= 0
This enables the system to trigger behaviors based on keyword detection.
"""
keyword = AstVar("Keyword")
message = AstVar("Message")
position = AstVar("Pos")
@@ -92,12 +118,32 @@ class AgentSpeakGenerator:
)
def _add_default_plans(self):
"""
Adds default plans for common operations.
This method sets up the standard plans that handle fundamental operations
like replying with goals, simple speech actions, general replies, and
cycle notifications. These plans provide the basic infrastructure for
the agent's reactive behavior.
"""
self._add_reply_with_goal_plan()
self._add_say_plan()
self._add_reply_plan()
self._add_notify_cycle_plan()
def _add_reply_with_goal_plan(self):
"""
Adds a plan for replying with a specific conversational goal.
This plan handles the case where the agent needs to respond to user input
while pursuing a specific conversational goal. It:
1. Marks that the agent has responded this turn
2. Gathers all active norms
3. Generates a reply that considers both the user message and the goal
Trigger: +!reply_with_goal(Goal)
Context: user_said(Message)
"""
self._asp.plans.append(
AstPlan(
TriggerType.ADDED_GOAL,
@@ -123,6 +169,17 @@ class AgentSpeakGenerator:
)
def _add_say_plan(self):
"""
Adds a plan for simple speech actions.
This plan handles direct speech actions where the agent needs to say
a specific text. It:
1. Marks that the agent has responded this turn
2. Executes the speech action
Trigger: +!say(Text)
Context: None (can be executed anytime)
"""
self._asp.plans.append(
AstPlan(
TriggerType.ADDED_GOAL,
@@ -136,6 +193,18 @@ class AgentSpeakGenerator:
)
def _add_reply_plan(self):
"""
Adds a plan for general reply actions.
This plan handles general reply actions where the agent needs to respond
to user input without a specific conversational goal. It:
1. Marks that the agent has responded this turn
2. Gathers all active norms
3. Generates a reply based on the user message and norms
Trigger: +!reply
Context: user_said(Message)
"""
self._asp.plans.append(
AstPlan(
TriggerType.ADDED_GOAL,
@@ -159,6 +228,19 @@ class AgentSpeakGenerator:
)
def _add_notify_cycle_plan(self):
"""
Adds a plan for cycle notification.
This plan handles the periodic notification cycle that allows the system
to monitor and report on the current state. It:
1. Gathers all active norms
2. Notifies the system about the current norms
3. Waits briefly to allow processing
4. Recursively triggers the next cycle
Trigger: +!notify_cycle
Context: None (can be executed anytime)
"""
self._asp.plans.append(
AstPlan(
TriggerType.ADDED_GOAL,
@@ -182,6 +264,16 @@ class AgentSpeakGenerator:
)
def _process_phases(self, phases: list[Phase]) -> None:
"""
Processes all phases in the program and their transitions.
This method iterates through each phase and:
1. Processes the current phase (norms, goals, triggers)
2. Sets up transitions between phases
3. Adds special handling for the end phase
:param phases: The list of phases to process.
"""
for curr_phase, next_phase in zip([None] + phases, phases + [None], strict=True):
if curr_phase:
self._process_phase(curr_phase)
@@ -204,6 +296,17 @@ class AgentSpeakGenerator:
)
def _process_phase(self, phase: Phase) -> None:
"""
Processes a single phase, including its norms, goals, and triggers.
This method handles the complete processing of a phase by:
1. Processing all norms in the phase
2. Setting up the default execution loop for the phase
3. Processing all goals in sequence
4. Processing all triggers for reactive behavior
:param phase: The phase to process.
"""
for norm in phase.norms:
self._process_norm(norm, phase)
@@ -218,6 +321,21 @@ class AgentSpeakGenerator:
self._process_trigger(trigger, phase)
def _add_phase_transition(self, from_phase: Phase | None, to_phase: Phase | None) -> None:
"""
Adds plans for transitioning between phases.
This method creates two plans for each phase transition:
1. A check plan that verifies if transition conditions are met
2. A force plan that actually performs the transition (can be forced externally)
The transition involves:
- Notifying the system about the phase change
- Removing the current phase belief
- Adding the next phase belief
:param from_phase: The phase being transitioned from (or None for initial setup).
:param to_phase: The phase being transitioned to (or None for end phase).
"""
if from_phase is None:
return
from_phase_ast = self._astify(from_phase)
@@ -247,18 +365,6 @@ class AgentSpeakGenerator:
AstStatement(StatementType.ADD_BELIEF, to_phase_ast),
]
# if from_phase:
# body.extend(
# [
# AstStatement(
# StatementType.TEST_GOAL, AstLiteral("user_said", [AstVar("Message")])
# ),
# AstStatement(
# StatementType.REPLACE_BELIEF, AstLiteral("user_said", [AstVar("Message")])
# ),
# ]
# )
# Check
self._asp.plans.append(
AstPlan(
@@ -279,6 +385,17 @@ class AgentSpeakGenerator:
)
def _process_norm(self, norm: Norm, phase: Phase) -> None:
"""
Processes a norm and adds it as an inference rule.
This method converts norms into AgentSpeak rules that define when
the norm should be active. It handles both basic norms (always active
in their phase) and conditional norms (active only when their condition
is met).
:param norm: The norm to process.
:param phase: The phase this norm belongs to.
"""
rule: AstRule | None = None
match norm:
@@ -297,6 +414,18 @@ class AgentSpeakGenerator:
self._asp.rules.append(rule)
def _add_default_loop(self, phase: Phase) -> None:
"""
Adds the default execution loop for a phase.
This method creates the main reactive loop that runs when the agent
receives user input during a phase. The loop:
1. Notifies the system about the user input
2. Resets the response tracking
3. Executes all phase goals
4. Attempts phase transition
:param phase: The phase to create the loop for.
"""
actions = []
actions.append(
@@ -305,7 +434,6 @@ class AgentSpeakGenerator:
)
)
actions.append(AstStatement(StatementType.REMOVE_BELIEF, AstLiteral("responded_this_turn")))
actions.append(AstStatement(StatementType.ACHIEVE_GOAL, AstLiteral("check_triggers")))
for goal in phase.goals:
actions.append(AstStatement(StatementType.ACHIEVE_GOAL, self._astify(goal)))
@@ -329,6 +457,22 @@ class AgentSpeakGenerator:
continues_response: bool = False,
main_goal: bool = False,
) -> None:
"""
Processes a goal and creates plans for achieving it.
This method creates two plans for each goal:
1. A main plan that executes the goal's steps when conditions are met
2. A fallback plan that provides a default empty implementation (prevents crashes)
The method also recursively processes any subgoals contained within
the goal's plan.
:param goal: The goal to process.
:param phase: The phase this goal belongs to.
:param previous_goal: The previous goal in sequence (for dependency tracking).
:param continues_response: Whether this goal continues an existing response.
:param main_goal: Whether this is a main goal (for UI notification purposes).
"""
context: list[AstExpression] = [self._astify(phase)]
context.append(~self._astify(goal, achieved=True))
if previous_goal and previous_goal.can_fail:
@@ -371,14 +515,39 @@ class AgentSpeakGenerator:
prev_goal = subgoal
def _step_to_statement(self, step: PlanElement) -> AstStatement:
"""
Converts a plan step to an AgentSpeak statement.
This method transforms different types of plan elements into their
corresponding AgentSpeak statements. Goals and speech-related actions
become achieve-goal statements, while gesture actions become do-action
statements.
:param step: The plan element to convert.
:return: The corresponding AgentSpeak statement.
"""
match step:
# Note that SpeechAction gets included in the ACHIEVE_GOAL, since it's a goal internally
case Goal() | SpeechAction() | LLMAction() as a:
return AstStatement(StatementType.ACHIEVE_GOAL, self._astify(a))
case GestureAction() as a:
return AstStatement(StatementType.DO_ACTION, self._astify(a))
# TODO: separate handling of keyword and others
def _process_trigger(self, trigger: Trigger, phase: Phase) -> None:
"""
Processes a trigger and creates plans for its execution.
This method creates plans that execute when trigger conditions are met.
It handles both automatic triggering (when conditions are detected) and
manual forcing (from UI). The trigger execution includes:
1. Notifying the system about trigger start
2. Executing all trigger steps
3. Waiting briefly for UI display
4. Notifying the system about trigger end
:param trigger: The trigger to process.
:param phase: The phase this trigger belongs to.
"""
body = []
subgoals = []
@@ -395,7 +564,12 @@ class AgentSpeakGenerator:
subgoals.append(step)
# Arbitrary wait for UI to display nicely
body.append(AstStatement(StatementType.DO_ACTION, AstLiteral("wait", [AstNumber(2000)])))
body.append(
AstStatement(
StatementType.DO_ACTION,
AstLiteral("wait", [AstNumber(settings.behaviour_settings.trigger_time_to_wait)]),
)
)
body.append(
AstStatement(
@@ -420,6 +594,18 @@ class AgentSpeakGenerator:
self._process_goal(subgoal, phase, continues_response=True)
def _add_fallbacks(self):
"""
Adds fallback plans for robust execution, preventing crashes.
This method creates fallback plans that provide default empty implementations
for key goals. These fallbacks ensure that the system can continue execution
even when no specific plans are applicable, preventing crashes.
The fallbacks are created for:
- check_triggers: When no triggers are applicable
- transition_phase: When phase transition conditions aren't met
- force_transition_phase: When forced transitions aren't possible
"""
# Trigger fallback
self._asp.plans.append(
AstPlan(
@@ -452,14 +638,44 @@ class AgentSpeakGenerator:
@singledispatchmethod
def _astify(self, element: ProgramElement) -> AstExpression:
"""
Converts program elements to AgentSpeak expressions (base method).
This is the base method for the singledispatch mechanism that handles
conversion of different program element types to their AgentSpeak
representations. Specific implementations are provided for each
element type through registered methods.
:param element: The program element to convert.
:return: The corresponding AgentSpeak expression.
:raises NotImplementedError: If no specific implementation exists for the element type.
"""
raise NotImplementedError(f"Cannot convert element {element} to an AgentSpeak expression.")
@_astify.register
def _(self, kwb: KeywordBelief) -> AstExpression:
"""
Converts a KeywordBelief to an AgentSpeak expression.
Keyword beliefs are converted to keyword_said literals that check
if the keyword was mentioned in user input.
:param kwb: The KeywordBelief to convert.
:return: An AstLiteral representing the keyword detection.
"""
return AstLiteral("keyword_said", [AstString(kwb.keyword)])
@_astify.register
def _(self, sb: SemanticBelief) -> AstExpression:
"""
Converts a SemanticBelief to an AgentSpeak expression.
Semantic beliefs are converted to literals using their slugified names,
which are used for LLM-based belief evaluation.
:param sb: The SemanticBelief to convert.
:return: An AstLiteral representing the semantic belief.
"""
return AstLiteral(self.slugify(sb))
@_astify.register
@@ -472,6 +688,15 @@ class AgentSpeakGenerator:
@_astify.register
def _(self, ib: InferredBelief) -> AstExpression:
"""
Converts an InferredBelief to an AgentSpeak expression.
Inferred beliefs are converted to binary operations that combine
their left and right operands using the appropriate logical operator.
:param ib: The InferredBelief to convert.
:return: An AstBinaryOp representing the logical combination.
"""
return AstBinaryOp(
self._astify(ib.left),
BinaryOperatorType.AND if ib.operator == LogicalOperator.AND else BinaryOperatorType.OR,
@@ -480,59 +705,187 @@ class AgentSpeakGenerator:
@_astify.register
def _(self, norm: Norm) -> AstExpression:
"""
Converts a Norm to an AgentSpeak expression.
Norms are converted to literals with either 'norm' or 'critical_norm'
functors depending on their critical flag, with the norm text as an argument.
Note that currently, critical norms are not yet functionally supported. They are possible
to astify for future use.
:param norm: The Norm to convert.
:return: An AstLiteral representing the norm.
"""
functor = "critical_norm" if norm.critical else "norm"
return AstLiteral(functor, [AstString(norm.norm)])
@_astify.register
def _(self, phase: Phase) -> AstExpression:
"""
Converts a Phase to an AgentSpeak expression.
Phases are converted to phase literals with their unique identifier
as an argument, which is used for phase tracking and transitions.
:param phase: The Phase to convert.
:return: An AstLiteral representing the phase.
"""
return AstLiteral("phase", [AstString(str(phase.id))])
@_astify.register
def _(self, goal: Goal, achieved: bool = False) -> AstExpression:
"""
Converts a Goal to an AgentSpeak expression.
Goals are converted to literals using their slugified names. If the
achieved parameter is True, the literal is prefixed with 'achieved_'.
:param goal: The Goal to convert.
:param achieved: Whether to represent this as an achieved goal.
:return: An AstLiteral representing the goal.
"""
return AstLiteral(f"{'achieved_' if achieved else ''}{self._slugify_str(goal.name)}")
@_astify.register
def _(self, trigger: Trigger) -> AstExpression:
"""
Converts a Trigger to an AgentSpeak expression.
Triggers are converted to literals using their slugified names,
which are used to identify and execute trigger plans.
:param trigger: The Trigger to convert.
:return: An AstLiteral representing the trigger.
"""
return AstLiteral(self.slugify(trigger))
@_astify.register
def _(self, sa: SpeechAction) -> AstExpression:
"""
Converts a SpeechAction to an AgentSpeak expression.
Speech actions are converted to say literals with the text content
as an argument, which are used for direct speech output.
:param sa: The SpeechAction to convert.
:return: An AstLiteral representing the speech action.
"""
return AstLiteral("say", [AstString(sa.text)])
@_astify.register
def _(self, ga: GestureAction) -> AstExpression:
"""
Converts a GestureAction to an AgentSpeak expression.
Gesture actions are converted to gesture literals with the gesture
type and name as arguments, which are used for physical robot gestures.
:param ga: The GestureAction to convert.
:return: An AstLiteral representing the gesture action.
"""
gesture = ga.gesture
return AstLiteral("gesture", [AstString(gesture.type), AstString(gesture.name)])
@_astify.register
def _(self, la: LLMAction) -> AstExpression:
"""
Converts an LLMAction to an AgentSpeak expression.
LLM actions are converted to reply_with_goal literals with the
conversational goal as an argument, which are used for LLM-generated
responses guided by specific goals.
:param la: The LLMAction to convert.
:return: An AstLiteral representing the LLM action.
"""
return AstLiteral("reply_with_goal", [AstString(la.goal)])
@singledispatchmethod
@staticmethod
def slugify(element: ProgramElement) -> str:
"""
Converts program elements to slugs (base method).
This is the base method for the singledispatch mechanism that handles
conversion of different program element types to their slug representations.
Specific implementations are provided for each element type through
registered methods.
Slugs are used outside of AgentSpeak, mostly for identifying what to send to the AgentSpeak
program as beliefs.
:param element: The program element to convert to a slug.
:return: The slug string representation.
:raises NotImplementedError: If no specific implementation exists for the element type.
"""
raise NotImplementedError(f"Cannot convert element {element} to a slug.")
@slugify.register
@staticmethod
def _(n: Norm) -> str:
"""
Converts a Norm to a slug.
Norms are converted to slugs with the 'norm_' prefix followed by
the slugified norm text.
:param n: The Norm to convert.
:return: The slug string representation.
"""
return f"norm_{AgentSpeakGenerator._slugify_str(n.norm)}"
@slugify.register
@staticmethod
def _(sb: SemanticBelief) -> str:
"""
Converts a SemanticBelief to a slug.
Semantic beliefs are converted to slugs with the 'semantic_' prefix
followed by the slugified belief name.
:param sb: The SemanticBelief to convert.
:return: The slug string representation.
"""
return f"semantic_{AgentSpeakGenerator._slugify_str(sb.name)}"
@slugify.register
@staticmethod
def _(g: BaseGoal) -> str:
"""
Converts a BaseGoal to a slug.
Goals are converted to slugs using their slugified names directly.
:param g: The BaseGoal to convert.
:return: The slug string representation.
"""
return AgentSpeakGenerator._slugify_str(g.name)
@slugify.register
@staticmethod
def _(t: Trigger):
def _(t: Trigger) -> str:
"""
Converts a Trigger to a slug.
Triggers are converted to slugs with the 'trigger_' prefix followed by
the slugified trigger name.
:param t: The Trigger to convert.
:return: The slug string representation.
"""
return f"trigger_{AgentSpeakGenerator._slugify_str(t.name)}"
@staticmethod
def _slugify_str(text: str) -> str:
"""
Converts a text string to a slug.
This helper method converts arbitrary text to a URL-friendly slug format
by converting to lowercase, removing special characters, and replacing
spaces with underscores. It also removes common stopwords.
:param text: The text string to convert.
:return: The slugified string.
"""
return slugify(text, separator="_", stopwords=["a", "an", "the", "we", "you", "I"])