From 128ba7fbc8728cca7c76048841bd50e1b39458db Mon Sep 17 00:00:00 2001 From: liangxinbing <1580466765@qq.com> Date: Sat, 29 Mar 2025 15:52:10 +0800 Subject: [PATCH] temporarily remove CoTAgent and PlanningAgent --- app/agent/__init__.py | 4 - app/agent/cot.py | 47 -------- app/agent/planning.py | 259 ------------------------------------------ app/agent/swe.py | 3 - 4 files changed, 313 deletions(-) delete mode 100644 app/agent/cot.py delete mode 100644 app/agent/planning.py diff --git a/app/agent/__init__.py b/app/agent/__init__.py index 23f26f4..f7df2b9 100644 --- a/app/agent/__init__.py +++ b/app/agent/__init__.py @@ -1,8 +1,6 @@ from app.agent.base import BaseAgent from app.agent.browser import BrowserAgent -from app.agent.cot import CoTAgent from app.agent.mcp import MCPAgent -from app.agent.planning import PlanningAgent from app.agent.react import ReActAgent from app.agent.swe import SWEAgent from app.agent.toolcall import ToolCallAgent @@ -11,8 +9,6 @@ from app.agent.toolcall import ToolCallAgent __all__ = [ "BaseAgent", "BrowserAgent", - "CoTAgent", - "PlanningAgent", "ReActAgent", "SWEAgent", "ToolCallAgent", diff --git a/app/agent/cot.py b/app/agent/cot.py deleted file mode 100644 index a7bd57e..0000000 --- a/app/agent/cot.py +++ /dev/null @@ -1,47 +0,0 @@ -from typing import Optional - -from pydantic import Field - -from app.agent.base import BaseAgent -from app.llm import LLM -from app.logger import logger -from app.prompt.cot import NEXT_STEP_PROMPT, SYSTEM_PROMPT -from app.schema import AgentState, Message - - -class CoTAgent(BaseAgent): - """Chain of Thought Agent - Focuses on demonstrating the thinking process of large language models without executing tools""" - - name: str = "cot" - description: str = "An agent that uses Chain of Thought reasoning" - - system_prompt: str = SYSTEM_PROMPT - next_step_prompt: Optional[str] = NEXT_STEP_PROMPT - - llm: LLM = Field(default_factory=LLM) - - max_steps: int = 1 # CoT typically only needs one step to complete reasoning - - async def step(self) -> str: - """Execute one step of chain of thought reasoning""" - logger.info(f"🧠 {self.name} is thinking...") - - # If next_step_prompt exists and this isn't the first message, add it to user messages - if self.next_step_prompt and len(self.messages) > 1: - self.memory.add_message(Message.user_message(self.next_step_prompt)) - - # Use system prompt and user messages - response = await self.llm.ask( - messages=self.messages, - system_msgs=[Message.system_message(self.system_prompt)] - if self.system_prompt - else None, - ) - - # Record assistant's response - self.memory.add_message(Message.assistant_message(response)) - - # Set state to finished after completion - self.state = AgentState.FINISHED - - return response diff --git a/app/agent/planning.py b/app/agent/planning.py deleted file mode 100644 index 7e98912..0000000 --- a/app/agent/planning.py +++ /dev/null @@ -1,259 +0,0 @@ -import time -from typing import Dict, List, Optional - -from pydantic import Field, model_validator - -from app.agent.toolcall import ToolCallAgent -from app.logger import logger -from app.prompt.planning import NEXT_STEP_PROMPT, PLANNING_SYSTEM_PROMPT -from app.schema import TOOL_CHOICE_TYPE, Message, ToolCall, ToolChoice -from app.tool import PlanningTool, Terminate, ToolCollection - - -class PlanningAgent(ToolCallAgent): - """ - An agent that creates and manages plans to solve tasks. - - This agent uses a planning tool to create and manage structured plans, - and tracks progress through individual steps until task completion. - """ - - name: str = "planning" - description: str = "An agent that creates and manages plans to solve tasks" - - system_prompt: str = PLANNING_SYSTEM_PROMPT - next_step_prompt: str = NEXT_STEP_PROMPT - - available_tools: ToolCollection = Field( - default_factory=lambda: ToolCollection(PlanningTool(), Terminate()) - ) - tool_choices: TOOL_CHOICE_TYPE = ToolChoice.AUTO # type: ignore - special_tool_names: List[str] = Field(default_factory=lambda: [Terminate().name]) - - tool_calls: List[ToolCall] = Field(default_factory=list) - active_plan_id: Optional[str] = Field(default=None) - - # Add a dictionary to track the step status for each tool call - step_execution_tracker: Dict[str, Dict] = Field(default_factory=dict) - current_step_index: Optional[int] = None - - max_steps: int = 20 - - @model_validator(mode="after") - def initialize_plan_and_verify_tools(self) -> "PlanningAgent": - """Initialize the agent with a default plan ID and validate required tools.""" - self.active_plan_id = f"plan_{int(time.time())}" - - if "planning" not in self.available_tools.tool_map: - self.available_tools.add_tool(PlanningTool()) - - return self - - async def think(self) -> bool: - """Decide the next action based on plan status.""" - prompt = ( - f"CURRENT PLAN STATUS:\n{await self.get_plan()}\n\n{self.next_step_prompt}" - if self.active_plan_id - else self.next_step_prompt - ) - self.messages.append(Message.user_message(prompt)) - - # Get the current step index before thinking - self.current_step_index = await self._get_current_step_index() - - result = await super().think() - - # After thinking, if we decided to execute a tool and it's not a planning tool or special tool, - # associate it with the current step for tracking - if result and self.tool_calls: - latest_tool_call = self.tool_calls[0] # Get the most recent tool call - if ( - latest_tool_call.function.name != "planning" - and latest_tool_call.function.name not in self.special_tool_names - and self.current_step_index is not None - ): - self.step_execution_tracker[latest_tool_call.id] = { - "step_index": self.current_step_index, - "tool_name": latest_tool_call.function.name, - "status": "pending", # Will be updated after execution - } - - return result - - async def act(self) -> str: - """Execute a step and track its completion status.""" - result = await super().act() - - # After executing the tool, update the plan status - if self.tool_calls: - latest_tool_call = self.tool_calls[0] - - # Update the execution status to completed - if latest_tool_call.id in self.step_execution_tracker: - self.step_execution_tracker[latest_tool_call.id]["status"] = "completed" - self.step_execution_tracker[latest_tool_call.id]["result"] = result - - # Update the plan status if this was a non-planning, non-special tool - if ( - latest_tool_call.function.name != "planning" - and latest_tool_call.function.name not in self.special_tool_names - ): - await self.update_plan_status(latest_tool_call.id) - - return result - - async def get_plan(self) -> str: - """Retrieve the current plan status.""" - if not self.active_plan_id: - return "No active plan. Please create a plan first." - - result = await self.available_tools.execute( - name="planning", - tool_input={"command": "get", "plan_id": self.active_plan_id}, - ) - return result.output if hasattr(result, "output") else str(result) - - async def run(self, request: Optional[str] = None) -> str: - """Run the agent with an optional initial request.""" - if request: - await self.create_initial_plan(request) - return await super().run() - - async def update_plan_status(self, tool_call_id: str) -> None: - """ - Update the current plan progress based on completed tool execution. - Only marks a step as completed if the associated tool has been successfully executed. - """ - if not self.active_plan_id: - return - - if tool_call_id not in self.step_execution_tracker: - logger.warning(f"No step tracking found for tool call {tool_call_id}") - return - - tracker = self.step_execution_tracker[tool_call_id] - if tracker["status"] != "completed": - logger.warning(f"Tool call {tool_call_id} has not completed successfully") - return - - step_index = tracker["step_index"] - - try: - # Mark the step as completed - await self.available_tools.execute( - name="planning", - tool_input={ - "command": "mark_step", - "plan_id": self.active_plan_id, - "step_index": step_index, - "step_status": "completed", - }, - ) - logger.info( - f"Marked step {step_index} as completed in plan {self.active_plan_id}" - ) - except Exception as e: - logger.warning(f"Failed to update plan status: {e}") - - async def _get_current_step_index(self) -> Optional[int]: - """ - Parse the current plan to identify the first non-completed step's index. - Returns None if no active step is found. - """ - if not self.active_plan_id: - return None - - plan = await self.get_plan() - - try: - plan_lines = plan.splitlines() - steps_index = -1 - - # Find the index of the "Steps:" line - for i, line in enumerate(plan_lines): - if line.strip() == "Steps:": - steps_index = i - break - - if steps_index == -1: - return None - - # Find the first non-completed step - for i, line in enumerate(plan_lines[steps_index + 1 :], start=0): - if "[ ]" in line or "[→]" in line: # not_started or in_progress - # Mark current step as in_progress - await self.available_tools.execute( - name="planning", - tool_input={ - "command": "mark_step", - "plan_id": self.active_plan_id, - "step_index": i, - "step_status": "in_progress", - }, - ) - return i - - return None # No active step found - except Exception as e: - logger.warning(f"Error finding current step index: {e}") - return None - - async def create_initial_plan(self, request: str) -> None: - """Create an initial plan based on the request.""" - logger.info(f"Creating initial plan with ID: {self.active_plan_id}") - - messages = [ - Message.user_message( - f"Analyze the request and create a plan with ID {self.active_plan_id}: {request}" - ) - ] - self.memory.add_messages(messages) - response = await self.llm.ask_tool( - messages=messages, - system_msgs=[Message.system_message(self.system_prompt)], - tools=self.available_tools.to_params(), - tool_choice=ToolChoice.AUTO, - ) - assistant_msg = Message.from_tool_calls( - content=response.content, tool_calls=response.tool_calls - ) - - self.memory.add_message(assistant_msg) - - plan_created = False - for tool_call in response.tool_calls: - if tool_call.function.name == "planning": - result = await self.execute_tool(tool_call) - logger.info( - f"Executed tool {tool_call.function.name} with result: {result}" - ) - - # Add tool response to memory - tool_msg = Message.tool_message( - content=result, - tool_call_id=tool_call.id, - name=tool_call.function.name, - ) - self.memory.add_message(tool_msg) - plan_created = True - break - - if not plan_created: - logger.warning("No plan created from initial request") - tool_msg = Message.assistant_message( - "Error: Parameter `plan_id` is required for command: create" - ) - self.memory.add_message(tool_msg) - - -async def main(): - # Configure and run the agent - agent = PlanningAgent(available_tools=ToolCollection(PlanningTool(), Terminate())) - result = await agent.run("Help me plan a trip to the moon") - print(result) - - -if __name__ == "__main__": - import asyncio - - asyncio.run(main()) diff --git a/app/agent/swe.py b/app/agent/swe.py index 9044204..e655a5b 100644 --- a/app/agent/swe.py +++ b/app/agent/swe.py @@ -22,6 +22,3 @@ class SWEAgent(ToolCallAgent): special_tool_names: List[str] = Field(default_factory=lambda: [Terminate().name]) max_steps: int = 20 - - bash: Bash = Field(default_factory=Bash) - working_dir: str = "."