mirror of
https://github.com/eigent-ai/eigent.git
synced 2026-08-16 12:13:31 +00:00
fix: stream output
This commit is contained in:
Wendong-Fan
parent
6fcb328d11
commit
a8da8b4aaa
4 changed files with 78 additions and 42 deletions
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
},
|
||||
)
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -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")
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue