mirror of
https://github.com/eigent-ai/eigent.git
synced 2026-08-22 15:13:35 +00:00
update multi turn
This commit is contained in:
puzhen
parent
3f0e0c36f4
commit
bc114f39ce
8 changed files with 492 additions and 52 deletions
|
|
@ -70,10 +70,26 @@ async def post(data: Chat, request: Request):
|
|||
def improve(id: str, data: SupplementChat):
|
||||
chat_logger.info(f"Improving chat for task_id: {id} with question: {data.question}")
|
||||
task_lock = get_task_lock(id)
|
||||
|
||||
# Allow continuing conversation even after task is done
|
||||
# This supports multi-turn conversation after complex task completion
|
||||
if task_lock.status == Status.done:
|
||||
raise UserException(code.error, "Task was done")
|
||||
chat_logger.info(f"[CONTEXT] Task {id} was done, restarting for context-aware conversation")
|
||||
# Reset status to allow processing new messages
|
||||
task_lock.status = Status.confirming
|
||||
# Clear any existing background tasks since workforce was stopped
|
||||
if hasattr(task_lock, 'background_tasks'):
|
||||
task_lock.background_tasks.clear()
|
||||
# Note: conversation_history and last_task_result are preserved
|
||||
|
||||
# Log context preservation
|
||||
if hasattr(task_lock, 'conversation_history'):
|
||||
chat_logger.info(f"[CONTEXT] Preserved {len(task_lock.conversation_history)} conversation entries")
|
||||
if hasattr(task_lock, 'last_task_result'):
|
||||
chat_logger.info(f"[CONTEXT] Preserved task result: {len(task_lock.last_task_result)} chars")
|
||||
|
||||
asyncio.run(task_lock.put_queue(ActionImproveData(data=data.question)))
|
||||
chat_logger.info(f"Improvement request queued for task_id: {id}")
|
||||
chat_logger.info(f"Improvement request queued with preserved context")
|
||||
return Response(status_code=201)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -102,6 +102,36 @@ def collect_previous_task_context(working_directory: str, previous_task_content:
|
|||
return "\n".join(context_parts)
|
||||
|
||||
|
||||
def build_context_for_workforce(task_lock: TaskLock, options: Chat) -> str:
|
||||
"""Build context information for workforce"""
|
||||
context = ""
|
||||
|
||||
# Add conversation history
|
||||
if task_lock.conversation_history:
|
||||
context = "=== CONVERSATION HISTORY ===\n"
|
||||
|
||||
# Only include recent conversations to avoid overly long context
|
||||
for entry in task_lock.conversation_history[-10:]:
|
||||
if entry['role'] == 'task_result':
|
||||
# Simplify task result display
|
||||
context += f"[Previous Task Completed]\n"
|
||||
else:
|
||||
context += f"{entry['role']}: {entry['content'][:200]}\n"
|
||||
|
||||
context += "\n"
|
||||
|
||||
# Add previous task's detailed result
|
||||
if task_lock.last_task_result:
|
||||
context += collect_previous_task_context(
|
||||
working_directory=options.file_save_path(),
|
||||
previous_task_content="",
|
||||
previous_task_result=task_lock.last_task_result,
|
||||
previous_summary=task_lock.last_task_summary
|
||||
)
|
||||
|
||||
return context
|
||||
|
||||
|
||||
@sync_step
|
||||
async def step_solve(options: Chat, request: Request, task_lock: TaskLock):
|
||||
# if True:
|
||||
|
|
@ -112,11 +142,41 @@ async def step_solve(options: Chat, request: Request, task_lock: TaskLock):
|
|||
# faulthandler.dump_traceback_later(second)
|
||||
|
||||
start_event_loop = True
|
||||
question_agent = question_confirm_agent(options)
|
||||
|
||||
# ========== Initialize context management ==========
|
||||
# Initialize context fields if they don't exist
|
||||
if not hasattr(task_lock, 'conversation_history'):
|
||||
task_lock.conversation_history = []
|
||||
if not hasattr(task_lock, 'last_task_result'):
|
||||
task_lock.last_task_result = ""
|
||||
if not hasattr(task_lock, 'last_task_summary'):
|
||||
task_lock.last_task_summary = ""
|
||||
if not hasattr(task_lock, 'question_agent'):
|
||||
task_lock.question_agent = None
|
||||
|
||||
# Create or reuse persistent question_agent
|
||||
if task_lock.question_agent is None:
|
||||
task_lock.question_agent = question_confirm_agent(options)
|
||||
logger.info(f"[CONTEXT] Created new persistent question_agent for project {options.project_id}")
|
||||
else:
|
||||
logger.info(f"[CONTEXT] Reusing existing question_agent with {len(task_lock.conversation_history)} history entries")
|
||||
|
||||
question_agent = task_lock.question_agent
|
||||
|
||||
# Other variables
|
||||
camel_task = None
|
||||
workforce = None
|
||||
last_completed_task_result = "" # Track the last completed task result
|
||||
summary_task_content = "" # Track task summary
|
||||
loop_iteration = 0
|
||||
|
||||
logger.info(f"[TRACE] === STARTING MAIN LOOP for project {options.project_id} ===")
|
||||
logger.info(f"[CONTEXT] Starting with {len(task_lock.conversation_history)} previous conversations")
|
||||
|
||||
while True:
|
||||
loop_iteration += 1
|
||||
logger.info(f"[TRACE] Main loop iteration {loop_iteration}, waiting for action...")
|
||||
|
||||
if await request.is_disconnected():
|
||||
logger.warning(f"Client disconnected for project {options.project_id}")
|
||||
if workforce is not None:
|
||||
|
|
@ -130,13 +190,19 @@ async def step_solve(options: Chat, request: Request, task_lock: TaskLock):
|
|||
logger.error(f"Error deleting task lock on disconnect: {e}")
|
||||
break
|
||||
try:
|
||||
logger.info(f"[TRACE] Waiting for queue item...")
|
||||
item = await task_lock.get_queue()
|
||||
logger.info(f"[TRACE] Received action: {item.action}, project_id: {options.project_id}")
|
||||
if hasattr(item, 'data'):
|
||||
logger.info(f"[TRACE] Action data preview: {str(item.data)[:200]}")
|
||||
# logger.info(f"item: {dump_class(item)}")
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting item from queue: {e}")
|
||||
break
|
||||
# Continue waiting instead of breaking on queue error
|
||||
continue
|
||||
|
||||
try:
|
||||
logger.info(f"[TRACE] Processing action: {item.action}, start_event_loop={start_event_loop}")
|
||||
if item.action == Action.improve or start_event_loop:
|
||||
# from viztracer import VizTracer
|
||||
|
||||
|
|
@ -144,41 +210,114 @@ async def step_solve(options: Chat, request: Request, task_lock: TaskLock):
|
|||
# tracer.start()
|
||||
if start_event_loop is True:
|
||||
question = options.question
|
||||
logger.info(f"[TRACE] Starting event loop with initial question: {question[:100]}...")
|
||||
start_event_loop = False
|
||||
else:
|
||||
assert isinstance(item, ActionImproveData)
|
||||
question = item.data
|
||||
if len(question) < 12 and len(options.attaches) == 0:
|
||||
confirm = await question_confirm(question_agent, question)
|
||||
logger.info(f"[TRACE] Processing improve action with question: {question[:100]}...")
|
||||
|
||||
logger.info(f"[TRACE] Question length: {len(question)}, Attaches: {len(options.attaches)}")
|
||||
|
||||
# Save user question to history
|
||||
task_lock.add_conversation('user', question)
|
||||
|
||||
# For questions that might reference previous context, always use context-aware confirmation
|
||||
# This allows the agent to determine if it can answer from context
|
||||
if len(options.attaches) == 0 and task_lock.last_task_result:
|
||||
# If there's previous task result, let agent decide based on context
|
||||
logger.info(f"[CONTEXT] Previous task result exists, using context-aware confirmation...")
|
||||
confirm = await question_confirm_with_context(question_agent, question, task_lock)
|
||||
logger.info(f"[CONTEXT] Question confirmation result: {type(confirm)}")
|
||||
elif len(question) < 12 and len(options.attaches) == 0:
|
||||
logger.info(f"[CONTEXT] Short question detected, confirming with context-aware agent...")
|
||||
confirm = await question_confirm_with_context(question_agent, question, task_lock)
|
||||
logger.info(f"[CONTEXT] Question confirmation result: {type(confirm)}")
|
||||
else:
|
||||
confirm = True
|
||||
logger.info(f"[CONTEXT] Long question with no context or has attachments, treating as complex task")
|
||||
|
||||
if confirm is not True:
|
||||
logger.info(f"[CONTEXT] Question not confirmed as complex task, returning simple response")
|
||||
logger.info(f"[TRACE] SSE Response being sent: {confirm}")
|
||||
|
||||
# Extract and save assistant response to history
|
||||
try:
|
||||
import json
|
||||
response_data = json.loads(confirm.split("data: ")[1].strip())
|
||||
response_content = response_data['data']['content']
|
||||
task_lock.add_conversation('assistant', response_content)
|
||||
logger.info(f"[CONTEXT] Simple response saved to history, now has {len(task_lock.conversation_history)} entries")
|
||||
except Exception as e:
|
||||
logger.error(f"[CONTEXT] Failed to save response to history: {e}")
|
||||
|
||||
yield confirm
|
||||
logger.info(f"[TRACE] Simple response sent, continuing main loop to wait for next action...")
|
||||
logger.info(f"[TRACE] Current state after simple response - workforce: {workforce is not None}, camel_task: {camel_task is not None}")
|
||||
logger.info(f"[TRACE] Waiting for next action (should be Action.improve for next question)...")
|
||||
# After sending simple response, continue waiting for next action
|
||||
else:
|
||||
logger.info(f"[CONTEXT] Task confirmed as complex, preparing workforce with context")
|
||||
yield sse_json("confirmed", {"question": question})
|
||||
|
||||
# ========== Prepare context for workforce ==========
|
||||
context_for_task = build_context_for_workforce(task_lock, options)
|
||||
logger.info(f"[CONTEXT] Built context for workforce: {len(context_for_task)} chars")
|
||||
|
||||
(workforce, mcp) = await construct_workforce(options)
|
||||
logger.info(f"[TRACE] Workforce created, initial state: {workforce._state.name if hasattr(workforce, '_state') else 'unknown'}")
|
||||
for new_agent in options.new_agents:
|
||||
workforce.add_single_agent_worker(
|
||||
format_agent_description(new_agent), await new_agent_model(new_agent, options)
|
||||
)
|
||||
summary_task_agent = task_summary_agent(options)
|
||||
task_lock.status = Status.confirmed
|
||||
question = question + options.summary_prompt
|
||||
|
||||
# Add context to task content
|
||||
question_with_context = context_for_task
|
||||
if context_for_task:
|
||||
question_with_context += "\n=== CURRENT TASK ===\n"
|
||||
question_with_context += question + options.summary_prompt
|
||||
|
||||
# Keep the task id consistent
|
||||
camel_task = Task(content=question, id=options.task_id)
|
||||
camel_task = Task(content=question_with_context, id=options.task_id)
|
||||
logger.info(f"[CONTEXT] Created task with context: {options.task_id}")
|
||||
if len(options.attaches) > 0:
|
||||
camel_task.additional_info = {Path(file_path).name: file_path for file_path in options.attaches}
|
||||
|
||||
logger.info(f"[TRACE] Starting task decomposition for task: {options.task_id}")
|
||||
sub_tasks = await asyncio.to_thread(workforce.eigent_make_sub_tasks, camel_task)
|
||||
logger.info(f"[TRACE] Task decomposed into {len(sub_tasks)} subtasks")
|
||||
summary_task_content = await summary_task(summary_task_agent, camel_task)
|
||||
|
||||
# Save task summary for future reference
|
||||
task_lock.last_task_summary = summary_task_content
|
||||
logger.info(f"[CONTEXT] Saved task summary for future context")
|
||||
|
||||
logger.info(f"[TRACE] Sending subtasks to frontend")
|
||||
yield to_sub_tasks(camel_task, summary_task_content)
|
||||
# tracer.stop()
|
||||
# tracer.save("trace.json")
|
||||
|
||||
# Auto-start workforce in specific scenarios
|
||||
should_auto_start = False
|
||||
|
||||
# If this is a follow-up question with context, auto-start
|
||||
if task_lock.last_task_result and len(sub_tasks) <= 2:
|
||||
logger.info(f"[CONTEXT] Auto-starting workforce for context-based follow-up task")
|
||||
should_auto_start = True
|
||||
|
||||
# If debug mode is on, auto-start
|
||||
if env("debug") == "on":
|
||||
logger.info(f"[CONTEXT] Auto-starting workforce in debug mode")
|
||||
should_auto_start = True
|
||||
|
||||
if should_auto_start:
|
||||
task_lock.status = Status.processing
|
||||
task = asyncio.create_task(workforce.eigent_start(sub_tasks))
|
||||
task_lock.add_background_task(task)
|
||||
else:
|
||||
logger.info(f"[CONTEXT] Waiting for manual start command from frontend")
|
||||
|
||||
elif item.action == Action.update_task:
|
||||
assert camel_task is not None
|
||||
|
|
@ -187,6 +326,17 @@ async def step_solve(options: Chat, request: Request, task_lock: TaskLock):
|
|||
add_sub_tasks(camel_task, item.data.task)
|
||||
yield to_sub_tasks(camel_task, summary_task_content)
|
||||
elif item.action == Action.add_task:
|
||||
logger.info(f"[TRACE] === ADD_TASK action received ===")
|
||||
logger.info(f"[TRACE] Task content: {item.content[:100] if hasattr(item, 'content') else 'N/A'}")
|
||||
logger.info(f"[TRACE] Task ID: {item.task_id if hasattr(item, 'task_id') else 'N/A'}")
|
||||
|
||||
# Check if this might be a misrouted second question
|
||||
if camel_task is None and workforce is None:
|
||||
logger.warning(f"[TRACE] ADD_TASK received but no active task/workforce - this might be the second question!")
|
||||
logger.warning(f"[TRACE] The frontend might be sending the second question as ADD_TASK instead of IMPROVE")
|
||||
logger.warning(f"[TRACE] Content being added: {item.content if hasattr(item, 'content') else 'N/A'}")
|
||||
continue
|
||||
|
||||
assert camel_task is not None
|
||||
if workforce is None:
|
||||
logger.error(f"Cannot add task: workforce not initialized for project {options.project_id}")
|
||||
|
|
@ -226,12 +376,19 @@ async def step_solve(options: Chat, request: Request, task_lock: TaskLock):
|
|||
workforce.resume()
|
||||
workforce.skip_gracefully()
|
||||
elif item.action == Action.start:
|
||||
if workforce is not None and workforce._state.name == 'PAUSED':
|
||||
# Resume paused workforce - subtasks should already be loaded
|
||||
logger.info(f"[CHAT] Resuming paused workforce with existing subtasks")
|
||||
workforce.resume()
|
||||
logger.info(f"[TRACE] === START action received ===")
|
||||
if workforce is not None:
|
||||
logger.info(f"[TRACE] Workforce state: {workforce._state.name if hasattr(workforce, '_state') else 'unknown'}")
|
||||
if workforce._state.name == 'PAUSED':
|
||||
# Resume paused workforce - subtasks should already be loaded
|
||||
logger.info(f"[TRACE] Resuming paused workforce with existing subtasks")
|
||||
workforce.resume()
|
||||
continue
|
||||
else:
|
||||
logger.info(f"[TRACE] Workforce is None, cannot start")
|
||||
continue
|
||||
|
||||
|
||||
logger.info(f"[TRACE] Starting workforce with {len(sub_tasks)} subtasks")
|
||||
task_lock.status = Status.processing
|
||||
task = asyncio.create_task(workforce.eigent_start(sub_tasks))
|
||||
task_lock.add_background_task(task)
|
||||
|
|
@ -239,8 +396,11 @@ async def step_solve(options: Chat, request: Request, task_lock: TaskLock):
|
|||
# Track completed task results for the end event
|
||||
if item.data.get('state') == 'DONE' and item.data.get('result'):
|
||||
last_completed_task_result = item.data.get('result', '')
|
||||
logger.info(f"[CONTEXT] Task completed with result: {last_completed_task_result[:100]}...")
|
||||
yield sse_json("task_state", item.data)
|
||||
elif item.action == Action.new_task_state:
|
||||
logger.info(f"[TRACE] === NEW_TASK_STATE action received ===")
|
||||
logger.info(f"[TRACE] Task data: {item.data}")
|
||||
assert camel_task is not None
|
||||
|
||||
# Store the old task information before updating camel_task
|
||||
|
|
@ -251,6 +411,7 @@ async def step_solve(options: Chat, request: Request, task_lock: TaskLock):
|
|||
# Extract task content from the new task data immediately
|
||||
# Don't return question field for new_tasks
|
||||
new_task_content = item.data.get('content', '')
|
||||
logger.info(f"[TRACE] New task content: {new_task_content[:100]}...")
|
||||
|
||||
# Collect context from previous task and prepend to new task
|
||||
if new_task_content:
|
||||
|
|
@ -288,43 +449,79 @@ async def step_solve(options: Chat, request: Request, task_lock: TaskLock):
|
|||
|
||||
# Then handle multi-turn processing
|
||||
if workforce is not None and new_task_content:
|
||||
logger.info(f"[CHAT] MULTI-TURN: Processing new task {item.data.get('task_id')}")
|
||||
logger.info(f"[TRACE] === MULTI-TURN PROCESSING STARTED ===")
|
||||
logger.info(f"[TRACE] Multi-turn task ID: {item.data.get('task_id')}")
|
||||
logger.info(f"[TRACE] Multi-turn content length: {len(new_task_content)}")
|
||||
logger.info(f"[TRACE] Workforce state before pause: {workforce._state if hasattr(workforce, '_state') else 'unknown'}")
|
||||
task_lock.status = Status.confirming
|
||||
workforce.pause()
|
||||
logger.info(f"[TRACE] Workforce paused, state: {workforce._state if hasattr(workforce, '_state') else 'unknown'}")
|
||||
|
||||
try:
|
||||
# Check if this is a simple query
|
||||
if len(new_task_content) < 12:
|
||||
logger.info(f"[TRACE] Short multi-turn question detected, checking with question_agent")
|
||||
multi_turn_confirm = await question_confirm(question_agent, new_task_content)
|
||||
logger.info(f"[TRACE] Multi-turn question confirmation result: {multi_turn_confirm}")
|
||||
|
||||
if multi_turn_confirm is not True:
|
||||
logger.info(f"[TRACE] Multi-turn question identified as simple query, not decomposing")
|
||||
# Still need to send appropriate responses
|
||||
yield sse_json("confirmed", {"question": new_task_content})
|
||||
yield multi_turn_confirm
|
||||
logger.info(f"[TRACE] Resuming workforce after simple query response")
|
||||
workforce.resume()
|
||||
logger.info(f"[TRACE] !!! IMPORTANT: Continuing to next iteration after simple query - this skips further processing !!!")
|
||||
continue # This continues the main while loop, waiting for next action
|
||||
|
||||
logger.info(f"[TRACE] Proceeding with multi-turn task decomposition")
|
||||
yield sse_json("confirmed", {"question": new_task_content})
|
||||
task_lock.status = Status.confirmed
|
||||
|
||||
|
||||
# Use existing workforce to decompose (without creating new one)
|
||||
# Append to _pending_tasks
|
||||
logger.info(f"[TRACE] Calling workforce.handle_decompose_append_task with reset=False")
|
||||
new_sub_tasks = await workforce.handle_decompose_append_task(
|
||||
camel_task, # Use the updated camel_task
|
||||
reset=False # Keep existing agents and context
|
||||
)
|
||||
|
||||
logger.info(f"[TRACE] Decomposition complete, got {len(new_sub_tasks)} subtasks")
|
||||
|
||||
# Generate summary using existing agents
|
||||
summary_task_agent_instance = task_summary_agent(options)
|
||||
new_summary_content = await summary_task(summary_task_agent_instance, camel_task)
|
||||
|
||||
logger.info(f"[TRACE] Summary generated for multi-turn task")
|
||||
|
||||
# Send the extracted events
|
||||
logger.info(f"[TRACE] Sending subtasks to frontend for multi-turn task")
|
||||
yield to_sub_tasks(camel_task, new_summary_content)
|
||||
|
||||
|
||||
# Update the context with new task data
|
||||
sub_tasks = new_sub_tasks
|
||||
summary_task_content = new_summary_content
|
||||
|
||||
logger.info(f"[CHAT] Multi-turn task decomposed into {len(sub_tasks)} subtasks")
|
||||
|
||||
logger.info(f"[TRACE] Multi-turn task decomposed successfully into {len(sub_tasks)} subtasks")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"[CHAT] Error processing multi-turn task: {e}")
|
||||
logger.error(f"[TRACE] Error processing multi-turn task: {e}")
|
||||
import traceback
|
||||
logger.error(f"[TRACE] Traceback: {traceback.format_exc()}")
|
||||
# Continue with existing context if decomposition fails
|
||||
yield sse_json("error", {"message": f"Failed to process task: {str(e)}"})
|
||||
else:
|
||||
logger.warning(f"[TRACE] Multi-turn processing skipped: workforce={workforce is not None}, new_task_content_exists={bool(new_task_content)}")
|
||||
if workforce is None:
|
||||
logger.warning(f"[TRACE] Workforce is None - this might be the issue")
|
||||
if not new_task_content:
|
||||
logger.warning(f"[TRACE] No new task content provided")
|
||||
elif item.action == Action.create_agent:
|
||||
logger.info(f"[TRACE] Processing create_agent action")
|
||||
yield sse_json("create_agent", item.data)
|
||||
elif item.action == Action.activate_agent:
|
||||
logger.info(f"[TRACE] Processing activate_agent action")
|
||||
yield sse_json("activate_agent", item.data)
|
||||
elif item.action == Action.deactivate_agent:
|
||||
logger.info(f"[TRACE] Processing deactivate_agent action")
|
||||
yield sse_json("deactivate_agent", dict(item.data))
|
||||
elif item.action == Action.assign_task:
|
||||
yield sse_json("assign_task", item.data)
|
||||
|
|
@ -355,11 +552,21 @@ async def step_solve(options: Chat, request: Request, task_lock: TaskLock):
|
|||
{"output": item.data, "process_task_id": item.process_task_id},
|
||||
)
|
||||
elif item.action == Action.pause:
|
||||
logger.info(f"[TRACE] === PAUSE action received ===")
|
||||
if workforce is not None:
|
||||
logger.info(f"[TRACE] Pausing workforce, current state: {workforce._state.name if hasattr(workforce, '_state') else 'unknown'}")
|
||||
workforce.pause()
|
||||
logger.info(f"[TRACE] Workforce paused, new state: {workforce._state.name if hasattr(workforce, '_state') else 'unknown'}")
|
||||
else:
|
||||
logger.info(f"[TRACE] Workforce is None, cannot pause")
|
||||
elif item.action == Action.resume:
|
||||
logger.info(f"[TRACE] === RESUME action received ===")
|
||||
if workforce is not None:
|
||||
logger.info(f"[TRACE] Resuming workforce, current state: {workforce._state.name if hasattr(workforce, '_state') else 'unknown'}")
|
||||
workforce.resume()
|
||||
logger.info(f"[TRACE] Workforce resumed, new state: {workforce._state.name if hasattr(workforce, '_state') else 'unknown'}")
|
||||
else:
|
||||
logger.info(f"[TRACE] Workforce is None, cannot resume")
|
||||
elif item.action == Action.new_agent:
|
||||
if workforce is not None:
|
||||
workforce.pause()
|
||||
|
|
@ -368,42 +575,100 @@ async def step_solve(options: Chat, request: Request, task_lock: TaskLock):
|
|||
)
|
||||
workforce.resume()
|
||||
elif item.action == Action.end:
|
||||
logger.info(f"[CONTEXT] === END action received, saving context ===")
|
||||
assert camel_task is not None
|
||||
task_lock.status = Status.done
|
||||
|
||||
# Get the final result from multiple sources in priority order:
|
||||
final_result = ""
|
||||
if last_completed_task_result:
|
||||
final_result = last_completed_task_result
|
||||
else:
|
||||
final_result = str(camel_task.result or "")
|
||||
yield sse_json("end", final_result)
|
||||
if workforce is not None:
|
||||
workforce.stop_gracefully()
|
||||
break
|
||||
elif item.action == Action.supplement:
|
||||
assert camel_task is not None
|
||||
task_lock.status = Status.processing
|
||||
camel_task.add_subtask(
|
||||
Task(
|
||||
content=item.data.question,
|
||||
id=f"{camel_task.id}.{len(camel_task.subtasks)}",
|
||||
)
|
||||
|
||||
# ========== Save task result to task_lock ==========
|
||||
task_lock.last_task_result = final_result
|
||||
|
||||
# Extract actual task content (remove context prefix)
|
||||
task_content = camel_task.content
|
||||
if "=== CURRENT TASK ===" in task_content:
|
||||
task_content = task_content.split("=== CURRENT TASK ===")[-1].strip()
|
||||
|
||||
# Collect full task context
|
||||
full_task_context = collect_previous_task_context(
|
||||
working_directory=options.file_save_path(),
|
||||
previous_task_content=task_content,
|
||||
previous_task_result=final_result,
|
||||
previous_summary=task_lock.last_task_summary
|
||||
)
|
||||
task = asyncio.create_task(workforce.eigent_start(camel_task.subtasks))
|
||||
task_lock.add_background_task(task)
|
||||
|
||||
# Save task result to conversation history (special type)
|
||||
task_lock.add_conversation('task_result', full_task_context)
|
||||
|
||||
logger.info(f"[CONTEXT] Task context saved: {len(full_task_context)} chars")
|
||||
logger.info(f"[CONTEXT] Conversation history now has {len(task_lock.conversation_history)} entries")
|
||||
|
||||
logger.info(f"[TRACE] Sending end signal with result: {final_result[:100]}...")
|
||||
yield sse_json("end", final_result)
|
||||
|
||||
if workforce is not None:
|
||||
logger.info(f"[TRACE] Stopping workforce gracefully")
|
||||
workforce.stop_gracefully()
|
||||
logger.info(f"[TRACE] Workforce stopped")
|
||||
# Reset workforce to None after stopping
|
||||
workforce = None
|
||||
else:
|
||||
logger.warning(f"[TRACE] Workforce is None at end action")
|
||||
|
||||
# Reset camel_task to None for next task
|
||||
camel_task = None
|
||||
|
||||
# Continue the loop to wait for next message instead of breaking
|
||||
logger.info(f"[CONTEXT] Task ended, continuing main loop to wait for next action...")
|
||||
logger.info(f"[CONTEXT] Context preserved for next conversation")
|
||||
# Don't break here - continue waiting for next action
|
||||
elif item.action == Action.supplement:
|
||||
logger.info(f"[TRACE] === SUPPLEMENT action received ===")
|
||||
logger.info(f"[TRACE] Supplement question: {item.data.question[:100] if hasattr(item.data, 'question') else 'N/A'}")
|
||||
|
||||
# Check if this might be a misrouted second question
|
||||
if camel_task is None:
|
||||
logger.warning(f"[TRACE] SUPPLEMENT received but camel_task is None - this might be a misrouted second question!")
|
||||
logger.warning(f"[TRACE] The frontend might be sending the second question as SUPPLEMENT instead of IMPROVE")
|
||||
else:
|
||||
assert camel_task is not None
|
||||
task_lock.status = Status.processing
|
||||
camel_task.add_subtask(
|
||||
Task(
|
||||
content=item.data.question,
|
||||
id=f"{camel_task.id}.{len(camel_task.subtasks)}",
|
||||
)
|
||||
)
|
||||
task = asyncio.create_task(workforce.eigent_start(camel_task.subtasks))
|
||||
task_lock.add_background_task(task)
|
||||
elif item.action == Action.budget_not_enough:
|
||||
if workforce is not None:
|
||||
workforce.pause()
|
||||
yield sse_json(Action.budget_not_enough, {"message": "budget not enouth"})
|
||||
elif item.action == Action.stop:
|
||||
logger.info(f"[TRACE] === STOP action received ===")
|
||||
if workforce is not None:
|
||||
logger.info(f"[TRACE] Workforce exists, state: {workforce._state.name if hasattr(workforce, '_state') else 'unknown'}")
|
||||
if workforce._running:
|
||||
logger.info(f"[TRACE] Workforce is running, stopping it")
|
||||
workforce.stop()
|
||||
logger.info(f"[TRACE] Stopping workforce gracefully")
|
||||
workforce.stop_gracefully()
|
||||
else:
|
||||
logger.warning(f"[TRACE] Workforce is None at stop action")
|
||||
logger.info(f"[TRACE] Deleting task lock")
|
||||
await delete_task_lock(task_lock.id)
|
||||
logger.info(f"[TRACE] Breaking main loop")
|
||||
break
|
||||
else:
|
||||
logger.warning(f"Unknown action: {item.action}")
|
||||
logger.warning(f"[TRACE] Unknown/Unhandled action: {item.action}")
|
||||
logger.warning(f"[TRACE] Current state - workforce: {workforce is not None}, camel_task: {camel_task is not None}")
|
||||
logger.warning(f"[TRACE] Full item data: {dump_class(item) if 'dump_class' in locals() else str(item)}")
|
||||
except ModelProcessingError as e:
|
||||
if "Budget has been exceeded" in str(e):
|
||||
# workforce decompose task don't use ListenAgent, this need return sse
|
||||
|
|
@ -484,6 +749,9 @@ def add_sub_tasks(camel_task: Task, update_tasks: list[TaskContent]):
|
|||
|
||||
|
||||
async def question_confirm(agent: ListenChatAgent, prompt: str) -> str | Literal[True]:
|
||||
logger.info(f"[TRACE] === question_confirm called ===")
|
||||
logger.info(f"[TRACE] Original prompt: {prompt[:100]}...")
|
||||
|
||||
prompt = f"""
|
||||
> **Your Role:** You are a highly capable agent. Your primary function is to analyze a user's request and determine the appropriate course of action.
|
||||
>
|
||||
|
|
@ -500,10 +768,80 @@ async def question_confirm(agent: ListenChatAgent, prompt: str) -> str | Literal
|
|||
> * **For a Complex Task:** Your *only* response should be "yes". This will trigger a specialized workforce to handle the task. Do not include any other text, punctuation, or pleasantries.
|
||||
"""
|
||||
resp = agent.step(prompt)
|
||||
logger.info(f"resp: {agent.chat_history}")
|
||||
if resp.msgs[0].content.lower() != "yes":
|
||||
logger.info(f"[TRACE] Agent response: {resp.msgs[0].content[:200]}...")
|
||||
logger.info(f"[TRACE] Full chat history: {agent.chat_history}")
|
||||
|
||||
is_complex = resp.msgs[0].content.lower() == "yes"
|
||||
logger.info(f"[TRACE] Is complex task? {is_complex}")
|
||||
|
||||
if not is_complex:
|
||||
logger.info(f"[TRACE] Returning simple query response")
|
||||
return sse_json("wait_confirm", {"content": resp.msgs[0].content})
|
||||
else:
|
||||
logger.info(f"[TRACE] Confirmed as complex task")
|
||||
return True
|
||||
|
||||
|
||||
async def question_confirm_with_context(agent: ListenChatAgent, prompt: str, task_lock: TaskLock) -> str | Literal[True]:
|
||||
"""Question confirmation with conversation context"""
|
||||
logger.info(f"[CONTEXT] === question_confirm with context ===")
|
||||
logger.info(f"[CONTEXT] History length: {len(task_lock.conversation_history)}")
|
||||
logger.info(f"[CONTEXT] Has previous task result: {bool(task_lock.last_task_result)}")
|
||||
|
||||
# Build context prompt
|
||||
context_prompt = ""
|
||||
|
||||
# Add conversation history (last 10 entries)
|
||||
if task_lock.conversation_history:
|
||||
context_prompt = "=== Previous Conversation ===\n"
|
||||
recent_history = task_lock.conversation_history[-10:]
|
||||
|
||||
for entry in recent_history:
|
||||
role = entry['role']
|
||||
content = entry['content']
|
||||
|
||||
if role == 'task_result':
|
||||
# Special handling for task results - show summary only
|
||||
context_prompt += f"[Task Completed]: {content[:200]}...\n"
|
||||
else:
|
||||
# Limit content length to avoid too long context
|
||||
context_prompt += f"{role.capitalize()}: {content[:200]}\n"
|
||||
|
||||
context_prompt += "\n"
|
||||
|
||||
# Add last task result if available
|
||||
if task_lock.last_task_result:
|
||||
context_prompt += f"=== Last Task Result ===\n{task_lock.last_task_result[:500]}...\n\n"
|
||||
|
||||
# Combine full prompt
|
||||
full_prompt = f"""{context_prompt}
|
||||
=== Current Query ===
|
||||
User: {prompt}
|
||||
|
||||
> **Your Role:** You are a highly capable agent with memory of previous conversations.
|
||||
>
|
||||
> **Instructions:**
|
||||
> 1. Consider the conversation history and any previous task results
|
||||
> 2. Analyze if the current query is:
|
||||
> - A simple question that can be answered directly (including referencing previous results)
|
||||
> - A complex task requiring multiple steps or tool usage
|
||||
> 3. Decision:
|
||||
> - For simple queries: Provide a helpful response (you may reference previous context)
|
||||
> - For complex tasks: Respond with only "yes"
|
||||
>
|
||||
> **Important:** If the user asks about something from a previous task, and you can answer based on the context provided, treat it as a simple query.
|
||||
"""
|
||||
|
||||
# Execute agent
|
||||
resp = agent.step(full_prompt)
|
||||
|
||||
is_complex = resp.msgs[0].content.lower() == "yes"
|
||||
|
||||
if not is_complex:
|
||||
logger.info(f"[CONTEXT] Simple query, providing direct response")
|
||||
return sse_json("wait_confirm", {"content": resp.msgs[0].content})
|
||||
else:
|
||||
logger.info(f"[CONTEXT] Complex task confirmed")
|
||||
return True
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -1,4 +1,5 @@
|
|||
from typing_extensions import Any, Literal, TypedDict
|
||||
from typing import List, Dict, Optional
|
||||
from pydantic import BaseModel
|
||||
from app.exception.exception import ProgramException
|
||||
from app.model.chat import McpServers, Status, SupplementChat, Chat, UpdateData
|
||||
|
|
@ -252,6 +253,16 @@ class TaskLock:
|
|||
background_tasks: set[asyncio.Task]
|
||||
"""Track all background tasks for cleanup"""
|
||||
|
||||
# Context management fields
|
||||
conversation_history: List[Dict[str, str]]
|
||||
"""Store conversation history for context"""
|
||||
last_task_result: str
|
||||
"""Store the last task execution result"""
|
||||
last_task_summary: str
|
||||
"""Store the last task summary"""
|
||||
question_agent: Optional[Any]
|
||||
"""Persistent question confirmation agent"""
|
||||
|
||||
def __init__(self, id: str, queue: asyncio.Queue, human_input: dict) -> None:
|
||||
self.id = id
|
||||
self.queue = queue
|
||||
|
|
@ -260,6 +271,12 @@ class TaskLock:
|
|||
self.last_accessed = datetime.now()
|
||||
self.background_tasks = set()
|
||||
|
||||
# Initialize context management fields
|
||||
self.conversation_history = []
|
||||
self.last_task_result = ""
|
||||
self.last_task_summary = ""
|
||||
self.question_agent = None
|
||||
|
||||
async def put_queue(self, data: ActionData):
|
||||
self.last_accessed = datetime.now()
|
||||
await self.queue.put(data)
|
||||
|
|
@ -293,6 +310,28 @@ class TaskLock:
|
|||
pass
|
||||
self.background_tasks.clear()
|
||||
|
||||
def add_conversation(self, role: str, content: str):
|
||||
"""Add a conversation entry to history"""
|
||||
self.conversation_history.append({
|
||||
'role': role,
|
||||
'content': content,
|
||||
'timestamp': datetime.now().isoformat()
|
||||
})
|
||||
|
||||
# Limit history length to prevent memory issues
|
||||
if len(self.conversation_history) > 20:
|
||||
self.conversation_history = self.conversation_history[-20:]
|
||||
|
||||
def get_recent_context(self, max_entries: int = 10) -> str:
|
||||
"""Get recent conversation context as a formatted string"""
|
||||
if not self.conversation_history:
|
||||
return ""
|
||||
|
||||
context = "=== Recent Conversation ===\n"
|
||||
for entry in self.conversation_history[-max_entries:]:
|
||||
context += f"{entry['role']}: {entry['content'][:200]}\n"
|
||||
return context
|
||||
|
||||
|
||||
task_locks = dict[str, TaskLock]()
|
||||
# Cleanup task for removing stale task locks
|
||||
|
|
|
|||
|
|
@ -200,7 +200,7 @@ const checkManagerInstance = (manager: any, name: string) => {
|
|||
function registerIpcHandlers() {
|
||||
// ==================== basic info handler ====================
|
||||
ipcMain.handle('get-browser-port', () => {
|
||||
log.info('Starting new task')
|
||||
log.info('Getting browser port')
|
||||
return browser_port
|
||||
});
|
||||
ipcMain.handle('get-app-version', () => app.getVersion());
|
||||
|
|
|
|||
|
|
@ -14,7 +14,7 @@ interface ProjectSectionProps {
|
|||
isPauseResumeLoading: boolean;
|
||||
}
|
||||
|
||||
export const ProjectSection: React.FC<ProjectSectionProps> = ({
|
||||
export const ProjectSection = React.forwardRef<HTMLDivElement, ProjectSectionProps>(({
|
||||
chatId,
|
||||
chatStore,
|
||||
activeQueryId,
|
||||
|
|
@ -22,22 +22,23 @@ export const ProjectSection: React.FC<ProjectSectionProps> = ({
|
|||
onPauseResume,
|
||||
onSkip,
|
||||
isPauseResumeLoading
|
||||
}) => {
|
||||
}, ref) => {
|
||||
const chatState = chatStore.getState();
|
||||
const activeTaskId = chatState.activeTaskId;
|
||||
|
||||
|
||||
if (!activeTaskId || !chatState.tasks[activeTaskId]) {
|
||||
return null;
|
||||
}
|
||||
|
||||
const task = chatState.tasks[activeTaskId];
|
||||
const messages = task.messages || [];
|
||||
|
||||
|
||||
// Group messages by query cycles and show in chronological order (oldest first)
|
||||
const queryGroups = groupMessagesByQuery(messages);
|
||||
|
||||
return (
|
||||
<motion.div
|
||||
ref={ref}
|
||||
initial={{ opacity: 0, y: 20 }}
|
||||
animate={{ opacity: 1, y: 0 }}
|
||||
exit={{ opacity: 0, y: -20 }}
|
||||
|
|
@ -71,7 +72,10 @@ export const ProjectSection: React.FC<ProjectSectionProps> = ({
|
|||
)}
|
||||
</motion.div>
|
||||
);
|
||||
};
|
||||
});
|
||||
|
||||
// Add display name for better debugging
|
||||
ProjectSection.displayName = 'ProjectSection';
|
||||
|
||||
// Helper function to group messages by query cycles
|
||||
function groupMessagesByQuery(messages: any[]) {
|
||||
|
|
|
|||
|
|
@ -35,7 +35,10 @@ export const UserQueryGroup: React.FC<UserQueryGroupProps> = ({
|
|||
const [isTaskBoxSticky, setIsTaskBoxSticky] = useState(false);
|
||||
const chatState = chatStore.getState();
|
||||
const activeTaskId = chatState.activeTaskId;
|
||||
const task = activeTaskId ? chatState.tasks[activeTaskId] : null;
|
||||
|
||||
// Only show task if this query group has a task message
|
||||
// This prevents all query groups from showing the same task
|
||||
const task = queryGroup.taskMessage && activeTaskId ? chatState.tasks[activeTaskId] : null;
|
||||
|
||||
// Set up intersection observer for this query group
|
||||
useEffect(() => {
|
||||
|
|
|
|||
|
|
@ -192,20 +192,49 @@ export default function ChatBox(): JSX.Element {
|
|||
return;
|
||||
}
|
||||
|
||||
if (chatStore.tasks[_taskId as string]?.hasWaitComfirm) {
|
||||
// If the task has not started yet (pending status), start it normally
|
||||
if (chatStore.tasks[_taskId as string].status === "pending") {
|
||||
// Check if we should continue the conversation or start a new task
|
||||
const hasMessages = chatStore.tasks[_taskId as string].messages.length > 0;
|
||||
const isFinished = chatStore.tasks[_taskId as string].status === "finished";
|
||||
const hasWaitComfirm = chatStore.tasks[_taskId as string]?.hasWaitComfirm;
|
||||
|
||||
// Continue conversation if:
|
||||
// 1. Has wait confirm (simple query response)
|
||||
// 2. Task is finished (complex task completed)
|
||||
// 3. Has any messages but pending (ongoing conversation)
|
||||
const shouldContinueConversation = hasWaitComfirm || isFinished || (hasMessages && chatStore.tasks[_taskId as string].status === "pending");
|
||||
|
||||
if (shouldContinueConversation) {
|
||||
// Check if this is the very first message and task hasn't started
|
||||
const hasSimpleResponse = chatStore.tasks[_taskId as string].messages.some(
|
||||
m => m.step === "wait_confirm"
|
||||
);
|
||||
const hasComplexTask = chatStore.tasks[_taskId as string].messages.some(
|
||||
m => m.step === "to_sub_tasks"
|
||||
);
|
||||
|
||||
// Only start a new task if: pending, no messages processed yet
|
||||
if (chatStore.tasks[_taskId as string].status === "pending" && !hasSimpleResponse && !hasComplexTask && !isFinished) {
|
||||
setMessage("");
|
||||
// Pass the message content to startTask instead of adding it to current chatStore
|
||||
const attachesToSend = JSON.parse(JSON.stringify(chatStore.tasks[_taskId]?.attaches)) || [];
|
||||
chatStore.startTask(_taskId, undefined, undefined, undefined, tempMessageContent, attachesToSend);
|
||||
// keep hasWaitComfirm as true so that follow-up improves work as usual
|
||||
} else {
|
||||
// Task already started and is waiting for user confirmation – use improve API
|
||||
// Continue conversation: simple response, complex task, or finished task
|
||||
console.log("[Multi-turn] Continuing conversation with improve API");
|
||||
fetchPost(`/chat/${projectStore.activeProjectId}`, {
|
||||
question: tempMessageContent,
|
||||
});
|
||||
chatStore.setIsPending(_taskId, true);
|
||||
// Add the user message to show it in UI
|
||||
chatStore.addMessages(_taskId, {
|
||||
id: generateUniqueId(),
|
||||
role: "user",
|
||||
content: tempMessageContent,
|
||||
attaches: JSON.parse(JSON.stringify(chatStore.tasks[_taskId]?.attaches)) || [],
|
||||
});
|
||||
chatStore.setAttaches(_taskId, []);
|
||||
setMessage("");
|
||||
}
|
||||
} else {
|
||||
if (!privacy) {
|
||||
|
|
|
|||
|
|
@ -28,9 +28,20 @@ export default function Home() {
|
|||
|
||||
const [activeWebviewId, setActiveWebviewId] = useState<string | null>(null);
|
||||
|
||||
window.ipcRenderer?.on("webview-show", (_event, id: string) => {
|
||||
setActiveWebviewId(id);
|
||||
});
|
||||
// Add webview-show listener in useEffect with cleanup
|
||||
useEffect(() => {
|
||||
const handleWebviewShow = (_event: any, id: string) => {
|
||||
setActiveWebviewId(id);
|
||||
};
|
||||
|
||||
window.ipcRenderer?.on("webview-show", handleWebviewShow);
|
||||
|
||||
// Cleanup: remove listener on unmount
|
||||
return () => {
|
||||
window.ipcRenderer?.off("webview-show", handleWebviewShow);
|
||||
};
|
||||
}, []); // Empty dependency array means this only runs once
|
||||
|
||||
useEffect(() => {
|
||||
let taskAssigning = [
|
||||
...(chatStore.tasks[chatStore.activeTaskId as string]?.taskAssigning ||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue