complete log analysis

Author: CTY-git

patchwork/app.py

@@ -8,12 +8,11 @@
 from collections import deque
 from contextlib import nullcontext
 from pathlib import Path
-from typing import Any
 
 import click
 import yaml
 from click import echo
-from typing_extensions import Iterable
+from typing_extensions import Any, Iterable
 
 from patchwork.common.client.patched import PatchedClient
 from patchwork.common.constants import PROMPT_TEMPLATE_FILE_KEY

patchwork/common/client/llm/utils.py

@@ -1,12 +1,11 @@
 from __future__ import annotations
 
 import json
-from typing import Any, Dict, List, Optional, Type
 
 from openai.lib._parsing._completions import type_to_response_format_param
 from openai.types.chat.completion_create_params import ResponseFormat
 from pydantic import BaseModel, Field, create_model
-from typing_extensions import List
+from typing_extensions import Any, Dict, List, Optional, Type
 
 from patchwork.logger import logger
 

patchwork/common/client/patched.py

@@ -9,13 +9,13 @@
 from importlib import metadata
 from pathlib import Path
 from threading import Thread
-from typing import Any
 
 import click
 import requests
 from git.repo.base import Repo
 from requests import Response, Session
 from requests.adapters import DEFAULT_POOLBLOCK, HTTPAdapter
+from typing_extensions import Any
 from urllib3 import HTTPConnectionPool, HTTPSConnectionPool, PoolManager
 
 from patchwork.common.utils.user_config import get_user_config

patchwork/common/client/sonar.py

@@ -1,8 +1,8 @@
 from collections import defaultdict
 from dataclasses import dataclass
-from typing import Optional, Union
 
 import requests
+from typing_extensions import Optional, Union
 
 
 @dataclass

patchwork/common/multiturn_strategy/agentic_strategy.py

@@ -1,122 +1,31 @@
 from __future__ import annotations
 
-import abc
-import json
 import logging
-import random
-import string
 import sys
-from json import JSONDecodeError
-from pathlib import Path
-from typing import Union, Any
 
-import chevron
-from openai.types.chat import ChatCompletionMessageParam
-from openai.types.chat.chat_completion_tool_param import ChatCompletionToolParam
+from pydantic import BaseModel
 from pydantic_ai import Agent
 from pydantic_ai.models.anthropic import AnthropicModel
-from pydantic import BaseModel
-
-from patchwork.common.client.llm.protocol import LlmClient
-from patchwork.common.client.llm.utils import example_string_to_base_model, example_json_to_base_model
-from patchwork.common.tools import CodeEditTool, Tool
-from patchwork.common.tools.agentic_tools import EndTool
-
-
-class Role(abc.ABC):
-    def __init__(self, llm_client: LlmClient, tool_set: dict[str, Tool]):
-        self.llm_client = llm_client
-        self.tool_set = tool_set
-        self.history: list[ChatCompletionMessageParam] = []
-
-    def generate_reply(self, message: str) -> str:
-        self.history.append(dict(role="user", content=message))
-        input_kwargs = dict(
-            messages=self.history,
-            model="claude-3-5-sonnet-latest",
-            tools=self.__get_tools_spec(),
-            max_tokens=8096,
-        )
-        is_prompt_safe = self.llm_client.is_prompt_supported(**input_kwargs)
-        if is_prompt_safe < 0:
-            raise ValueError("The subsequent prompt is not supported, due to large size.")
-        response = self.llm_client.chat_completion(**input_kwargs)
-        choices = response.choices or []
-
-        message_content = ""
-        for choice in choices:
-            new_message = choice.message.to_dict()
-            self.history.append(new_message)
-            if new_message.get("tool_calls") is not None:
-                self.history.extend(self.__execute_tools(new_message))
-            else:
-                message_content = new_message["content"]
-
-        return message_content
-
-    def __execute_tools(self, last_message: ChatCompletionMessageParam) -> list[ChatCompletionMessageParam]:
-        rv = []
-        for tool_call in last_message.get("tool_calls", []):
-            tool_name_to_use = tool_call.get("function", {}).get("name")
-            tool_to_use = self.tool_set.get(tool_name_to_use, None)
-            if tool_to_use is None:
-                logging.info("LLM just used an non-existent tool!")
-                continue
-
-            logging.info(f"Running tool: {tool_name_to_use}")
-            try:
-                tool_arguments = tool_call.get("function", {}).get("arguments", "{}")
-                tool_kwargs = json.loads(tool_arguments)
-                tool_output = tool_to_use.execute(**tool_kwargs)
-            except JSONDecodeError:
-                tool_output = "Arguments must be passed through a valid JSON object"
-
-            rv.append({"tool_call_id": tool_call.get("id", ""), "role": "tool", "content": tool_output})
-
-        return rv
-
-    def __get_tools_spec(self) -> list[ChatCompletionToolParam]:
-        return [
-            dict(
-                type="function",
-                function={"name": k, **v.json_schema},
-            )
-            for k, v in self.tool_set.items()
-        ]
-
-
-class UserProxy(Role):
-    def __init__(
-        self, llm_client: LlmClient, tool_set: dict[str, Tool], system_prompt: str = None, reply_message: str = ""
-    ):
-        super().__init__(llm_client, tool_set)
-        if system_prompt is not None:
-            self.history.append(dict(role="system", content=system_prompt))
-
-        self.__reply_message = reply_message
-
-    def generate_reply(self, message: str) -> str:
-        if self.__reply_message is not None:
-            self.history.append(dict(role="user", content=message))
-            self.history.append(dict(role="assistant", content=self.__reply_message))
-            return self.__reply_message
-        else:
-            return super().generate_reply(message)
-
-
-class Assistant(Role):
-    def __init__(self, llm_client: LlmClient, tool_set: dict[str, Tool], system_prompt: str = None):
-        super().__init__(llm_client, tool_set)
-        if system_prompt is not None:
-            self.history.append(dict(role="system", content=system_prompt))
+from typing_extensions import Any, Dict, Optional, Union
 
