Upload 5 files

examples/plan_execute/agent.py

@@ -0,0 +1,246 @@
+# coding: utf-8
+# Copyright (c) 2025 inclusionAI.
+import copy
+import json
+import traceback
+from typing import Dict, Any, List, Union
+
+from examples.tools.common import Agents
+from aworld.core.agent.base import AgentResult
+from aworld.agents.llm_agent import Agent
+from aworld.models.llm import call_llm_model
+from aworld.config.conf import AgentConfig, ConfigDict
+from aworld.core.common import Observation, ActionModel
+from aworld.logs.util import logger
+from examples.plan_execute.prompts import *
+from examples.plan_execute.utils import extract_pattern
+
+
+class ExecuteAgent(Agent):
+    def __init__(self, conf: Union[Dict[str, Any], ConfigDict, AgentConfig], **kwargs):
+        super(ExecuteAgent, self).__init__(conf, **kwargs)
+
+    def id(self) -> str:
+        return Agents.EXECUTE.value
+
+    def reset(self, options: Dict[str, Any]):
+        """Execute agent reset need query task as input."""
+        super().reset(options)
+
+        self.system_prompt = execute_system_prompt.format(task=self.task)
+        self.step_reset = False
+
+    async def async_policy(self, observation: Observation, info: Dict[str, Any] = {}, **kwargs) -> Union[
+        List[ActionModel], None]:
+        await self.async_desc_transform()
+        return self._common(observation, info)
+
+    def policy(self,
+               observation: Observation,
+               info: Dict[str, Any] = None,
+               **kwargs) -> List[ActionModel] | None:
+        self.desc_transform()
+        return self._common(observation, info)
+
+    def _common(self, observation, info):
+        self._finished = False
+        content = observation.content
+
+        llm_result = None
+        ## build input of llm
+        input_content = [
+            {'role': 'system', 'content': self.system_prompt},
+        ]
+        for traj in self.trajectory:
+            # Handle multiple messages in content
+            if isinstance(traj[0].content, list):
+                input_content.extend(traj[0].content)
+            else:
+                input_content.append(traj[0].content)
+
+            if traj[-1].tool_calls is not None:
+                input_content.append(
+                    {'role': 'assistant', 'content': '', 'tool_calls': traj[-1].tool_calls})
+            else:
+                input_content.append({'role': 'assistant', 'content': traj[-1].content})
+
+        if content is None:
+            content = observation.action_result[0].error
+        if not self.trajectory:
+            new_messages = [{"role": "user", "content": content}]
+            input_content.extend(new_messages)
+        else:
+            # Collect existing tool_call_ids from input_content
+            existing_tool_call_ids = {
+                msg.get("tool_call_id") for msg in input_content
+                if msg.get("role") == "tool" and msg.get("tool_call_id")
+            }
+
+            new_messages = []
+            for traj in self.trajectory:
+                if traj[-1].tool_calls is not None:
+                    # Handle multiple tool calls
+                    for tool_call in traj[-1].tool_calls:
+                        # Only add if this tool_call_id doesn't exist in input_content
+                        if tool_call.id not in existing_tool_call_ids:
+                            new_messages.append({
+                                "role": "tool",
+                                "content": content,
+                                "tool_call_id": tool_call.id
+                            })
+            if new_messages:
+                input_content.extend(new_messages)
+            else:
+                input_content.append({"role": "user", "content": content})
+
+            # Validate tool_calls and tool messages pairing
+            assistant_tool_calls = []
+            tool_responses = []
+            for msg in input_content:
+                if msg.get("role") == "assistant" and msg.get("tool_calls"):
+                    assistant_tool_calls.extend(msg["tool_calls"])
+                elif msg.get("role") == "tool":
+                    tool_responses.append(msg.get("tool_call_id"))
+
+            # Check if all tool_calls have corresponding responses
+            tool_call_ids = {call.id for call in assistant_tool_calls}
+            tool_response_ids = set(tool_responses)
+            if tool_call_ids != tool_response_ids:
+                missing_calls = tool_call_ids - tool_response_ids
+                extra_responses = tool_response_ids - tool_call_ids
+                error_msg = f"Tool calls and responses mismatch. Missing responses for tool_calls: {missing_calls}, Extra responses: {extra_responses}"
+                logger.error(error_msg)
+                raise ValueError(error_msg)
+
+        tool_calls = []
+        try:
+            llm_result = call_llm_model(self.llm, input_content, model=self.model_name,
+                                        tools=self.tools, temperature=0)
+            logger.info(f"Execute response: {llm_result.message}")
+            res = self.response_parse(llm_result)
+            content = res.actions[0].policy_info
+            tool_calls = llm_result.tool_calls
+        except Exception as e:
+            logger.warning(traceback.format_exc())
+        finally:
+            if llm_result:
+                ob = copy.deepcopy(observation)
+                ob.content = new_messages
+                self.trajectory.append((ob, info, llm_result))
+            else:
+                logger.warning("no result to record!")
+
+        res = []
+        if tool_calls:
+            for tool_call in tool_calls:
+                tool_action_name: str = tool_call.function.name
+                if not tool_action_name:
+                    continue
+
+                names = tool_action_name.split("__")
+                tool_name = names[0]
+                action_name = '__'.join(names[1:]) if len(names) > 1 else ''
+                params = json.loads(tool_call.function.arguments)
+                res.append(ActionModel(agent_name=Agents.EXECUTE.value,
+                                       tool_name=tool_name,
+                                       action_name=action_name,
+                                       params=params))
+
+        if res:
+            res[0].policy_info = content
+            self._finished = False
+        elif content:
+            policy_info = extract_pattern(content, "final_answer")
+            if policy_info:
+                res.append(ActionModel(agent_name=Agents.EXECUTE.value,
+                                       policy_info=policy_info))
+                self._finished = True
+            else:
+                res.append(ActionModel(agent_name=Agents.EXECUTE.value,
+                                       policy_info=content))
+
+        logger.info(f">>> execute result: {res}")
+
+        result = AgentResult(actions=res,
+                             current_state=None)
+        return result.actions
+
+
+class PlanAgent(Agent):
+    def __init__(self, conf: Union[Dict[str, Any], ConfigDict, AgentConfig], **kwargs):
+        super(PlanAgent, self).__init__(conf, **kwargs)
+
+    def id(self) -> str:
+        return Agents.PLAN.value
+
+    def reset(self, options: Dict[str, Any]):
+        """Execute agent reset need query task as input."""
+        super().reset(options)
+
+        self.system_prompt = plan_system_prompt.format(task=self.task)
+        self.done_prompt = plan_done_prompt.format(task=self.task)
+        self.postfix_prompt = plan_postfix_prompt.format(task=self.task)
+        self.first_prompt = init_prompt
+        self.first = True
+        self.step_reset = False
+
+    async def async_policy(self, observation: Observation, info: Dict[str, Any] = {}, **kwargs) -> Union[
+        List[ActionModel], None]:
+        await self.async_desc_transform()
+        return self._common(observation, info)
+
+    def policy(self,
+               observation: Observation,
+               info: Dict[str, Any] = None,
+               **kwargs) -> List[ActionModel] | None:
+        self._finished = False
+        self.desc_transform()
+        return self._common(observation, info)
+
+    def _common(self, observation, info):
+        llm_result = None
+        input_content = [
+            {'role': 'system', 'content': self.system_prompt},
+        ]
+        # build input of llm based history
+        for traj in self.trajectory:
+            input_content.append({'role': 'user', 'content': traj[0].content})
+            # plan agent no tool to call, use content
+            input_content.append({'role': 'assistant', 'content': traj[-1].content})
+
+        message = observation.content
+        if self.first_prompt:
+            message = self.first_prompt
+            self.first_prompt = None
+
+        input_content.append({"role": "user", "content": message})
+        try:
+            llm_result = call_llm_model(self.llm, messages=input_content, model=self.model_name)
+            logger.info(f"Plan response: {llm_result.message}")
+        except Exception as e:
+            logger.warning(traceback.format_exc())
+            raise e
+        finally:
+            if llm_result:
+                ob = copy.deepcopy(observation)
+                ob.content = message
+                self.trajectory.append((ob, info, llm_result))
+            else:
+                logger.warning("no result to record!")
+        res = self.response_parse(llm_result)
+        content = res.actions[0].policy_info
+        if "TASK_DONE" not in content:
+            content += self.done_prompt
+        else:
+            # The task is done, and the assistant agent need to give the final answer about the original task
+            content += self.postfix_prompt
+            if not self.first:
+                self._finished = True
+
+        self.first = False
+        logger.info(f">>> plan result: {content}")
+        result = AgentResult(actions=[ActionModel(agent_name=Agents.PLAN.value,
+                                                  tool_name=Agents.EXECUTE.value,
+                                                  policy_info=content)],
+                             current_state=None)
+        return result.actions

