feat: integrate AgentSpeak with semantic belief extraction

ref: N25B-429
This commit is contained in:
Twirre Meulenbelt
2026-01-07 11:44:48 +01:00
parent de8e829d3e
commit cabe35cdbd
10 changed files with 120 additions and 60 deletions

View File

@@ -20,7 +20,7 @@ def mock_agentspeak_env():
@pytest.fixture
def agent():
agent = BDICoreAgent("bdi_agent", "dummy.asl")
agent = BDICoreAgent("bdi_agent")
agent.send = AsyncMock()
agent.bdi_agent = MagicMock()
return agent
@@ -133,14 +133,14 @@ async def test_custom_actions(agent):
# Invoke action
mock_term = MagicMock()
mock_term.args = ["Hello", "Norm", "Goal"]
mock_term.args = ["Hello", "Norm"]
mock_intention = MagicMock()
# Run generator
gen = action_fn(agent, mock_term, mock_intention)
next(gen) # Execute
agent._send_to_llm.assert_called_with("Hello", "Norm", "Goal")
agent._send_to_llm.assert_called_with("Hello", "Norm", "")
def test_add_belief_sets_event(agent):

View File

@@ -32,6 +32,8 @@ def make_valid_program_json(norm="N1", goal="G1") -> str:
Goal(
id=uuid.uuid4(),
name=goal,
description="This description can be used to determine whether the goal "
"has been achieved.",
plan=Plan(
id=uuid.uuid4(),
name="Goal Plan",
@@ -75,6 +77,7 @@ async def test_receive_programs_valid_and_invalid():
]
manager = BDIProgramManager(name="program_manager_test")
manager._internal_pub_socket = AsyncMock()
manager.sub_socket = sub
manager._create_agentspeak_and_send_to_bdi = AsyncMock()

View File

@@ -8,9 +8,11 @@ import pytest
from control_backend.agents.bdi import TextBeliefExtractorAgent
from control_backend.core.agent_system import InternalMessage
from control_backend.core.config import settings
from control_backend.schemas.belief_list import BeliefList
from control_backend.schemas.belief_message import BeliefMessage
from control_backend.schemas.program import (
ConditionalNorm,
KeywordBelief,
LLMAction,
Phase,
Plan,
@@ -186,13 +188,31 @@ async def test_retry_query_llm_fail_immediately(agent):
@pytest.mark.asyncio
async def test_extracting_beliefs_from_program(agent, sample_program):
async def test_extracting_semantic_beliefs(agent):
"""
The Program Manager sends beliefs to this agent. Test whether the agent handles them correctly.
"""
assert len(agent.available_beliefs) == 0
beliefs = BeliefList(
beliefs=[
KeywordBelief(
id=uuid.uuid4(),
name="keyword_hello",
keyword="hello",
),
SemanticBelief(
id=uuid.uuid4(), name="semantic_hello_1", description="Some semantic belief 1"
),
SemanticBelief(
id=uuid.uuid4(), name="semantic_hello_2", description="Some semantic belief 2"
),
]
)
await agent.handle_message(
InternalMessage(
to=settings.agent_settings.text_belief_extractor_name,
sender=settings.agent_settings.bdi_program_manager_name,
body=sample_program.model_dump_json(),
body=beliefs.model_dump_json(),
),
)
assert len(agent.available_beliefs) == 2

View File

@@ -43,6 +43,8 @@ def make_valid_program_dict():
Goal(
id=uuid.uuid4(),
name="Some goal",
description="This description can be used to determine whether the goal "
"has been achieved.",
plan=Plan(
id=uuid.uuid4(),
name="Goal Plan",

View File

@@ -31,6 +31,7 @@ def base_goal() -> Goal:
return Goal(
id=uuid.uuid4(),
name="testGoalName",
description="This description can be used to determine whether the goal has been achieved.",
plan=Plan(
id=uuid.uuid4(),
name="testGoalPlanName",