Spaces:
Sleeping
Sleeping
fix: address remaining review issues (3/5)
Browse files- Drop cosmetic reranking 'running' event β reranking completes inside
tool execution, so only emit 'done' with results (issue #4)
- Replace for/else with explicit response.tool_calls check, matching
the pattern used by run() (issue #5)
- Fix test_stream_events_ordered to filter legacy events, accommodating
new stage events in the stream (issue #3)
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
- agent_bench/agents/orchestrator.py +5 -7
- tests/test_serving.py +10 -8
- tests/test_stream_stages.py +3 -1
agent_bench/agents/orchestrator.py
CHANGED
|
@@ -266,11 +266,8 @@ class Orchestrator:
|
|
| 266 |
"chunks_pre_rerank": pre_rerank,
|
| 267 |
})
|
| 268 |
|
| 269 |
-
# --- Reranking stage (
|
| 270 |
if pre_rerank > 0:
|
| 271 |
-
yield StreamEvent(type="stage", metadata={
|
| 272 |
-
"stage": "reranking", "status": "running", "iteration": iteration,
|
| 273 |
-
})
|
| 274 |
yield StreamEvent(type="stage", metadata={
|
| 275 |
"stage": "reranking", "status": "done", "iteration": iteration,
|
| 276 |
"chunks": result.metadata.get("chunks", []),
|
|
@@ -278,8 +275,10 @@ class Orchestrator:
|
|
| 278 |
|
| 279 |
if "sources" in result.metadata:
|
| 280 |
all_sources.extend(result.metadata["sources"])
|
| 281 |
-
|
| 282 |
-
|
|
|
|
|
|
|
| 283 |
yield StreamEvent(type="stage", metadata={
|
| 284 |
"stage": "llm", "status": "running", "iteration": iteration,
|
| 285 |
})
|
|
@@ -293,7 +292,6 @@ class Orchestrator:
|
|
| 293 |
"stage": "llm", "status": "done", "iteration": iteration,
|
| 294 |
})
|
| 295 |
|
| 296 |
-
# Handle max_iterations=0
|
| 297 |
if self.max_iterations == 0:
|
| 298 |
response = await self.provider.complete(
|
| 299 |
messages, tools=None, temperature=self.temperature
|
|
|
|
| 266 |
"chunks_pre_rerank": pre_rerank,
|
| 267 |
})
|
| 268 |
|
| 269 |
+
# --- Reranking stage (already completed inside tool execution) ---
|
| 270 |
if pre_rerank > 0:
|
|
|
|
|
|
|
|
|
|
| 271 |
yield StreamEvent(type="stage", metadata={
|
| 272 |
"stage": "reranking", "status": "done", "iteration": iteration,
|
| 273 |
"chunks": result.metadata.get("chunks", []),
|
|
|
|
| 275 |
|
| 276 |
if "sources" in result.metadata:
|
| 277 |
all_sources.extend(result.metadata["sources"])
|
| 278 |
+
|
| 279 |
+
# Max iterations hit β force text answer without tools
|
| 280 |
+
# (same pattern as run(): explicit call after loop)
|
| 281 |
+
if response.tool_calls:
|
| 282 |
yield StreamEvent(type="stage", metadata={
|
| 283 |
"stage": "llm", "status": "running", "iteration": iteration,
|
| 284 |
})
|
|
|
|
| 292 |
"stage": "llm", "status": "done", "iteration": iteration,
|
| 293 |
})
|
| 294 |
|
|
|
|
| 295 |
if self.max_iterations == 0:
|
| 296 |
response = await self.provider.complete(
|
| 297 |
messages, tools=None, temperature=self.temperature
|
tests/test_serving.py
CHANGED
|
@@ -451,7 +451,7 @@ class TestStreaming:
|
|
| 451 |
|
| 452 |
@pytest.mark.asyncio
|
| 453 |
async def test_stream_events_ordered(self, test_app):
|
| 454 |
-
"""
|
| 455 |
import json as json_mod
|
| 456 |
|
| 457 |
async with AsyncClient(
|
|
@@ -461,15 +461,17 @@ class TestStreaming:
|
|
| 461 |
"/ask/stream", json={"question": "How do path parameters work?"}
|
| 462 |
)
|
| 463 |
|
| 464 |
-
|
| 465 |
for line in response.text.strip().split("\n"):
|
| 466 |
if line.startswith("data: "):
|
| 467 |
-
|
| 468 |
-
|
| 469 |
-
|
| 470 |
-
|
| 471 |
-
assert
|
| 472 |
-
assert
|
|
|
|
|
|
|
| 473 |
|
| 474 |
@pytest.mark.asyncio
|
| 475 |
async def test_stream_chunks_assemble(self, test_app):
|
|
|
|
| 451 |
|
| 452 |
@pytest.mark.asyncio
|
| 453 |
async def test_stream_events_ordered(self, test_app):
|
| 454 |
+
"""Legacy event sequence preserved: sources β chunk* β done."""
|
| 455 |
import json as json_mod
|
| 456 |
|
| 457 |
async with AsyncClient(
|
|
|
|
| 461 |
"/ask/stream", json={"question": "How do path parameters work?"}
|
| 462 |
)
|
| 463 |
|
| 464 |
+
all_events = []
|
| 465 |
for line in response.text.strip().split("\n"):
|
| 466 |
if line.startswith("data: "):
|
| 467 |
+
all_events.append(json_mod.loads(line[6:]))
|
| 468 |
+
|
| 469 |
+
# Filter to legacy event types only (stage events are additive)
|
| 470 |
+
legacy = [e for e in all_events if e["type"] in ("sources", "chunk", "done", "_orchestrator_done")]
|
| 471 |
+
assert len(legacy) >= 3 # at least sources + 1 chunk + done
|
| 472 |
+
assert legacy[0]["type"] == "sources"
|
| 473 |
+
assert legacy[-1]["type"] in ("done", "_orchestrator_done")
|
| 474 |
+
assert all(e["type"] == "chunk" for e in legacy[1:-1])
|
| 475 |
|
| 476 |
@pytest.mark.asyncio
|
| 477 |
async def test_stream_chunks_assemble(self, test_app):
|
tests/test_stream_stages.py
CHANGED
|
@@ -47,7 +47,9 @@ class TestOrchestratorStageEvents:
|
|
| 47 |
|
| 48 |
stage_events = [e for e in events if e.type == "stage"]
|
| 49 |
reranking_events = [e for e in stage_events if e.metadata.get("stage") == "reranking"]
|
| 50 |
-
assert len(reranking_events) >= 1 #
|
|
|
|
|
|
|
| 51 |
|
| 52 |
@pytest.mark.asyncio
|
| 53 |
async def test_stream_emits_llm_stage(self, orchestrator):
|
|
|
|
| 47 |
|
| 48 |
stage_events = [e for e in events if e.type == "stage"]
|
| 49 |
reranking_events = [e for e in stage_events if e.metadata.get("stage") == "reranking"]
|
| 50 |
+
assert len(reranking_events) >= 1 # done event with chunk details
|
| 51 |
+
# Reranking completes inside tool execution, so only a done event is emitted
|
| 52 |
+
assert all(e.metadata.get("status") == "done" for e in reranking_events)
|
| 53 |
|
| 54 |
@pytest.mark.asyncio
|
| 55 |
async def test_stream_emits_llm_stage(self, orchestrator):
|