Autogen integration

README.md

@@ -64,9 +64,7 @@ By default, if the `SMART_ACCOUNT_ADDRESS` environment variable is not defined,
 AutoTx can be connected to your existing smart account by doing the following:
 
 1. Set the `SMART_ACCOUNT_ADDRESS` to the address of your smart account in your `.env`. This tells AutoTx which account it should interact with.
-
 2. AutoTx's agent address, which it generates locally, must be set as a signer in your Safe's configuration to allow it to create transactions on behalf of the smart account. To get this address, run `poetry run agent address`.
-
 3. Update the `CHAIN_RPC_URL` value in your `.env` with the correct RPC URL of the network where your smart account is deployed.
 
 

autotx/AutoTx.py

@@ -1,73 +1,98 @@
-from typing import Optional, Callable
+from textwrap import dedent
+from typing import Any, Dict, Optional, Callable
 from dataclasses import dataclass
 from typing import Optional
-from crewai import Agent, Crew, Process, Task
+from autotx.autotx_agent import AutoTxAgent
 from autotx.utils.PreparedTx import PreparedTx
 from autotx.utils.agent.build_goal import build_goal
-from autotx.utils.agent.define_tasks import define_tasks
-from langchain_core.tools import StructuredTool
-from crewai import Agent, Crew, Process, Task
 from autotx.utils.ethereum import SafeManager
 from autotx.utils.ethereum.networks import NetworkInfo
-from autotx.utils.llm import open_ai_llm
+from autogen import UserProxyAgent, AssistantAgent, Agent
+import autogen
 
 @dataclass(kw_only=True)
 class Config:
     verbose: bool
 
 class AutoTx:
     manager: SafeManager
-    agents: list[Agent]
     config: Config = Config(verbose=False)
     transactions: list[PreparedTx] = []
     network: NetworkInfo
+    get_llm_config: Callable[[], Optional[Dict[str, Any]]]
+    user_proxy: UserProxyAgent
+    agent_factories: list[Callable[['AutoTx', UserProxyAgent, Optional[Dict[str, Any]]], AutoTxAgent]]
 
     def __init__(
-        self, manager: SafeManager, network: NetworkInfo, agent_factories: list[Callable[['AutoTx'], Agent]], config: Optional[Config]
+        self, manager: SafeManager, network: NetworkInfo, agent_factories: list[Callable[['AutoTx', UserProxyAgent, Optional[Dict[str, Any]]], Agent]], config: Optional[Config],
+        get_llm_config: Callable[[], Optional[Dict[str, Any]]]
     ):
         self.manager = manager
         self.network = network
+        self.get_llm_config = get_llm_config
         if config:
             self.config = config
-        self.agents = [factory(self) for factory in agent_factories]
+        self.agent_factories = agent_factories
 
     def run(self, prompt: str, non_interactive: bool):
-        print(f"Defining goal for prompt: '{prompt}'")
-
-        agents_information = self.get_agents_information()
+        print("Running AutoTx with the following prompt: ", prompt)
+
+        user_proxy = UserProxyAgent(
+            name="user_proxy",
+            is_termination_msg=lambda x: x.get("content", "") and x.get("content", "").rstrip().endswith("TERMINATE"),
+            human_input_mode="NEVER",
+            max_consecutive_auto_reply=20,
+            system_message=f"You are a user proxy. You will be interacting with the agents to accomplish the tasks.",
+            llm_config=self.get_llm_config(),
+            code_execution_config=False,
+        )
+        agents = [factory(self, user_proxy, self.get_llm_config()) for factory in self.agent_factories]
+
+        agents_information = self.get_agents_information(agents)
 
         goal = build_goal(prompt, agents_information, self.manager.address, non_interactive)
 
-        print(f"Defining tasks for goal: '{goal}'")
-        tasks: list[Task] = define_tasks(goal, agents_information, self.agents)
-
-        self.run_for_tasks(tasks, non_interactive)
+        verifier_agent = AssistantAgent(
+            name="verifier",
+            is_termination_msg=lambda x: x.get("content", "") and x.get("content", "").rstrip().endswith("TERMINATE"),
+            system_message=dedent(
+                    """
+                    You are an expert in verifiying if user goals are met.
+                    You analyze chat and respond with TERMINATE if the goal is met.
+                    You can consider the goal met if the other agents have prepared the necessary transactions.
+                    """
+                ),
+            llm_config=self.get_llm_config(),
+            human_input_mode="NEVER",
+            code_execution_config=False,
+        )
 