examples/plan_execute/mock.py

@@ -0,0 +1,27 @@
+# coding: utf-8
+# Copyright (c) 2025 inclusionAI.
+import os
+
+from aworld.utils.import_package import import_packages
+
+import_packages(["pandas", "numpy"])
+
+import pandas as pd
+import numpy as np
+
+from aworld.utils import import_package
+
+
+def mock_dataset(name: str):
+    if name == 'gaia':
+        npy_path = f"{os.getcwd()}/gaia.npy"
+
+        numpy_array = np.load(npy_path, allow_pickle=True)
+        df = pd.DataFrame(numpy_array[:-1])
+        query = numpy_array[-1][0]
+
+        save_file_path = f"{os.getcwd()}/gaia.xlsx"
+        import_package("openpyxl")
+        df.to_excel(save_file_path, index=False, header=None)
+        return query.format(file_path=save_file_path)
+    return None

examples/plan_execute/prompts.py

@@ -0,0 +1,88 @@
+init_prompt = f"""
+Please give me clear step-by-step instructions to complete the entire task. If the task needs any special knowledge, let me know which tools I should use to help me get it done.
+"""
+
+execute_system_prompt = """
+===== RULES FOR THE ASSISTANT =====
+You are my assistant, and I am your user. Always remember this! Do not flip roles. You are here to help me. Do not give me instructions.
+Use the tools available to you to solve the tasks I give you.
+Our goal is to work together to successfully solve complex tasks.
+
+The Task:
+Our overall task is: {task}. Never forget this.
+
+Instructions:
+I will give you instructions to help solve the task. These instructions will usually be smaller sub-tasks or questions.
+You must use your tools, do your best to solve the problem, and clearly explain your solutions.
+
+How You Should Answer:
+Always begin your response with: Solution: [YOUR_SOLUTION]
+[YOUR_SOLUTION] should be clear, detailed, and specific. Provide examples, lists, or detailed implementations if needed.
+
+Additional Notes:
+Our overall task may be complicated. Here are tips to help you:
+<tips>
+- If one method fails, try another. There is always a solution.
+- If a search snippet is not helpful, but the link is from a reliable source, visit the link for more details.
+- For specific values like numbers, prioritize credible sources.
+- Start with Wikipedia when researching, then explore other websites if needed.
+- Solve math problems using Python and libraries like sympy. Test your code for results and debug when necessary.
+- Validate your answers by cross-checking them through different methods.
+- If a tool or code fails, do not assume its result is correct. Investigate the problem, fix it, and try again.
+- Search results rarely provide exact answers. Use simple search queries to find sources, then process them further (e.g., by extracting webpage data).
+- For downloading files, either use a browser simulation tool or write code to download them.
+</tips>
+
+Remember:
+Your goal is to support me in solving the task successfully.
+Unless I say the task is complete, always strive for a detailed, accurate, and useful solution.
+"""
+
+
+plan_system_prompt = """
+===== USER INSTRUCTIONS ===== 
+Remember that you are the user, and I am the assistant. I will always follow your instructions. We are working together to successfully complete a task.
+My role is to help you accomplish a difficult task. You will guide me step by step based on my expertise and your needs. Your instructions should be in the following format: Instruction: [YOUR INSTRUCTION], where "Instruction" is a sub-task or question.
+You should give me one instruction at a time. I will respond with a solution for that instruction. You should instruct me rather than asking me questions.
+
+Please note that the task may be complex. Do not attempt to solve it all at once. You should break the task down and guide me step by step.
+Here are some tips to help you give better instructions: 
+<tips>
+- I have access to various tools like search, web browsing, document management, and code execution. Think about how humans would approach solving the task step by step, and give me instructions accordingly. For example, you may first use Google search to gather initial information and a URL, then retrieve the content from that URL, or interact with a webpage to find the answer.
+- Even if the task is complex, there is always a solution. If you can’t find the answer using one method, try another approach or use different tools to find the solution.
+- Always remind me to verify the final answer using multiple tools (e.g., screenshots, webpage analysis, etc.), or other methods.
+- If I’ve written code, remind me to run it and check the results.
+- Search results generally don’t give direct answers. Focus on finding sources through search, and use other tools to process the URL or interact with the webpage content.
+- If the task involves a YouTube video, I will need to process the content of the video.
+- For file downloads, use web browser tools or write code (e.g., download from a GitHub link).
+- Feel free to write code to solve tasks like Excel-related tasks.
+</tips>
+
+Now, here is the overall task: <task>{task}</task>. Stay focused on the task!
+
+Start giving me instructions step by step. Only provide the next instruction after I’ve completed the current one. When the task is finished, respond with <TASK_DONE>.
+Do not say <TASK_DONE> until I’ve completed the task.
+"""
+
+plan_done_prompt = """\n
+Below is some additional information about the overall task that can help you better understand the purpose of the current task: 
+<auxiliary_information>
+{task}
+</auxiliary_information>
+If there are any available tools that can assist with the task, instead of saying "I will...", first call the tool and respond based on the results it provides. Please also specify which tool you used.
+"""
+
+plan_postfix_prompt = """\n
+Now, please provide the final answer to the original task based on our conversation: <task>{task}</task>
+Pay close attention to the required answer format. First, analyze the expected format based on the question, and then generate the final answer accordingly.
+Your response should include the following:
+- Analysis: Enclosed within <analysis> </analysis>, this section should provide a detailed breakdown of the reasoning process.
+- Final Answer: Enclosed within <final_answer> </final_answer>, this section should contain the final answer in the required format.
+Here are some important guidelines for formatting the final answer:
+<hint>
+- Your final answer must strictly follow the format specified in the question. The answer should be a single number, a short string, or a comma-separated list of numbers and/or strings:
+- If the answer is a number, don't use commas as thousands separators, and don't  include units (such as "$" or "%") unless explicitly required. 
+- If the answer is a string, don't include articles (e.g., "a", "the"), don't use abbreviations (e.g., city names), and write numbers in full words unless instructed otherwise. 
+- If the answer is a comma-separated list, apply the above rules based on whether each element is a number or a string.
+</hint>
+"""

