From e704ec5ed44b3d645ea86f47715d790c67bda3a1 Mon Sep 17 00:00:00 2001 From: Kasper Marinus Date: Tue, 16 Dec 2025 17:00:32 +0100 Subject: [PATCH] feat: basic flow and phase transitions ref: N25B-376 --- .../agents/bdi/bdi_program_manager.py | 106 ++++++++++++++---- src/control_backend/schemas/program.py | 2 - 2 files changed, 85 insertions(+), 23 deletions(-) diff --git a/src/control_backend/agents/bdi/bdi_program_manager.py b/src/control_backend/agents/bdi/bdi_program_manager.py index 0eae52a..11c0b00 100644 --- a/src/control_backend/agents/bdi/bdi_program_manager.py +++ b/src/control_backend/agents/bdi/bdi_program_manager.py @@ -173,12 +173,20 @@ class AgentSpeakGenerator: Converts Pydantic representation of behavior programs into AgentSpeak(L) code string. """ + arrow_prefix = f"{' ' * 2}<-{' ' * 2}" + colon_prefix = f"{' ' * 2}:{' ' * 3}" + indent_prefix = " " * 6 + def generate(self, program: Program) -> str: lines = [] lines.append("") lines += self._generate_initial_beliefs(program) + lines += self._generate_basic_flow(program) + + lines += self._generate_phase_transitions(program) + lines += self._generate_norms(program) lines += self._generate_belief_inference(program) @@ -192,10 +200,60 @@ class AgentSpeakGenerator: def _generate_initial_beliefs(self, program: Program) -> Iterable[str]: yield "// --- Initial beliefs ---" - yield f"phase({program.phases[0].id})." + yield "phase(start)." yield from ["", ""] + def _generate_basic_flow(self, program: Program) -> Iterable[str]: + yield "// --- Basic flow ---" + + for phase in program.phases: + yield from self._generate_basic_flow_per_phase(phase) + + yield from ["", ""] + + def _generate_basic_flow_per_phase(self, phase: Phase) -> Iterable[str]: + yield "+user_said(Message)" + yield f"{self.colon_prefix}phase({phase.id})" + + goals = phase.goals + if goals: + yield f"{self.arrow_prefix}{self._slugify(goals[0], include_prefix=True)}" + for goal in goals[1:]: + yield f"{self.indent_prefix}{self._slugify(goal, include_prefix=True)}" + + yield f"{self.indent_prefix if goals else self.arrow_prefix}!transition_phase." + + def _generate_phase_transitions(self, program: Program) -> Iterable[str]: + yield "// --- Phase transitions ---" + + if len(program.phases) == 0: + yield from ["", ""] + return + + # TODO: remove outdated things + + for i in range(-1, len(program.phases)): + predecessor = program.phases[i] if i >= 0 else None + successor = program.phases[i + 1] if i < len(program.phases) - 1 else None + yield from self._generate_phase_transition(predecessor, successor) + + yield from self._generate_phase_transition(None, None) # to avoid failing plan + + yield from ["", ""] + + def _generate_phase_transition( + self, phase: Phase | None = None, next_phase: Phase | None = None + ) -> Iterable[str]: + yield "+!transition_phase" + + if phase is None and next_phase is None: # base case true to avoid failing plan + yield f"{self.arrow_prefix}true." + return + + yield f"{self.colon_prefix}phase({phase.id if phase else 'start'})" + yield f"{self.arrow_prefix}-+phase({next_phase.id if next_phase else 'end'})." + def _generate_norms(self, program: Program) -> Iterable[str]: yield "// --- Norms ---" @@ -259,46 +317,49 @@ class AgentSpeakGenerator: yield f"+{self._slugify(goal, include_prefix=True)}" # Context - yield f"{' ' * 2}:{' ' * 3}phase({phase.id}) &" - yield f"{' ' * 6}not responded_this_turn &" - yield f"{' ' * 6}not achieved_{self._slugify(goal)} &" + yield f"{self.colon_prefix}phase({phase.id}) &" + yield f"{self.indent_prefix}not responded_this_turn &" + yield f"{self.indent_prefix}not achieved_{self._slugify(goal)} &" if previous_goal: - yield f"{' ' * 6}achieved_{self._slugify(previous_goal)}" + yield f"{self.indent_prefix}achieved_{self._slugify(previous_goal)}" else: - yield f"{' ' * 6}true" + yield f"{self.indent_prefix}true" extra_goals_to_generate = [] steps = goal.plan.steps if len(steps) == 0: - yield f"{' ' * 2}<-{' ' * 2}true." + yield f"{self.arrow_prefix}true." return first_step = steps[0] yield ( - f"{' ' * 2}<-{' ' * 2}{self._slugify(first_step, include_prefix=True)}" + f"{self.arrow_prefix}{self._slugify(first_step, include_prefix=True)}" f"{'.' if len(steps) == 1 and goal.can_fail else ';'}" ) if isinstance(first_step, Goal): extra_goals_to_generate.append(first_step) for step in steps[1:-1]: - yield f"{' ' * 6}{self._slugify(step, include_prefix=True)};" + yield f"{self.indent_prefix}{self._slugify(step, include_prefix=True)};" if isinstance(step, Goal): extra_goals_to_generate.append(step) if len(steps) > 1: last_step = steps[-1] yield ( - f"{' ' * 6}{self._slugify(last_step, include_prefix=True)}" + f"{self.indent_prefix}{self._slugify(last_step, include_prefix=True)}" f"{'.' if goal.can_fail else ';'}" ) if isinstance(last_step, Goal): extra_goals_to_generate.append(last_step) if not goal.can_fail: - yield f"{' ' * 6}+achieved_{self._slugify(goal)}." + yield f"{self.indent_prefix}+achieved_{self._slugify(goal)}." + + yield f"+{self._slugify(goal, include_prefix=True)}" + yield f"{self.arrow_prefix}true." yield "" @@ -320,32 +381,32 @@ class AgentSpeakGenerator: belief_name = self._slugify(trigger.condition) yield f"+{belief_name}" - yield f"{' ' * 2}:{' ' * 3}phase({phase.id})" + yield f"{self.colon_prefix}phase({phase.id})" extra_goals_to_generate = [] steps = trigger.plan.steps if len(steps) == 0: - yield f"{' ' * 2}<-{' ' * 2}true." + yield f"{self.arrow_prefix}true." return first_step = steps[0] yield ( - f"{' ' * 2}<-{' ' * 2}{self._slugify(first_step, include_prefix=True)}" + f"{self.arrow_prefix}{self._slugify(first_step, include_prefix=True)}" f"{'.' if len(steps) == 1 else ';'}" ) if isinstance(first_step, Goal): extra_goals_to_generate.append(first_step) for step in steps[1:-1]: - yield f"{' ' * 6}{self._slugify(step, include_prefix=True)};" + yield f"{self.indent_prefix}{self._slugify(step, include_prefix=True)};" if isinstance(step, Goal): extra_goals_to_generate.append(step) if len(steps) > 1: last_step = steps[-1] - yield f"{' ' * 6}{self._slugify(last_step, include_prefix=True)}." + yield f"{self.indent_prefix}{self._slugify(last_step, include_prefix=True)}." if isinstance(last_step, Goal): extra_goals_to_generate.append(last_step) @@ -366,33 +427,36 @@ class AgentSpeakGenerator: steps = goal.plan.steps if len(steps) == 0: - yield f"{' ' * 2}<-{' ' * 2}true." + yield f"{self.arrow_prefix}true." return first_step = steps[0] yield ( - f"{' ' * 2}<-{' ' * 2}{self._slugify(first_step, include_prefix=True)}" + f"{self.arrow_prefix}{self._slugify(first_step, include_prefix=True)}" f"{'.' if len(steps) == 1 and goal.can_fail else ';'}" ) if isinstance(first_step, Goal): extra_goals_to_generate.append(first_step) for step in steps[1:-1]: - yield f"{' ' * 6}{self._slugify(step, include_prefix=True)};" + yield f"{self.indent_prefix}{self._slugify(step, include_prefix=True)};" if isinstance(step, Goal): extra_goals_to_generate.append(step) if len(steps) > 1: last_step = steps[-1] yield ( - f"{' ' * 6}{self._slugify(last_step, include_prefix=True)}" + f"{self.indent_prefix}{self._slugify(last_step, include_prefix=True)}" f"{'.' if goal.can_fail else ';'}" ) if isinstance(last_step, Goal): extra_goals_to_generate.append(last_step) if not goal.can_fail: - yield f"{' ' * 6}+achieved_{self._slugify(goal)}." + yield f"{self.indent_prefix}+achieved_{self._slugify(goal)}." + + yield f"+{self._slugify(goal, include_prefix=True)}" + yield f"{self.arrow_prefix}true." yield "" diff --git a/src/control_backend/schemas/program.py b/src/control_backend/schemas/program.py index d02923e..605694b 100644 --- a/src/control_backend/schemas/program.py +++ b/src/control_backend/schemas/program.py @@ -90,8 +90,6 @@ class ConditionalNorm(BasicNorm): :ivar condition: When to activate this norm. """ - name: str = "" - id: int = -1 condition: Belief