From a8da8b4aaa7b575896768f0b6da2c33db04c9481 Mon Sep 17 00:00:00 2001 From: Wendong-Fan Date: Fri, 9 Jan 2026 03:25:05 +0800 Subject: [PATCH] fix: stream output --- backend/app/service/chat_service.py | 83 +++++++++++++++--------- backend/app/utils/agent.py | 11 ++-- backend/app/utils/single_agent_worker.py | 20 ++++-- backend/app/utils/workforce.py | 6 ++ 4 files changed, 78 insertions(+), 42 deletions(-) diff --git a/backend/app/service/chat_service.py b/backend/app/service/chat_service.py index a648efad..afbbc169 100644 --- a/backend/app/service/chat_service.py +++ b/backend/app/service/chat_service.py @@ -435,7 +435,7 @@ async def step_solve(options: Chat, request: Request, task_lock: TaskLock): # Stream decomposition in background so queue items (decompose_text) are processed immediately logger.info(f"[NEW-QUESTION] 🧩 Starting task decomposition via workforce.eigent_make_sub_tasks") - stream_state = {"subtasks": [], "seen_ids": set()} + stream_state = {"subtasks": [], "seen_ids": set(), "last_content": ""} state_holder: dict[str, Any] = {"sub_tasks": [], "summary_task": ""} def on_stream_batch(new_tasks: list[Task], is_final: bool = False): @@ -446,20 +446,32 @@ async def step_solve(options: Chat, request: Request, task_lock: TaskLock): def on_stream_text(chunk): try: - # Extract content from chunk object (CAMEL now passes chunk instead of accumulated content) - text_content = chunk.msg.content if hasattr(chunk, 'msg') and chunk.msg else str(chunk) - asyncio.run_coroutine_threadsafe( - task_lock.put_queue( - ActionDecomposeTextData( - data={ - "project_id": options.project_id, - "task_id": options.task_id, - "content": text_content, - } - ) - ), - event_loop, - ) + # With task_agent using stream_accumulate=True, chunk.msg.content is accumulated content + # We need to calculate the delta to send only new content to frontend + accumulated_content = chunk.msg.content if hasattr(chunk, 'msg') and chunk.msg else str(chunk) + last_content = stream_state["last_content"] + + # Calculate delta: new content that wasn't in the previous chunk + if accumulated_content.startswith(last_content): + delta_content = accumulated_content[len(last_content):] + else: + delta_content = accumulated_content + + stream_state["last_content"] = accumulated_content + + if delta_content: + asyncio.run_coroutine_threadsafe( + task_lock.put_queue( + ActionDecomposeTextData( + data={ + "project_id": options.project_id, + "task_id": options.task_id, + "content": delta_content, + } + ) + ), + event_loop, + ) except Exception as e: logger.warning(f"Failed to stream decomposition text: {e}") @@ -769,7 +781,7 @@ async def step_solve(options: Chat, request: Request, task_lock: TaskLock): context_for_multi_turn = build_context_for_workforce(task_lock, options) logger.info(f"[LIFECYCLE] Multi-turn: calling workforce.handle_decompose_append_task for new task decomposition") - stream_state = {"subtasks": [], "seen_ids": set()} + stream_state = {"subtasks": [], "seen_ids": set(), "last_content": ""} def on_stream_batch(new_tasks: list[Task], is_final: bool = False): fresh_tasks = [t for t in new_tasks if t.id not in stream_state["seen_ids"]] @@ -779,20 +791,31 @@ async def step_solve(options: Chat, request: Request, task_lock: TaskLock): def on_stream_text(chunk): try: - # Extract content from chunk object (CAMEL now passes chunk instead of accumulated content) - text_content = chunk.msg.content if hasattr(chunk, 'msg') and chunk.msg else str(chunk) - asyncio.run_coroutine_threadsafe( - task_lock.put_queue( - ActionDecomposeTextData( - data={ - "project_id": options.project_id, - "task_id": options.task_id, - "content": text_content, - } - ) - ), - event_loop, - ) + # With task_agent using stream_accumulate=True, chunk.msg.content is accumulated content + # We need to calculate the delta to send only new content to frontend + accumulated_content = chunk.msg.content if hasattr(chunk, 'msg') and chunk.msg else str(chunk) + last_content = stream_state["last_content"] + + if accumulated_content.startswith(last_content): + delta_content = accumulated_content[len(last_content):] + else: + delta_content = accumulated_content + + stream_state["last_content"] = accumulated_content + + if delta_content: + asyncio.run_coroutine_threadsafe( + task_lock.put_queue( + ActionDecomposeTextData( + data={ + "project_id": options.project_id, + "task_id": options.task_id, + "content": delta_content, + } + ) + ), + event_loop, + ) except Exception as e: logger.warning(f"Failed to stream decomposition text: {e}") new_sub_tasks = await workforce.handle_decompose_append_task( diff --git a/backend/app/utils/agent.py b/backend/app/utils/agent.py index 69fdec25..a3494498 100644 --- a/backend/app/utils/agent.py +++ b/backend/app/utils/agent.py @@ -187,17 +187,18 @@ class ListenChatAgent(ChatAgent): if isinstance(res, StreamingChatAgentResponse): def _stream_with_deactivate(): last_response: ChatAgentResponse | None = None + # With stream_accumulate=False, we need to accumulate delta content + accumulated_content = "" try: for chunk in res: last_response = chunk + # Accumulate content from each chunk (delta mode) + if chunk.msg and chunk.msg.content: + accumulated_content += chunk.msg.content yield chunk finally: - final_message = "" total_tokens = 0 if last_response: - final_message = ( - last_response.msg.content if last_response.msg else "" - ) usage_info = ( last_response.info.get("usage") or last_response.info.get("token_usage") @@ -212,7 +213,7 @@ class ListenChatAgent(ChatAgent): "agent_name": self.agent_name, "process_task_id": self.process_task_id, "agent_id": self.agent_id, - "message": final_message, + "message": accumulated_content, "tokens": total_tokens, }, ) diff --git a/backend/app/utils/single_agent_worker.py b/backend/app/utils/single_agent_worker.py index a6e82871..5f3fae8f 100644 --- a/backend/app/utils/single_agent_worker.py +++ b/backend/app/utils/single_agent_worker.py @@ -104,11 +104,12 @@ class SingleAgentWorker(BaseSingleAgentWorker): # Handle streaming response if isinstance(response, AsyncStreamingChatAgentResponse): - content = "" + # With stream_accumulate=False, we need to accumulate delta content + accumulated_content = "" async for chunk in response: - if chunk.msg: - content = chunk.msg.content - response_content = content + if chunk.msg and chunk.msg.content: + accumulated_content += chunk.msg.content + response_content = accumulated_content else: # Regular ChatAgentResponse response_content = response.msg.content if response.msg else "" @@ -128,10 +129,15 @@ class SingleAgentWorker(BaseSingleAgentWorker): # Handle streaming response for native output if isinstance(response, AsyncStreamingChatAgentResponse): task_result = None + # With stream_accumulate=False, we need to accumulate delta content + accumulated_content = "" async for chunk in response: - if chunk.msg and chunk.msg.parsed: - task_result = chunk.msg.parsed - response_content = chunk.msg.content + if chunk.msg: + if chunk.msg.content: + accumulated_content += chunk.msg.content + if chunk.msg.parsed: + task_result = chunk.msg.parsed + response_content = accumulated_content # If no parsed result found in streaming, create fallback if task_result is None: task_result = TaskResult( diff --git a/backend/app/utils/workforce.py b/backend/app/utils/workforce.py index 92d1ea77..d5bbb888 100644 --- a/backend/app/utils/workforce.py +++ b/backend/app/utils/workforce.py @@ -64,6 +64,8 @@ class Workforce(BaseWorkforce): enabled_strategies=["retry", "replan"], ), ) + self.task_agent.stream_accumulate = True + self.task_agent._stream_accumulate_explicit = True logger.info(f"[WF-LIFECYCLE] ✅ Workforce.__init__ COMPLETED, id={id(self)}") def eigent_make_sub_tasks( @@ -260,6 +262,10 @@ class Workforce(BaseWorkforce): except Exception as e: logger.warning(f"Streaming callback failed: {e}") logger.info(f"[DECOMPOSE] Collected {len(subtasks)} subtasks from generator") + + # After consuming the generator, check task.subtasks for final result as fallback + if not subtasks and task.subtasks: + subtasks = task.subtasks else: subtasks = subtasks_result logger.info(f"[DECOMPOSE] Got {len(subtasks) if subtasks else 0} subtasks directly")