-    def run_for_tasks(self, tasks: list[Task], non_interactive: bool):
-        print(f"Running tasks...")
-        Crew(
-            agents=self.agents,
-            tasks=tasks,
-            verbose=self.config.verbose,
-            process=Process.sequential,
-            function_calling_llm=open_ai_llm,
-        ).kickoff()
+        groupchat = autogen.GroupChat(agents=[agent.autogen_agent for agent in agents] + [user_proxy, verifier_agent], messages=[], max_round=12)
+        manager = autogen.GroupChatManager(groupchat=groupchat, llm_config=self.get_llm_config())
+
+        user_proxy.initiate_chat(manager, message=dedent(
+            f"""
+                My goal is: {prompt}
+                Advisor reworded: {goal}
+            """
+        ))
 
         self.manager.send_tx_batch(self.transactions, require_approval=not non_interactive)
         self.transactions.clear()
+
 
-    def get_agents_information(self) -> str:
+    def get_agents_information(self, agents: list[AutoTxAgent]) -> str:
         agent_descriptions = []
-        for agent in self.agents:
-            agent_default_tools: list[StructuredTool] = agent.tools
+        for agent in agents:
             tools_available = "\n".join(
                 [
-                    f"  - Name: {tool.name}\n  - Description: {tool.description} \n"
-                    for tool in agent_default_tools
+                    f"\n- {tool}"
+                    for tool in agent.tools
                 ]
             )
-            description = f"Agent name: {agent.name}\nRole: {agent.role}\nTools available:\n{tools_available}"
+            description = f"Agent name: {agent.autogen_agent.name}\nTools available:{tools_available}"
             agent_descriptions.append(description)
 
         agents_information = "\n".join(agent_descriptions)
-        return agents_information
+
+        return agents_information

autotx/agents/ExampleAgent.py

@@ -1,51 +1,46 @@
-from typing import Callable
 from textwrap import dedent
-from crewai import Agent
 from autotx.AutoTx import AutoTx
-from autotx.auto_tx_agent import AutoTxAgent
-from autotx.auto_tx_tool import AutoTxTool
+from typing import Annotated, Callable
+from autogen import AssistantAgent, UserProxyAgent, Agent
+from autotx.autotx_agent import AutoTxAgent
 
-class ExampleTool(AutoTxTool):
-    name: str = "Example tool that does something useful"
-    description: str = dedent(
-        """
-        This tool does something very useful.
+example_info = {
+    "name": "example_tool",
+    "description": "Example of an agent tool."
+}
 
-        Args:
-            amount (float): Amount of something.
-            receiver (str): The receiver of something.
-        Returns:
-            The result of the useful tool in a useful format.
-        """
-    )
-
-    def _run(
-        self, amount: float, receiver: str
-    ) -> str:
-
-        # TODO: do something useful
-        print(f"ExampleTool run: {amount} {receiver}")
-
-        # NOTE: you can add transactions to AutoTx's current bundle
-        # self.autotx.transactions.append(tx)
+def build_agent_factory() -> Callable[[AutoTx, UserProxyAgent, dict], Agent]:
+    def agent_factory(autotx: AutoTx, user_proxy: UserProxyAgent, llm_config: dict) -> AutoTxAgent:
+        agent = AssistantAgent(
+            name="swap-tokens",
+            system_message=dedent(f"""
+                Example of an agent system message.
+                """
+            ),
+            llm_config=llm_config,
+            human_input_mode="NEVER",
+            code_execution_config=False,
+        )
+
+        @user_proxy.register_for_execution()
+        @agent.register_for_llm(
+            name=example_info["name"],
+            description=example_info["description"]
+        )
+        def example_tool(
+            amount: Annotated[float, "Amount of something."],
+            receiver: Annotated[str, "The receiver of something."]
+        ) -> str:
+            # TODO: do something useful
+            print(f"ExampleTool run: {amount} {receiver}")
+
+            # NOTE: you can add transactions to AutoTx's current bundle
+            # self.autotx.transactions.append(tx)
 
-        return f"Something useful has been done with {amount} to {receiver}"
+            return f"Something useful has been done with {amount} to {receiver}"
 
-class ExampleAgent(AutoTxAgent):
-    def __init__(self, autotx: AutoTx):
-        super().__init__(
-            name="example-agent",
-            role="Example agent role",
-            goal="Example agent goal",
-            backstory="Example agent backstory",
-            tools=[
-                ExampleTool(autotx),
-                # AnotherTool(...),
-                # AndAnotherTool(...)
-            ],
-        )
+        return AutoTxAgent(agent, tools=[
+            f"{example_info['name']}: {example_info['description']}"
+        ])
 
-def build_agent_factory() -> Callable[[AutoTx], Agent]:
-    def agent_factory(autotx: AutoTx) -> ExampleAgent:
-        return ExampleAgent(autotx)
     return agent_factory