examples/plan_execute/run.py

@@ -0,0 +1,51 @@
+# coding: utf-8
+# Copyright (c) 2025 inclusionAI.
+
+from aworld.config.conf import ModelConfig, AgentConfig
+from aworld.core.agent.swarm import Swarm, GraphBuildType
+from aworld.core.task import Task
+from aworld.runner import Runners
+from examples.plan_execute.agent import PlanAgent, ExecuteAgent
+from examples.plan_execute.mock import mock_dataset
+from examples.tools.common import Agents, Tools
+
+
+def main():
+    test_sample = mock_dataset("gaia")
+
+    model_config = ModelConfig(
+        llm_provider="openai",
+        llm_temperature=1,
+        llm_model_name="gpt-4o",
+        # need to set llm_api_key for use LLM
+    )
+
+    agent1_config = AgentConfig(
+        name=Agents.PLAN.value,
+        llm_config=model_config
+    )
+    agent1 = PlanAgent(conf=agent1_config, step_reset=False)
+
+    agent2_config = AgentConfig(
+        name=Agents.EXECUTE.value,
+        llm_config=model_config
+    )
+    agent2 = ExecuteAgent(conf=agent2_config, step_reset=False, tool_names=[Tools.DOCUMENT_ANALYSIS.value])
+
+    # Create swarm for multi-agents
+    # define (head_node1, tail_node1), (head_node1, tail_node1) edge in the topology graph
+    swarm = Swarm((agent1, agent2), build_type=GraphBuildType.HANDOFF)
+
+    # Define a task
+    task_id = 'task'
+    task = Task(id=task_id, input=test_sample, swarm=swarm, endless_threshold=10)
+
+    # Run task
+    result = Runners.sync_run_task(task=task)
+
+    print(f"Time cost: {result[task_id].time_cost}")
+    print(f"Task Answer: {result[task_id].answer}")
+
+
+if __name__ == '__main__':
+    main()

examples/plan_execute/utils.py

@@ -0,0 +1,18 @@
+import re
+from typing import Any, Dict, List, Literal, Optional, Union, Tuple
+import logging as logger
+
+
+def extract_pattern(content: str, pattern: str) -> Optional[str]:
+    try:
+        _pattern = fr"<{pattern}>(.*?)</{pattern}>"
+        match = re.search(_pattern, content, re.DOTALL)
+        if match:
+            text = match.group(1)
+            return text.strip()
+        else:
+            return None
+    except Exception as e:
+        logger.warning(f"Error extracting answer: {e}, current content: {content}")
+        return None
+