48 lines
1.5 KiB
Python
48 lines
1.5 KiB
Python
import os
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
|
|
|
|
PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
if PROJECT_ROOT not in sys.path:
|
|
sys.path.insert(0, PROJECT_ROOT)
|
|
|
|
|
|
from SimpleLLMFunc.hooks.events import ReActEventType, ReactEndEvent
|
|
from SimpleLLMFunc.hooks.stream import EventYield, ResponseYield
|
|
|
|
from tools.subagent_utils import run_subagent_with_events
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_subagent_with_events_ignores_empty_chunk_repr_and_uses_react_end_text():
|
|
async def fake_specialist(**kwargs):
|
|
yield ResponseYield(response="", messages=[])
|
|
yield EventYield(
|
|
event=ReactEndEvent(
|
|
event_type=ReActEventType.REACT_END,
|
|
timestamp=datetime.now(timezone.utc),
|
|
trace_id="trace-1",
|
|
func_name="specialist",
|
|
iteration=1,
|
|
final_response="## Refined User Requirements\n- final answer",
|
|
final_messages=[],
|
|
total_iterations=1,
|
|
total_execution_time=0.1,
|
|
total_tool_calls=0,
|
|
total_llm_calls=1,
|
|
total_token_usage=None,
|
|
)
|
|
)
|
|
|
|
result = await run_subagent_with_events(
|
|
specialist_callable=fake_specialist,
|
|
specialist_kwargs={"message": "test", "history": []},
|
|
subagent_label="Requirement Refinement Specialist",
|
|
event_emitter=None,
|
|
)
|
|
|
|
assert result == "## Refined User Requirements\n- final answer"
|