+from patchwork.common.client.llm.utils import example_json_to_base_model
+from patchwork.common.tools import Tool
+from patchwork.common.utils.utils import mustache_render
 
+_COMPLETION_FLAG_ATTRIBUTE = "is_task_completed"
+_MESSAGE_ATTRIBUTE = "message"
 
 
 class AgentConfig(BaseModel):
+    class Config:
+        arbitrary_types_allowed = True
+
     name: str
-    tool_set: dict[str, Tool]
-    system_prompt: str = ''
+    tool_set: Dict[str, Tool]
+    system_prompt: str = ""
+    example_json: Union[
+        str, Dict[str, Any]
+    ] = f'{{"{_MESSAGE_ATTRIBUTE}":"message", "{_COMPLETION_FLAG_ATTRIBUTE}": false}}'
 
 
 class AgenticStrategy:
@@ -128,73 +37,75 @@ def __init__(
         user_prompt_template: str,
         agent_configs: list[AgentConfig],
         example_json: Union[str, dict[str, Any]] = '{"output":"output text"}',
-        *args,
-        **kwargs,
+        limit: Optional[int] = None,
     ):
+        self.__limit = limit
         self.__template_data = template_data
         self.__user_prompt_template = user_prompt_template
         model = AnthropicModel("claude-3-5-sonnet-latest", api_key=api_key)
