## Summary - Follow-up to [#71](#71) / [#63](#63). - **Bugfix:** `run_agentic_turn` discarded `ws_send`, so `run_started` / `plan_ready` / `step_*` / `run_completed` frames only hit an empty Redis group and never the connected client during inline agent runs. - Thread `ws_send` through `execute_agent_run` / `_broadcast` (WS + channel-layer group). - Route agentic turns in `consumers_graph` the same way as `consumers.py`. - Add offline `test_agent_orchestrator.py` coverage. ## Test plan - [x] `SKIP_RAG_INIT=1 uv run python manage.py test chat_backend.tests.test_agent_orchestrator chat_backend.tests.test_agent_tools chat_backend.tests.test_consumers` - [ ] Manual: `ALLOW_AGENTIC_TASKS=true`, multi-step prompt — confirm agent frames arrive on the live WS - [ ] Manual: graph consumer path (`ws/conditional_chat/`) also routes agentic turnsReviewed-on: #72
This commit was merged in pull request #72.
This commit is contained in:
westfarn committed
1 parent
093e5462a5
commit
bef151c92c
3 files changed
+457
-36
No files matched your search
@@ -532,44 +532,89 @@ class ChatConsumerGraph(AsyncWebsocketConsumer):
|
||||
status_token = set_status_emitter(_send_status)
|
||||
try:
|
||||
await emit_status("queued")
|
||||
# Run Graph (moderation emits moderating; grounding emits evaluating/…)
|
||||
final_state = await app.ainvoke(initial_state)
|
||||
print("Final State: ", final_state)
|
||||
|
||||
response_generator_or_dict = final_state["response_generator"]
|
||||
print("Response Generator: ", response_generator_or_dict)
|
||||
# Agentic path (#63) — same gate as consumers.py; skip the
|
||||
# fixed LangGraph when the heuristic says multi-step.
|
||||
from chat_backend.services.agent import (
|
||||
run_agentic_turn,
|
||||
should_use_agent,
|
||||
)
|
||||
|
||||
full_response = ""
|
||||
tokens_in = tokens_out = None
|
||||
citations = []
|
||||
final_model = resolved_model
|
||||
|
||||
if isinstance(response_generator_or_dict, dict):
|
||||
content = response_generator_or_dict.get("content", "")
|
||||
await self.send_json_message(
|
||||
json.dumps(response_generator_or_dict)
|
||||
if should_use_agent(message):
|
||||
try:
|
||||
await enforce_feature_gate(chat_user, "agentic_tasks")
|
||||
except FeatureNotAllowed as exc:
|
||||
await self.send_json_message(
|
||||
json.dumps(
|
||||
{
|
||||
"type": "error",
|
||||
"code": exc.code,
|
||||
"content": exc.message,
|
||||
}
|
||||
)
|
||||
)
|
||||
await self.send("END_OF_THE_STREAM_ENDER_GAME_42")
|
||||
return
|
||||
|
||||
async def _ws_send(raw: str):
|
||||
await self.send_json_message(raw)
|
||||
|
||||
_run, answer = await run_agentic_turn(
|
||||
user=chat_user,
|
||||
scope=tenant_scope,
|
||||
conversation_id=conversation_id,
|
||||
goal=message,
|
||||
prompt=prompt_instance,
|
||||
ws_send=_ws_send,
|
||||
)
|
||||
full_response = content
|
||||
tokens_in, tokens_out = extract_token_usage(
|
||||
response_generator_or_dict
|
||||
)
|
||||
else:
|
||||
final_model = _run.model_orchestrator or resolved_model
|
||||
await emit_status("writing")
|
||||
usage = TokenUsageCollector()
|
||||
async for chunk in aiter_text_chunks(
|
||||
response_generator_or_dict, usage
|
||||
):
|
||||
full_response += chunk
|
||||
await self.send_json_message(chunk)
|
||||
tokens_in, tokens_out = usage.pair
|
||||
await self.send_json_message(answer)
|
||||
full_response = answer
|
||||
else:
|
||||
# Run Graph (moderation emits moderating; grounding emits evaluating/…)
|
||||
final_state = await app.ainvoke(initial_state)
|
||||
print("Final State: ", final_state)
|
||||
|
||||
response_generator_or_dict = final_state["response_generator"]
|
||||
print("Response Generator: ", response_generator_or_dict)
|
||||
|
||||
if isinstance(response_generator_or_dict, dict):
|
||||
content = response_generator_or_dict.get("content", "")
|
||||
await self.send_json_message(
|
||||
json.dumps(response_generator_or_dict)
|
||||
)
|
||||
full_response = content
|
||||
tokens_in, tokens_out = extract_token_usage(
|
||||
response_generator_or_dict
|
||||
)
|
||||
else:
|
||||
await emit_status("writing")
|
||||
usage = TokenUsageCollector()
|
||||
async for chunk in aiter_text_chunks(
|
||||
response_generator_or_dict, usage
|
||||
):
|
||||
full_response += chunk
|
||||
await self.send_json_message(chunk)
|
||||
tokens_in, tokens_out = usage.pair
|
||||
|
||||
citations = final_state.get("citations") or []
|
||||
final_model = (
|
||||
final_state.get("resolved_model") or resolved_model
|
||||
)
|
||||
|
||||
await self.send("END_OF_THE_STREAM_ENDER_GAME_42")
|
||||
|
||||
citations = final_state.get("citations") or []
|
||||
if citations:
|
||||
await self.send_json_message(
|
||||
json.dumps(citations_frame(citations))
|
||||
)
|
||||
|
||||
final_model = final_state.get("resolved_model") or resolved_model
|
||||
if final_model and final_model != prompt_metric.model_name:
|
||||
prompt_metric.model_name = final_model
|
||||
await database_sync_to_async(prompt_metric.save)(
|
||||
|
||||
Reference in new issue
Block a user