-        self.__user_role = Agent(
+        self.__summariser = Agent(
             model,
-            system_prompt=self.__render_prompt(system_prompt_template),
+            system_prompt=mustache_render(system_prompt_template, self.__template_data),
             result_type=example_json_to_base_model(example_json),
+            model_settings=dict(parallel_tool_calls=False),
         )
-        self.__assistants = []
-        for assistant_config in agent_configs:
+        self.__agents = []
+        for agent_config in agent_configs:
             tools = []
-            for tool in assistant_config.tool_set.values():
+            for tool in agent_config.tool_set.values():
                 tools.append(tool.to_pydantic_ai_function_tool())
-            assistant = Agent(
-                "claude-3-5-sonnet-latest",
-                system_prompt=self.__render_prompt(assistant_config.system_prompt),
-                tools=tools
+            agent = Agent(
+                model,
+                name=agent_config.name,
+                system_prompt=mustache_render(agent_config.system_prompt, self.__template_data),
+                tools=tools,
+                result_type=example_json_to_base_model(agent_config.example_json),
+                model_settings=dict(parallel_tool_calls=False),
             )
 
-            self.__assistants.append(assistant)
+            self.__agents.append(agent)
 
-    def __render_prompt(self, prompt_template: str) -> str:
-        chevron.render.__globals__["_html_escape"] = lambda x: x
-        return chevron.render(
-            template=prompt_template,
-            data=self.__template_data,
-            partials_path=None,
-            partials_ext="".join(random.choices(string.ascii_uppercase + string.digits, k=32)),
-            partials_dict=dict(),
-        )
-
-    def __is_session_completed(self) -> bool:
-        for message in reversed(self.__assistant_role.history):
-            if message.get("tool") is not None:
-                continue
-            if message.get("content") == EndTool.MESSAGE:
-                return True
-
-        return False
-
-    def execute(self, limit: int | None = None) -> None:
-        message = self.__render_prompt(self.__user_prompt_template)
+    def execute(self, limit: Optional[int] = None) -> dict:
+        agents_result = dict()
         try:
-            for i in range(limit or self.__limit or sys.maxsize):
-                self.__user_role.run_sync(self.__user_prompt_template)
-                self.run_count = i + 1
-                for role in [*self.__assistants, self.__user_role]:
-                    message = role.run_sync(message)
-                if self.__is_session_completed():
-                    break
+            for index, agent in enumerate(self.__agents):
+                user_message = mustache_render(self.__user_prompt_template, self.__template_data)
+                message_history = None
+                agent_output = None
+                for i in range(limit or self.__limit or sys.maxsize):
+                    agent_output = agent.run_sync(user_message, message_history=message_history)
+                    message_history = agent_output.all_messages()
+                    if getattr(agent_output.data, _COMPLETION_FLAG_ATTRIBUTE, False):
+                        break
+                    user_message = "Please continue"
+                agents_result[index] = agent_output
         except Exception as e:
             logging.error(e)
-        finally:
-            self.run_count = 0
 
-    @property
-    def history(self):
-        return self.__user_role.history
+        if len(agents_result) == 0:
+            return dict()
+
+        if len(agents_result) == 1:
+            final_result = self.__summariser.run_sync(
+                "From the actions taken by the assistant. Please give me the result.",
+                message_history=next(v for _, v in agents_result.items()).all_messages(),
+            )
+        else:
+            agent_summaries = []
+            for agent_result in agents_result.values():
+                agent_summary_result = self.__summariser.run_sync(
+                    "From the actions taken by the assistant. Please give me the result.",
+                    message_history=agent_result.all_messages(),
+                )
+                agent_summary = getattr(agent_summary_result.data, _MESSAGE_ATTRIBUTE, None)
+                if agent_summary is None:
+                    continue
+
+                agent_summaries.append(agent_summary)
+            agent_summary_list = "\n* " + "\n* ".join(agent_summaries)
+            final_result = self.__summariser.run_sync(
+                "Please give me the result from the following summary of what the assistants have done."
+                + agent_summary_list,
+            )
 
-    @property
-    def tool_records(self):
-        for tool in self.tool_set.values():
-            if isinstance(tool, CodeEditTool):
-                cwd = Path.cwd()
-                modified_files = [file_path.relative_to(cwd) for file_path in tool.tool_records["modified_files"]]
-                return [dict(path=str(file)) for file in modified_files]
-        return []
+        return final_result.data.dict()

patchwork/common/tools/__init__.py

@@ -1,9 +1,12 @@
 from patchwork.common.tools.bash_tool import BashTool
 from patchwork.common.tools.code_edit_tools import CodeEditTool
+from patchwork.common.tools.grep_tool import FindTextTool, FindTool
 from patchwork.common.tools.tool import Tool
 
 __all__ = [
     "Tool",
     "CodeEditTool",
     "BashTool",
+    "FindTool",
+    "FindTextTool",
 ]

patchwork/common/tools/bash_tool.py

@@ -3,6 +3,8 @@
 import subprocess
 from pathlib import Path
 
+from typing_extensions import Optional
+
 from patchwork.common.tools.tool import Tool
 
 
@@ -35,9 +37,7 @@ def json_schema(self) -> dict:
 
     def execute(
         self,
-        command: str | None = None,
-        *args,
-        **kwargs,
+        command: Optional[str] = None,
     ) -> str:
         """Execute editor commands on files in the repository."""
         if command is None: