diff --git a/app/agents/agent.py b/app/agents/agent.py index 81f5f9f..23ed897 100644 --- a/app/agents/agent.py +++ b/app/agents/agent.py @@ -15,12 +15,31 @@ from app.agents.tools import ( read_file_tool, read_file_tool_def, list_files_tool, list_files_tool_def, + write_to_file_tool, write_to_file_tool_def, ) from app.agents.web_search_tool import web_search_tool, web_search_tool_def from app.agents.memory_tool import save_memory_tool, save_memory_tool_def from app.memory.memory_manager import MemoryManager from app.safety.custom_guardrails import CustomGuardrailManager +def _extract_json_from_response(response_text: str) -> Optional[str]: + """ + Extracts a JSON object from a string that might be wrapped in markdown code blocks. + """ + # Regex to find JSON wrapped in ```json ... ``` or just ``` ... ``` + match = re.search(r'```(json\s*)?(?P\{.*?\})```', response_text, re.DOTALL) + if match: + return match.group('json') + + # If no markdown block is found, assume the whole string is a JSON object + # and try to find the start of a JSON object. + first_brace = response_text.find('{') + last_brace = response_text.rfind('}') + if first_brace != -1 and last_brace != -1: + return response_text[first_brace:last_brace+1] + + return None + class ToolExecutionError(Exception): """Custom exception for errors during tool execution.""" pass @@ -28,18 +47,22 @@ class ToolExecutionError(Exception): # Новый системный промпт, основанный на ReAct (Reason + Act) REACT_SYSTEM_PROMPT = """ You are a smart, autonomous AI agent. Your name is {agent_name}, and your role is {agent_role}. -Your ultimate goal is: {agent_goal}. + +Your current, specific task is described below. Focus ONLY on this task. +--- +{task_description} +--- You operate in a loop of Thought, Action, and Observation. At each step, you MUST respond in a specific JSON format. Your entire response must be a single JSON object. -1. **Thought**: First, think about your plan. Analyze the user's request, your goal, and the previous steps. Describe your reasoning for the current action. This is a mandatory field. +1. **Thought**: First, think about your plan to accomplish your assigned task. Analyze the task description and the previous steps. Describe your reasoning for the current action. This is a mandatory field. 2. **Action** or **Answer**: Based on your thought, you must choose ONE of the following: a. `action`: An object representing the tool to use. It must contain: - `name`: The name of the tool to execute. - `input`: An object with the parameters for the tool. - b. `answer`: A final, comprehensive answer to the user's request. Use this ONLY when the task is fully complete. + b. `answer`: A final, comprehensive summary of what you did to complete YOUR ASSIGNED TASK. Use this ONLY when your specific task is fully complete. # AVAILABLE TOOLS: You have access to the following tools. Use them to gather information and perform actions. @@ -70,7 +93,7 @@ class ToolExecutionError(Exception): - Your ENTIRE output MUST be a single, valid JSON object. Do not add any text before or after the JSON. - You must choose either `action` or `answer`, not both. - The `thought` field is always required. -- Think step by step. Your goal is to complete the task, not just use tools. +- Once you have completed your specific task, you MUST use the 'answer' field to finish your work. """ # Настройка логирования @@ -110,11 +133,8 @@ def __init__( # Initialize Guardrails self.guardrail_manager = CustomGuardrailManager(api_key=api_key) - # Add default tools - self.add_tool(read_file_tool, read_file_tool_def) - self.add_tool(list_files_tool, list_files_tool_def) - self.add_tool(web_search_tool, web_search_tool_def) - self.add_tool(save_memory_tool, save_memory_tool_def) + # Tools are now added by the AgentFactory based on the role's configuration. + # The base agent starts with an empty toolset. # RAG specific attributes self.use_rag = use_rag @@ -137,64 +157,6 @@ def add_tool(self, tool_func: Callable, tool_definition: Dict[str, Any]): self.tools[tool_name] = tool_func self.tool_definitions.append(tool_definition) - def get_openai_tools(self) -> Optional[List[Dict[str, Any]]]: - """Возвращает список определений инструментов для OpenAI API.""" - if not self.tool_definitions: - return None - return self.tool_definitions - - def _create_rag_query(self, briefing: str) -> str: - """ - Creates a focused query for RAG from the detailed briefing. - Extracts the current task and the last result from history. - """ - task_match = re.search(r"\*\*YOUR CURRENT TASK \(Step .*\):\*\*\n\n\*\*Task:\*\* (.*)\n\*\*Description:\*\* (.*)", briefing) - history_match = re.search(r"\*\*EXECUTION HISTORY:\*\*\n(.*)", briefing, re.DOTALL) - - if not task_match: - return briefing # Fallback to full briefing - - task = task_match.group(1) - description = task_match.group(2) - - query_parts = [f"Task: {task}", f"Description: {description}"] - - if history_match: - history_str = history_match.group(1).strip() - # Get the last entry from the history - last_entry = history_str.split("- **Step")[0].strip() - if last_entry: - query_parts.append(f"Context from previous step: {last_entry}") - - focused_query = "\n".join(query_parts) - logging.info(f"Created focused RAG query for {self.name}: '{focused_query}'") - return focused_query - - def _enrich_with_knowledge(self, query: str) -> str: - """Enriches a query with context from the knowledge base if RAG is enabled.""" - if not self.use_rag or not self.retriever: - return "" - - top_k = self.rag_config.get("top_k", 3) - filters = self.rag_config.get("filters", None) - - logging.info(f"Agent '{self.name}' is retrieving knowledge with top_k={top_k}, filters={filters}") - retrieved_knowledge = self.retriever.retrieve(query=query, top_k=top_k, filters=filters) - - if not retrieved_knowledge: - logging.info("No specific internal standards found for this query.") - return "" - - formatted_knowledge = "\n\n---\n\n".join( - [f"Source: {chunk['source']}\n\n{chunk['text']}" for chunk in retrieved_knowledge] - ) - knowledge_context = ( - "Before you begin, consult these internal standards and best practices:" - f"\n\n--- RELEVANT KNOWLEDGE ---\n{formatted_knowledge}\n--------------------------\n" - ) - logging.info(f"Knowledge context added for agent '{self.name}'.") - return knowledge_context - def _get_memory_context(self) -> str: """Retrieves relevant facts from long-term memory.""" recent_facts = self.memory_manager.get_recent_facts(limit=10) @@ -237,152 +199,139 @@ def _get_tools_description(self) -> str: ) return "\n".join(desc) - def _get_user_input(self) -> Optional[str]: - """Обрабатывает получение ввода от пользователя.""" - print("\033[94mВы:\033[0m ", end="") - try: - user_input = input() - return user_input if user_input.strip() else None - except EOFError: - return None - def _get_model_response(self) -> str: - """ - Отправляет текущую историю беседы в OpenAI и возвращает текстовый ответ модели. - """ + """Получает и возвращает ответ от модели.""" + logging.info(f"Sending request to OpenAI with {len(self.conversation_history)} messages.") try: - logging.info(f"Sending request to OpenAI with {len(self.conversation_history)} messages.") response = self.client.chat.completions.create( model=self.model, messages=self.conversation_history, - # Убираем tools и tool_choice, так как теперь мы парсим JSON + # tools=self.get_openai_tools(), + # tool_choice="auto", + temperature=0.1, + top_p=0.1, + response_format={"type": "json_object"} ) return response.choices[0].message.content or "" - except Exception as e: - logging.error(f"Error calling OpenAI API: {e}", exc_info=True) - return json.dumps({ - "thought": "An API error occurred. I cannot proceed.", - "answer": f"API Error: {e}" - }) + except APIError as e: + logging.error("OpenAI API error: %s", e) + return f'{{"thought": "Encountered an API error. I should try again.", "action": {{"name": "wait", "input": {{"seconds": 2}}}}}}' def execute_task(self, briefing: str) -> str: """ - Выполняет одну задачу на основе предоставленного брифинга, используя ReAct цикл. + Основной цикл выполнения задачи агентом. """ - logging.info(f"Agent {self.name} received task.") - - # 1. Prepare Prompts - knowledge_context = self._enrich_with_knowledge(self._create_rag_query(briefing)) if self.use_rag else "" + # 1. Формируем системный промпт tools_description = self._get_tools_description() memory_context = self._get_memory_context() - + + # The 'briefing' from the orchestrator is the definitive task. system_prompt = self.system_prompt_template.format( agent_name=self.name, agent_role=self.role, - agent_goal=self.goal, + task_description=briefing, # The briefing is the task tools_description=tools_description, memory_context=memory_context ) - final_system_prompt = f"{knowledge_context}\n{system_prompt}" + self.conversation_history = [{"role": "system", "content": system_prompt}] + # A simple message to kick off the process. + self.conversation_history.append({"role": "user", "content": "Begin your task."}) - self.conversation_history = [ - {"role": "system", "content": final_system_prompt}, - {"role": "user", "content": briefing} - ] - - # 2. Start ReAct Loop + # 2. Основной цикл Reason-Act for i in range(self.max_iterations): logging.info(f"--- Iteration {i+1}/{self.max_iterations} ---") - model_response_str = self._get_model_response() - self.conversation_history.append({"role": "assistant", "content": model_response_str}) - + # 3. Получаем ответ от модели + response_text = self._get_model_response() + + # --- Guardrails Input Check --- + # guardrail_verdict = self.guardrail_manager.check_input(response_text) + # if guardrail_verdict: + # final_answer = f"Input check failed: {guardrail_verdict}" + # logging.warning(final_answer) + # self.conversation_history.append({"role": "assistant", "content": final_answer}) + # return final_answer + + # 4. Парсим JSON из ответа + json_response = None try: - parsed_response = json.loads(model_response_str) - thought = parsed_response.get("thought") - if thought: - logging.info(f"🤖 Thought: {thought}") - else: - raise ValueError("Missing 'thought' in response.") - - if parsed_response.get("answer"): - final_answer = parsed_response["answer"] - # Validate final answer with guardrails before returning - validated_answer = self.guardrail_manager.validate_and_format_response(final_answer) - logging.info(f"✅ Agent {self.name} finished task with answer: {validated_answer}") - return validated_answer - - elif parsed_response.get("tool"): - tool_name = parsed_response["tool"]["name"] - tool_input = parsed_response["tool"]["input"] - - if not tool_name: - raise ValueError("Missing 'name' in action.") - - logging.info(f"🛠️ Action: Calling tool '{tool_name}' with input: {tool_input}") - tool_result = self._execute_tool(tool_name, tool_input) - - observation = f"Tool '{tool_name}' returned:\n```\n{tool_result}\n```" - logging.info(f"👀 Observation: {observation}") - self.conversation_history.append({"role": "user", "content": observation}) - else: - raise ValueError("Response must contain 'action' or 'answer'.") - - except ToolExecutionError as e: - error_message = f"A tool failed to execute: {e}" - logging.error(error_message) - # Запускаем цикл рефлексии - reflection_prompt = ( - f"CRITICAL_ERROR: Your last action failed with the following error: '{e}'.\n" - "You MUST analyze this error and the execution history to understand what went wrong.\n" - "Then, devise a new plan. Either try a different approach, use a different tool, or modify the input to the tool.\n" - "Your next 'thought' MUST explain how you are correcting your course of action." - ) - self.conversation_history.append({"role": "user", "content": reflection_prompt}) - continue # Продолжаем цикл, чтобы агент мог ответить на сообщение об ошибке + # Используем новую функцию для извлечения JSON + cleaned_response = _extract_json_from_response(response_text) + if not cleaned_response: + raise json.JSONDecodeError("No JSON object found in response", response_text, 0) + + json_response = json.loads(cleaned_response) + + self.conversation_history.append({"role": "assistant", "content": cleaned_response}) - except (json.JSONDecodeError, ValueError) as e: - error_message = f"Error parsing model response: {e}. Response was: '{model_response_str}'" + except json.JSONDecodeError: + error_message = f"Error parsing model response: Expecting value: line 1 column 1 (char 0). Response was: '{response_text}'" logging.error(error_message) - # Даем агенту шанс исправиться - error_feedback = ( - f"Error: Your last response was not a valid JSON object. " - f"Please correct your output to strictly follow the required format. " - f"The `thought` field is mandatory, and you must include either an `action` or an `answer`. " - f"Error details: {e}" - ) - self.conversation_history.append({"role": "user", "content": error_feedback}) - continue - - warning_message = f"Agent {self.name} reached max iterations ({self.max_iterations}) without a final answer." - logging.warning(warning_message) - # Validate the warning message as well, in case it contains sensitive info (less likely but good practice) - return warning_message - - def run(self) -> None: - """Запускает основной цикл общения с агентом в интерактивном режиме.""" - print(f"Запуск агента '{self.name}' в интерактивном режиме. Введите 'exit' для завершения.") - - while True: - try: - user_input = input("\033[94mВы > \033[0m") - if user_input.lower() == 'exit': - print("Завершение сеанса.") - break + # Добавляем сообщение об ошибке в историю, чтобы модель могла исправиться + self.conversation_history.append({ + "role": "user", + "content": f"Your last response was not a valid JSON. Please correct your output to be a single JSON object. Error: {error_message}" + }) + continue # Переходим к следующей итерации, чтобы модель могла исправиться + + # 5. Извлекаем мысль и действие/ответ + thought = json_response.get("thought") + action = json_response.get("action") + answer = json_response.get("answer") + + logging.info(f"Thought: {thought}") + + if answer: + logging.info(f"Final Answer: {answer}") + # --- Guardrails Output Check --- + # guardrail_verdict = self.guardrail_manager.check_output(answer) + # if guardrail_verdict: + # final_answer = f"Output check failed: {guardrail_verdict}" + # logging.warning(final_answer) + # return final_answer - if not user_input.strip(): - continue - - # Используем execute_task для обработки ввода пользователя - print(f"\033[93m{self.name} >\033[0m", end="", flush=True) - final_response = self.execute_task(user_input) + return str(answer) # Ensure answer is a string + + if action: + # The model sometimes returns a string in the 'action' field when it should + # be providing a final 'answer'. We'll treat this as a final answer to + # make the system more robust. + if not isinstance(action, dict): + logging.warning(f"Model provided 'action' as a string instead of an object: '{action}'. Treating as Final Answer.") + return str(action) + + tool_name = action.get("name") + tool_input = action.get("input", {}) + + # The model sometimes tries to call 'answer' as a tool. This is incorrect. + # We'll catch this and treat it as the final answer. + if tool_name == 'answer': + logging.warning("Model incorrectly used 'answer' as a tool name. Treating as Final Answer.") + return json.dumps(tool_input) if isinstance(tool_input, dict) else str(tool_input) - # Печатаем финальный ответ, который execute_task вернул - # execute_task уже логирует промежуточные шаги - print(final_response) - - except (KeyboardInterrupt, EOFError): - print("\nЗавершение сеанса.") - break + if tool_name and isinstance(tool_input, dict): + logging.info(f"Action: {tool_name}({tool_input})") + try: + # 6. Выполняем инструмент + tool_result = self._execute_tool(tool_name, tool_input) + observation = f"Tool {tool_name} executed successfully. Result:\n{tool_result}" + except ToolExecutionError as e: + observation = str(e) # Используем сообщение из кастомного исключения + + logging.info(f"Observation: {observation}") + + # 7. Добавляем результат в историю для следующего шага + self.conversation_history.append({"role": "user", "content": f"Observation: {observation}"}) + else: + logging.warning("Invalid action format in model response.") + self.conversation_history.append({"role": "user", "content": "Observation: Invalid action format. Please provide a valid 'name' and 'input' for the action."}) + else: + logging.warning("Model did not provide an 'action' or 'answer'.") + # Просим модель предоставить либо действие, либо финальный ответ + self.conversation_history.append({"role": "user", "content": "Observation: You must provide either an 'action' or a final 'answer'."}) + + final_answer = f"Agent {self.name} reached max iterations ({self.max_iterations}) without a final answer." + logging.warning(final_answer) + return final_answer diff --git a/app/agents/roles/coding_agent.py b/app/agents/roles/coding_agent.py index f1a69ef..25179d8 100644 --- a/app/agents/roles/coding_agent.py +++ b/app/agents/roles/coding_agent.py @@ -3,8 +3,9 @@ from app.agents.agent import Agent from app.agents.tools import ( read_file_tool, read_file_tool_def, - edit_file_tool, edit_file_tool_def, + write_to_file_tool, write_to_file_tool_def, list_files_tool, list_files_tool_def, + run_tests_tool, run_tests_tool_def ) @@ -15,9 +16,10 @@ def __init__(self, name: str, role: str, goal: str, **kwargs): # Добавляем специфику роли в цель, чтобы она попала в основной промпт refined_goal = ( f"{goal}\n\n" - "IMPORTANT: Your primary goal is to make precise, targeted changes (surgical edits). " - "Do not rewrite entire files. Instead, identify the specific function, method, " - "or block of code that needs changing and use the 'edit_file_tool' to replace only that part." + "IMPORTANT: Your primary goal is to write high-quality, efficient, and clean Python code. " + "You must follow the provided coding standards. " + "When you need to modify a file, read its content first, then provide the full, complete, updated content to the 'write_to_file_tool'. " + "This tool will overwrite the entire file with your new content." ) super().__init__( @@ -28,8 +30,9 @@ def __init__(self, name: str, role: str, goal: str, **kwargs): ) # Self-register tools self.add_tool(read_file_tool, read_file_tool_def) - self.add_tool(edit_file_tool, edit_file_tool_def) self.add_tool(list_files_tool, list_files_tool_def) + self.add_tool(write_to_file_tool, write_to_file_tool_def) + self.add_tool(run_tests_tool, run_tests_tool_def) # Старый system_prompt и _create_initial_messages больше не нужны, # так как вся логика теперь в базовом классе Agent. \ No newline at end of file diff --git a/app/agents/roles/standard_roles.py b/app/agents/roles/standard_roles.py index 723e462..5a6e6aa 100644 --- a/app/agents/roles/standard_roles.py +++ b/app/agents/roles/standard_roles.py @@ -1,43 +1,60 @@ -""" -This module defines the standard roles for specialized agents in the system. -""" +"""This module defines the standard agent roles and their configurations.""" -# Configuration for an agent specialized in file system operations +# Default configuration for a general-purpose agent +DEFAULT_AGENT = { + "name": "DefaultAgent", + "role": "A helpful and versatile AI assistant.", + "goal": "Fulfill the user's request to the best of your ability." +} + +# Base configurations with specific tools for each agent role. +# The factory will use the 'tools' list to equip each agent. + +# Configuration for a file system expert agent FILESYSTEM_EXPERT = { "name": "FileSystemExpert", - "role": "An expert in browsing and reading files on a local file system.", - "goal": "To help users understand the project structure by listing and reading files.", - "tools": ["list_files", "read_file", "save_memory"] + "role": "An expert in interacting with the local file system.", + "goal": "Manage files and directories, such as creating, reading, and writing files.", + "tools": ["read_file", "list_files", "write_to_file", "update_file"], } -# Configuration for an agent specialized in web searching +# Configuration for a web search expert agent WEB_SEARCH_EXPERT = { "name": "WebSearchExpert", - "role": "An expert in searching the web for real-time information.", - "goal": "To find the most relevant and up-to-date information online in response to a user's query.", - "tools": ["web_search", "save_memory"] + "role": "An expert in finding information on the web.", + "goal": "Answer questions and provide information by searching the web.", + "tools": ["web_search"], } -# Add other specialized agent configurations here as needed. -# For example, a CodeWriterAgent, a DatabaseExpert, etc. +# Configuration for a coding agent +CODING_AGENT = { + "name": "CodingAgent", + "role": "A professional coder who writes high-quality, efficient, and clean Python code.", + "goal": "Write high-quality, efficient, and clean Python code according to provided standards.", + "tools": ["read_file", "list_files", "write_to_file", "run_tests", "update_file"], +} -ALL_ROLES = { - "FileSystemExpert": FILESYSTEM_EXPERT, - "WebSearchExpert": WEB_SEARCH_EXPERT, +# Configuration for a reviewer agent +REVIEWER_AGENT = { + "name": "ReviewerAgent", + "role": "A meticulous reviewer who ensures code quality, adherence to standards, and correctness.", + "goal": "Ensure code quality, adherence to standards, and correctness.", + "tools": ["read_file"], } -# Configuration for the planner agent -PLANNER_AGENT = { - "name": "PlannerAgent", - "role": "A master planner who specializes in breaking down complex goals into a sequence of actionable steps for a team of specialized agents.", - "goal": "To create a clear, step-by-step JSON plan that efficiently leads to the user's desired outcome.", - "tools": [] # The planner does not use tools, it only thinks. +# Configuration for a testing agent +TESTING_AGENT = { + "name": "TestingAgent", + "role": "Software Quality Assurance Engineer", + "goal": "Thoroughly test code to find bugs and ensure reliability.", + "tools": ["run_tests", "read_file"], } -# Configuration for the evaluator agent -EVALUATOR_AGENT = { - "name": "EvaluatorAgent", - "role": "A meticulous evaluator who analyzes multiple execution plans and selects the most optimal one.", - "goal": "To choose the most efficient, logical, and safe plan from a given set of options.", - "tools": [] # The evaluator only thinks and chooses. +# A dictionary of all available agent roles that the Orchestrator can assign tasks to. +ALL_ROLES = { + "FileSystemExpert": FILESYSTEM_EXPERT, + "WebSearchExpert": WEB_SEARCH_EXPERT, + "CodingAgent": CODING_AGENT, + "ReviewerAgent": REVIEWER_AGENT, + "TestingAgent": TESTING_AGENT, } \ No newline at end of file diff --git a/app/agents/roles/task_decomposer.py b/app/agents/roles/task_decomposer.py index 867aa19..90215dc 100644 --- a/app/agents/roles/task_decomposer.py +++ b/app/agents/roles/task_decomposer.py @@ -19,18 +19,33 @@ def __init__(self, name: str, role: str, goal: str, **kwargs): ) self.system_prompt = """ You are an expert project manager. Your task is to break down a high-level user goal into a concise, step-by-step plan. -Each step must be a single, clear action assigned to one of the available roles. +Each step must be a single, clear action assigned to an appropriate agent. +Combine simple, related actions into a single, comprehensive step. For example, instead of one step to create a file and another to write to it, create a single step that does both. -# AVAILABLE ROLES: -- CodingAgent: Writes, modifies, and fixes code. -- TestingAgent: Runs tests and reports results. -- ReviewerAgent: Performs code reviews, checking for quality and adherence to standards. -- EvaluatorAgent: Analyzes test failures and creates bug reports. +# AVAILABLE ROLES & THEIR KEY TOOLS: +- **FileSystemExpert**: Works with files. Key tool: `write_to_file_tool(path, content)`. +- **CodingAgent**: Writes, modifies, and fixes Python code. Key tool: `write_to_file_tool(path, content)`. +- **TestingAgent**: Runs tests. Key tool: `run_tests_tool(path)`. +- **ReviewerAgent**: Performs code reviews. Key tool: `read_file_tool(path)`. # OUTPUT FORMAT: Your output MUST be a single JSON object with a single key "plan", which contains a list of steps. Do not include any other text, explanation, or markdown code fences. -# EXAMPLE: +# EXAMPLE 1: Simple file operation +Goal: "Create a file named 'hello.txt' and write 'Hello World' in it." + +Your output: +{ + "plan": [ + { + "step": 1, + "agent": "FileSystemExpert", + "task": "Create a new file 'hello.txt' with the content 'Hello World'." + } + ] +} + +# EXAMPLE 2: More complex coding task Goal: "Create a function to add two numbers and test it." Your output: @@ -38,27 +53,18 @@ def __init__(self, name: str, role: str, goal: str, **kwargs): "plan": [ { "step": 1, - "assignee": "CodingAgent", - "task": "Create a new function 'add(a, b)' in 'app/utils/math.py'", - "description": "Implement the core logic for the addition function." + "agent": "CodingAgent", + "task": "Create a new file 'app/utils/math.py' with an 'add(a, b)' function that returns the sum of two numbers." }, { "step": 2, - "assignee": "ReviewerAgent", - "task": "Review the 'add' function in 'app/utils/math.py'", - "description": "Ensure the code quality and correctness of the new function." + "agent": "CodingAgent", + "task": "Create a new test file 'tests/test_math.py' to test the 'add' function. Include at least one test case." }, { "step": 3, - "assignee": "CodingAgent", - "task": "Create a new test file 'tests/test_math.py' with tests for the 'add' function", - "description": "Write unit tests to verify the behavior of the 'add' function." - }, - { - "step": 4, - "assignee": "TestingAgent", - "task": "Run the tests in 'tests/test_math.py'", - "description": "Execute the newly created tests to confirm the function works as expected." + "agent": "TestingAgent", + "task": "Run the tests in 'tests/test_math.py'." } ] } diff --git a/app/agents/tools.py b/app/agents/tools.py index 8930a05..bc77c90 100644 --- a/app/agents/tools.py +++ b/app/agents/tools.py @@ -30,14 +30,13 @@ def read_file_tool(input_data: Dict[str, Any]) -> str: def list_files_tool(input_data: Dict[str, Any]) -> str: """ - Рекурсивно выводит дерево файлов и директорий по указанному пути, - игнорируя служебные файлы/директории. Помогает понять структуру проекта. - + Выводит список файлов и директорий по указанному пути. + Args: input_data (Dict[str, Any]): Словарь, который может содержать ключ 'path' с путем к директории. По умолчанию - текущая. Returns: - Отформатированное дерево файлов и директории в виде строки. + Отформатированное дерево файлов и директорий в виде строки. """ path = input_data.get("path", ".") if not os.path.isdir(path): @@ -58,66 +57,32 @@ def list_files_tool(input_data: Dict[str, Any]) -> str: return output.strip() -def edit_file_tool(input_data: Dict[str, Any]) -> str: +def write_to_file_tool(input_data: Dict[str, Any]) -> str: """ - Создает, перезаписывает, добавляет или заменяет контент в файле. - - 'overwrite': Полностью перезаписывает файл. - - 'append': Добавляет контент в конец файла. - - 'replace': Заменяет один фрагмент строки на другой. + Создает новый файл или полностью перезаписывает существующий. Args: input_data (Dict[str, Any]): Словарь, содержащий: 'path' (str): Путь к файлу. - 'mode' (str): Режим работы ('overwrite', 'append', 'replace'). - 'content' (str, optional): Содержимое для 'overwrite' или 'append'. - 'old_content' (str, optional): Исходный фрагмент для 'replace'. - 'new_content' (str, optional): Новый фрагмент для 'replace'. + 'content' (str): Содержимое для записи. """ path = input_data.get("path") - mode = input_data.get("mode", "overwrite") + content = input_data.get("content") - if not path: - return "Ошибка: Аргумент 'path' обязателен." - if mode not in ['overwrite', 'append', 'replace']: - return "Ошибка: Недопустимый режим. Используйте 'overwrite', 'append' или 'replace'." + if not path or content is None: + return "Ошибка: Аргументы 'path' и 'content' обязательны." try: + # Убедимся, что директория для файла существует directory = os.path.dirname(path) if directory: os.makedirs(directory, exist_ok=True) - - if mode == 'replace': - old_content = input_data.get("old_content") - new_content = input_data.get("new_content") - if old_content is None or new_content is None: - return "Ошибка: Для режима 'replace' необходимы 'old_content' и 'new_content'." - - with open(path, "r", encoding="utf-8") as f: - file_content = f.read() - - if old_content not in file_content: - return f"Ошибка: Исходный фрагмент 'old_content' не найден в файле '{path}'." - - file_content = file_content.replace(old_content, new_content, 1) - - with open(path, "w", encoding="utf-8") as f: - f.write(file_content) - return f"Файл '{path}' успешно обновлен в режиме 'replace'." - - else: # overwrite or append - content = input_data.get("content") - if content is None: - return f"Ошибка: Для режима '{mode}' обязателен аргумент 'content'." - - write_mode = "w" if mode == "overwrite" else "a" - with open(path, write_mode, encoding="utf-8") as f: - f.write(content) - return f"Файл '{path}' успешно обновлен в режиме '{mode}'." - - except FileNotFoundError: - return f"Ошибка: Файл не найден по пути '{path}'." + + with open(path, "w", encoding="utf-8") as f: + f.write(content) + return f"Файл '{path}' успешно создан/перезаписан." except Exception as e: - return f"Ошибка: Не удалось выполнить операцию с файлом '{path}': {e}" + return f"Ошибка: Не удалось записать в файл '{path}': {e}" def delete_file_tool(input_data: Dict[str, Any]) -> str: @@ -181,6 +146,30 @@ def run_tests_tool(input_data: Dict[str, Any]) -> str: return f"Критическая ошибка: Не удалось запустить тесты. Причина: {e}" +def update_file_tool(input_data: Dict[str, Any]) -> str: + """ + Appends content to an existing file. If the file doesn't exist, it creates it. + + Args: + input_data (Dict[str, Any]): A dictionary containing: + 'path' (str): The path to the file. + 'content' (str): The content to append to the file. + """ + path = input_data.get("path") + content = input_data.get("content") + + if not path or content is None: + return "Error: 'path' and 'content' are required arguments." + + try: + # 'a' mode appends to the file, and creates it if it doesn't exist. + with open(path, "a", encoding="utf-8") as f: + f.write("\n\n" + content) + return f"Content successfully appended to '{path}'." + except Exception as e: + return f"Error: Could not update file '{path}': {e}" + + # Определения инструментов (Tool Definitions) read_file_tool_def = { "type": "function", @@ -212,21 +201,18 @@ def run_tests_tool(input_data: Dict[str, Any]) -> str: }, } -edit_file_tool_def = { +write_to_file_tool_def = { "type": "function", "function": { - "name": "edit_file_tool", - "description": "Создает, перезаписывает, добавляет или заменяет контент в файле. Режимы: 'overwrite', 'append', 'replace'.", + "name": "write_to_file_tool", + "description": "Создает новый файл или полностью перезаписывает существующий указанным контентом.", "parameters": { "type": "object", "properties": { - "path": {"type": "string", "description": "Полный путь к файлу."}, - "mode": {"type": "string", "enum": ["overwrite", "append", "replace"], "description": "Режим записи."}, - "content": {"type": "string", "description": "Содержимое для 'overwrite' или 'append'."}, - "old_content": {"type": "string", "description": "Исходный фрагмент для 'replace'."}, - "new_content": {"type": "string", "description": "Новый фрагмент для 'replace'."}, + "path": {"type": "string", "description": "Полный путь к файлу (включая имя файла)."}, + "content": {"type": "string", "description": "Полное содержимое для записи в файл."}, }, - "required": ["path", "mode"], + "required": ["path", "content"], }, }, } @@ -260,3 +246,25 @@ def run_tests_tool(input_data: Dict[str, Any]) -> str: }, }, } + +update_file_tool_def = { + "type": "function", + "function": { + "name": "update_file_tool", + "description": "Appends content to an existing file. Creates the file if it does not exist.", + "parameters": { + "type": "object", + "properties": { + "path": { + "type": "string", + "description": "The path to the file to be updated." + }, + "content": { + "type": "string", + "description": "The content to append to the file." + } + }, + "required": ["path", "content"] + } + } +} diff --git a/app/factory/agent_factory.py b/app/factory/agent_factory.py index 36a148c..5479ad3 100644 --- a/app/factory/agent_factory.py +++ b/app/factory/agent_factory.py @@ -1,75 +1,31 @@ """ -Factory for creating agent teams from configuration. +Factory for creating specialized agents. """ -import logging -import yaml -import os -from importlib import import_module -from typing import Dict, Tuple, Any, List +from typing import Dict, Any from app.agents.agent import Agent -from app.agents.roles.task_decomposer import TaskDecomposer -from app.agents.tools import read_file_tool, read_file_tool_def, list_files_tool, list_files_tool_def +from app.agents.tools import ( + read_file_tool, read_file_tool_def, + list_files_tool, list_files_tool_def, + write_to_file_tool, write_to_file_tool_def, + update_file_tool, update_file_tool_def, + run_tests_tool, run_tests_tool_def +) from app.agents.web_search_tool import web_search_tool, web_search_tool_def from app.agents.memory_tool import save_memory_tool, save_memory_tool_def # A mapping of tool names to their functions and definitions +# This makes it easy to add tools to agents based on their config. AVAILABLE_TOOLS = { "read_file": (read_file_tool, read_file_tool_def), "list_files": (list_files_tool, list_files_tool_def), + "write_to_file": (write_to_file_tool, write_to_file_tool_def), + "update_file": (update_file_tool, update_file_tool_def), + "run_tests": (run_tests_tool, run_tests_tool_def), "web_search": (web_search_tool, web_search_tool_def), "save_memory": (save_memory_tool, save_memory_tool_def), } -def load_config(path: str) -> dict: - """Loads a YAML configuration file.""" - with open(path, 'r') as f: - return yaml.safe_load(f) - -def get_class_from_string(class_path: str): - """Dynamically imports a class from a string path.""" - module_path, class_name = class_path.rsplit('.', 1) - module = import_module(module_path) - return getattr(module, class_name) - -def create_agent_team(main_config_path: str) -> Tuple[TaskDecomposer, Dict[str, Agent]]: - """ - Creates and configures the agent team based on YAML files. - - Args: - main_config_path (str): The path to the main configuration file. - - Returns: - A tuple containing the TaskDecomposer and the dictionary of worker agents. - """ - logging.info("Loading configurations from %s...", main_config_path) - main_config = load_config(main_config_path) - - workers = {} - api_key = os.getenv("OPENAI_API_KEY") - common_kwargs = {"api_key": api_key, "model": main_config.get('default_model', 'o4-mini')} - - for agent_name, agent_info in main_config['agents'].items(): - logging.info(f"Initializing {agent_name}...") - - agent_config_path = agent_info['config_path'] - agent_config = load_config(agent_config_path) - - agent_class = get_class_from_string(agent_info['_target_']) - - init_params = agent_config.copy() - init_params.update(common_kwargs) - - agent_instance = agent_class(**init_params) - - # The logic for adding tools has been moved to the agent classes themselves. - # The factory is now cleaner and only responsible for instantiation. - - workers[agent_name] = agent_instance - - task_decomposer = workers.pop("TaskDecomposer") - return task_decomposer, workers - class AgentFactory: """ A factory class for creating different types of specialized agents. @@ -84,8 +40,7 @@ def create_agent( Creates an agent based on a configuration dictionary. Args: - agent_config: A dictionary containing the agent's name, role, goal, - and a list of tool names it should have. + agent_config: A dictionary containing the agent's name, role, and goal. api_key: The OpenAI API key. model: The name of the model to use. @@ -97,6 +52,7 @@ def create_agent( goal = agent_config.get("goal", "To complete tasks efficiently.") tool_names = agent_config.get("tools", []) + # Create a base agent agent = Agent( name=name, role=role, @@ -105,13 +61,16 @@ def create_agent( model=model ) - # Clear default tools and add only the specified ones - agent.tools = {} - agent.tool_definitions = [] - + # Add only the tools specified in the agent's configuration for tool_name in tool_names: if tool_name in AVAILABLE_TOOLS: tool_func, tool_def = AVAILABLE_TOOLS[tool_name] agent.add_tool(tool_func, tool_def) + else: + # This could be a configuration error, but we'll just log it for now. + print(f"Warning: Tool '{tool_name}' not found for agent '{name}'.") + + # The 'save_memory' tool is essential for all agents to learn. + agent.add_tool(save_memory_tool, save_memory_tool_def) return agent \ No newline at end of file diff --git a/app/orchestration/orchestrator.py b/app/orchestration/orchestrator.py index f50a655..01e7230 100644 --- a/app/orchestration/orchestrator.py +++ b/app/orchestration/orchestrator.py @@ -1,6 +1,6 @@ """ -This module defines the Orchestrator, the central brain of the multi-agent team. -It creates a plan and manages its execution by delegating tasks to specialized agents. +This module defines the Orchestrator. +It is the central brain that creates a plan and executes it by delegating tasks to agents. """ import json import logging @@ -8,68 +8,74 @@ from openai import OpenAI from app.factory.agent_factory import AgentFactory -from app.agents.roles.standard_roles import ALL_ROLES, PLANNER_AGENT, EVALUATOR_AGENT +from app.agents.roles.standard_roles import ALL_ROLES -# Updated prompt to ask for multiple plans (Tree-of-Thoughts) +# A new, simplified, and much more direct planner prompt. +# This prompt is designed to generate a single, efficient plan. PLANNER_PROMPT_TEMPLATE = """ -You are a master planner for a team of AI agents. Your job is to create THREE DISTINCT step-by-step plans to accomplish the user's goal. +You are an expert project planner and a senior software architect. Your job is to create a robust, production-ready, step-by-step plan to accomplish the user's goal. **User's Goal:** "{user_goal}" -**Available Team of Specialists:** -{agents_description} - -Based on the goal, create a JSON object with a key "plans", containing a list of THREE different plans. Each plan is a JSON array of tasks. Each task must have: -- `step`: An integer for the step number (e.g., 1, 2, 3). -- `agent`: The name of the single most appropriate agent from the available team. -- `task`: A clear and specific instruction for the agent. - -**Important Rules:** -- The three plans should represent different strategies to achieve the goal. -- Your entire response MUST be a single, valid JSON object like: `{{"plans": [[...plan1...], [...plan2...], [...plan3...]]}}` - -**Example:** +**Project Structure & Quality Standards:** +- **Tool Directory**: All new tools MUST be appended to the existing file `app/agents/tools.py`. Do NOT create new files for tools. +- **Test Directory**: All new tests for tools MUST be appended to the existing file `tests/agents/test_tools.py`. Do NOT create new files for tests. +- **Code Quality**: All generated Python functions MUST include detailed Google-style docstrings explaining their purpose, arguments, and return values. +- **Test Quality**: Tests MUST be written using `pytest`. Each feature MUST have multiple tests covering standard cases, edge cases (e.g., empty strings, null inputs, different data types), and potential failure modes. +- **Mandatory Review**: Any plan that involves writing or modifying code MUST include a final step where the `ReviewerAgent` inspects the newly created files. + +**Available Team of Specialists & Their Exact Responsibilities:** +- **CodingAgent**: Writes and modifies Python code. Can ONLY use `write_to_file_tool`, `read_file_tool`, and `list_files_tool`. Assign ALL code and test writing/modification tasks to this agent. +- **TestingAgent**: Runs tests using `pytest`. Can ONLY use `run_tests_tool` and `read_file_tool`. It CANNOT write or modify files. Assign ONLY test execution tasks to this agent. +- **ReviewerAgent**: Performs code reviews. Can ONLY use `read_file_tool`. It CANNOT write, modify, or test code. Assign ONLY file review tasks to this agent. +- **FileSystemExpert**: Handles generic file operations. Can ONLY use `write_to_file_tool`, `read_file_tool`, and `list_files_tool`. + +**Your Task:** +Create a JSON object with a key "plan" containing a list of tasks. Each task must have: +- `step`: An integer for the step number. +- `agent`: The name of the single most appropriate agent from the team based on their exact responsibilities. +- `task`: A clear and specific instruction for that agent, including full file paths and complete, high-quality code with docstrings. + +**CRITICAL RULES:** +1. **ADHERE TO STANDARDS:** Your plan MUST follow all project structure and quality standards defined above. +2. **CORRECT ASSIGNMENT:** Assign tasks ONLY to agents that have the tools and responsibilities to complete them. +3. **BE PRECISE:** The task description must contain all necessary information, including full code content. +4. **JSON ONLY:** Your entire response MUST be a single, valid JSON object. + +**Example of a HIGH-QUALITY Plan:** +Goal: "Create a Python tool to reverse a string." + +Your output: {{ - "plans": [ - [ - {{"step": 1, "agent": "WebSearchExpert", "task": "Find official pytest docs."}} - ], - [ - {{"step": 1, "agent": "FileSystemExpert", "task": "Check existing test files for pytest usage examples."}} - ], - [ - {{"step": 1, "agent": "WebSearchExpert", "task": "Search for tutorials on how to use pytest with FastAPI."}} - ] + "plan": [ + {{ + "step": 1, + "agent": "CodingAgent", + "task": "Read the content of 'app/agents/tools.py'. If the 'reverse_string_tool' function does not already exist, append the following code to the end of the file: \\n\\n# ... (existing code) ...\\n\\ndef reverse_string_tool(input_data: dict) -> str:\\n '''Reverses a string.'''\\n return input_data.get('text', '')[::-1]\\n\\nreverse_string_tool_def = {{...}} # (full tool definition here)" + }}, + {{ + "step": 2, + "agent": "CodingAgent", + "task": "Read the content of 'tests/agents/test_tools.py'. If tests for 'reverse_string_tool' do not already exist, append new tests to the end of the file." + }}, + {{ + "step": 3, + "agent": "TestingAgent", + "task": "Run all tests in 'tests/test_tools.py' to verify correctness of all tools, including the newly added one." + }}, + {{ + "step": 4, + "agent": "ReviewerAgent", + "task": "Review the final code in 'app/agents/tools.py' and 'tests/agents/test_tools.py' for quality and correctness." + }} ] }} """ -EVALUATOR_PROMPT_TEMPLATE = """ -You are a meticulous and rational Evaluator. Your task is to analyze a list of proposed plans and select the single best one to achieve the user's goal. - -**User's Goal:** -"{user_goal}" - -**Proposed Plans:** -{plans_json_string} - -**Evaluation Criteria:** -1. **Efficiency:** Which plan is likely to achieve the goal in the fewest steps? -2. **Robustness:** Which plan is least likely to fail or run into errors? -3. **Clarity:** Which plan is the most logical and straightforward? - -Based on your analysis, respond with a JSON object containing the index (starting from 0) of the best plan. - -**Example:** -{{ - "best_plan_index": 1 -}} -""" - class Orchestrator: """ - Creates multiple plans, evaluates them, and manages the execution of the best one. + Creates a single, efficient plan and manages its execution. """ def __init__(self, api_key: str, model: str = "gpt-4o-mini"): self.client = OpenAI(api_key=api_key) @@ -78,23 +84,11 @@ def __init__(self, api_key: str, model: str = "gpt-4o-mini"): self.api_key = api_key self.execution_history: List[Dict[str, Any]] = [] - def _get_agents_description(self) -> str: - """Creates a formatted string describing the available worker agents.""" - descriptions = [] - for name, config in ALL_ROLES.items(): - descriptions.append(f"- Agent: {name}\n - Role: {config['role']}\n - Best for: {config['goal']}") - return "\n".join(descriptions) - - def _create_plan(self, user_goal: str) -> List[List[Dict[str, Any]]]: - """ - Uses the PlannerAgent's logic to create multiple distinct plans. - """ - logging.info(f"Orchestrator is creating multiple plans for the goal: '{user_goal}'") - agents_description = self._get_agents_description() - prompt = PLANNER_PROMPT_TEMPLATE.format( - user_goal=user_goal, - agents_description=agents_description - ) + def _create_plan(self, user_goal: str) -> List[Dict[str, Any]]: + """Creates a single, direct plan to achieve the user's goal.""" + logging.info(f"Orchestrator is creating a plan for the goal: '{user_goal}'") + + prompt = PLANNER_PROMPT_TEMPLATE.format(user_goal=user_goal) try: response = self.client.chat.completions.create( @@ -102,91 +96,29 @@ def _create_plan(self, user_goal: str) -> List[List[Dict[str, Any]]]: messages=[{"role": "user", "content": prompt}], response_format={"type": "json_object"} ) - # We expect a dict with a "plans" key, which is a list of lists - plan_variants = json.loads(response.choices[0].message.content or "{}").get("plans", []) - logging.info(f"Planner proposed {len(plan_variants)} plans.") - return plan_variants + content = response.choices[0].message.content or "{}" + plan = json.loads(content).get("plan", []) + logging.info("Planner proposed a plan with %s steps.", len(plan)) + return plan except Exception as e: - logging.error(f"Failed to create plans: {e}", exc_info=True) + logging.error(f"Failed to create a valid plan: {e}", exc_info=True) return [] - def _evaluate_and_select_plan(self, user_goal: str, plans: List[List[Dict[str, Any]]]) -> List[Dict[str, Any]]: - """Uses the Evaluator's logic to select the best plan.""" - if not plans: - return [] - if len(plans) == 1: - logging.info("Only one plan was generated, selecting it by default.") - return plans[0] - - logging.info("Evaluating plans to select the best one...") - prompt = EVALUATOR_PROMPT_TEMPLATE.format( - user_goal=user_goal, - plans_json_string=json.dumps(plans, indent=2) - ) - - try: - response = self.client.chat.completions.create( - model=self.model, - messages=[{"role": "user", "content": prompt}], - response_format={"type": "json_object"} - ) - choice = json.loads(response.choices[0].message.content or "{}") - best_plan_index = choice.get("best_plan_index", 0) - - if 0 <= best_plan_index < len(plans): - logging.info(f"Evaluator selected plan #{best_plan_index + 1}.") - return plans[best_plan_index] - else: - logging.warning("Evaluator returned an invalid index, defaulting to the first plan.") - return plans[0] - except Exception as e: - logging.error(f"Failed to evaluate plans: {e}. Defaulting to the first plan.", exc_info=True) - return plans[0] - - def _create_briefing(self, goal: str, plan: List[Dict[str, Any]], current_task: Dict[str, Any]) -> str: - """Creates a detailed context (briefing) for an agent.""" - history_log = "" - if not self.execution_history: - history_log = "This is the first step. No execution history yet." - else: - history_log = "Here are the results from the previous steps:\n" - for record in self.execution_history: - history_log += ( - f"Step {record['step']} ({record['agent']}) completed.\n" - f"Task: {record['task']}\n" - f"Result: {record['result']}\n\n" - ) - - return ( - f"**Overall Goal:** {goal}\n\n" - f"**Full Plan:** {json.dumps(plan, indent=2)}\n\n" - f"**Execution History:**\n{history_log}\n" - f"-----------------------------------\n" - f"**Your Current Task (Step {current_task['step']}):**\n" - f"Your task is: \"{current_task['task']}\".\n" - f"Analyze the goal, plan, and history, then execute your task to produce the required result." - ) - def run(self, user_goal: str) -> str: """ - Runs the full orchestration process: generate multiple plans, evaluate them, and execute the best one. + Runs the full orchestration process: create a plan and execute it. """ self.execution_history = [] - # 1. Create multiple plan variants - plan_variants = self._create_plan(user_goal) - if not plan_variants: - return "I'm sorry, I couldn't create any plans to address your request. Please try rephrasing it." - - # 2. Evaluate and select the best plan - best_plan = self._evaluate_and_select_plan(user_goal, plan_variants) - if not best_plan: - return "I'm sorry, I couldn't select a valid plan to execute. Please try again." + # 1. Create a single, efficient plan + plan = self._create_plan(user_goal) + if not plan: + return "I'm sorry, I couldn't create a plan to address your request. Please try rephrasing it." - print(f"\033[95mOrchestrator's Selected Plan:\033[0m\n{json.dumps(best_plan, indent=2)}") + print(f"\033[95mOrchestrator's Plan:\033[0m\n{json.dumps(plan, indent=2, ensure_ascii=False)}") - # 3. Execute the selected plan step-by-step - for task in best_plan: + # 2. Execute the plan step-by-step + for task in plan: agent_name = task.get("agent") task_description = task.get("task") @@ -196,17 +128,24 @@ def run(self, user_goal: str) -> str: logging.info(f"--- Executing Step {task['step']}: {task_description} (Agent: {agent_name}) ---") - # Create the specialist agent agent_config = ALL_ROLES[agent_name] specialist_agent = self.agent_factory.create_agent(agent_config, self.api_key, self.model) - # Create the briefing for the agent - briefing = self._create_briefing(user_goal, best_plan, task) - - # Execute the task + # This briefing is now lean and focused. It does NOT contain the overall goal. + briefing = "" + if self.execution_history: + briefing += ( + "Here's a summary of what has been done so far:\n" + + "\n".join([f"- {record['result']}" for record in self.execution_history]) + ) + briefing += ( + f"\nYour specific, immediate task is: '{task_description}'.\n" + "Focus ONLY on completing this single task. Do not move on to other tasks. " + "Once you have completed your task, provide a 'Final Answer' with a summary of what you did." + ) + result = specialist_agent.execute_task(briefing) - # Save the result to history self.execution_history.append({ "step": task['step'], "agent": agent_name, @@ -216,6 +155,5 @@ def run(self, user_goal: str) -> str: logging.info(f"--- Step {task['step']} Result: {result} ---") - # Return the result of the final step final_result = self.execution_history[-1]["result"] if self.execution_history else "The plan was executed, but there is no final result." return final_result \ No newline at end of file diff --git a/db/chunks.json b/db/chunks.json index cef79a3..8848241 100644 --- a/db/chunks.json +++ b/db/chunks.json @@ -1,6 +1,6 @@ [ { - "text": "Testing Guidelines\n1. Test Structure: Arrange-Act-Assert (AAA)\nAll tests should follow the AAA pattern for clarity and readability. Arrange: Prepare all necessary data and mocks. Act: Call the function or method being tested. Assert: Check that the result meets expectations. ```python def test_user_creation(): # Arrange user_data = {\"username\": \"test\", \"email\": \"test@example.com\"} mock_db = MagicMock()\n# Act\ncreated_user = create_user(db=mock_db, data=user_data)", + "text": "Testing Guidelines\n1. Test Structure: Arrange-Act-Assert (AAA)\nAll tests should follow the AAA pattern for clarity and readability.\nArrange: Prepare all necessary data and mocks.\nAct: Call the function or method being tested.\nAssert: Check that the result meets expectations.\n```python def test_user_creation(): # Arrange user_data = {\"username\": \"test\", \"email\": \"test@example.com\"} mock_db = MagicMock()\n# Act\ncreated_user = create_user(db=mock_db, data=user_data)", "source": "testing_guidelines.md", "metadata": { "tags": [], @@ -9,7 +9,7 @@ } }, { - "text": "# Assert\nassert created_user.username == user_data[\"username\"]\nmock_db.add.assert_called_once()\n```\n2. Test Naming\nTest function names should be descriptive and start with test_. Follow the format test___. Example: test_add_items_with_negative_quantity_raises_error()\n3. Use pytest.raises for Exception Testing\nTo verify that code correctly raises exceptions, use the pytest.raises context manager. ```python import pytest\ndef test_divide_by_zero_raises_exception(): with pytest.raises(ZeroDivisionError): divide(10, 0) ```", + "text": "# Assert\nassert created_user.username == user_data[\"username\"]\nmock_db.add.assert_called_once()\n```\n2. Test Naming\nTest function names should be descriptive and start with test_. Follow the format test___.\nExample: test_add_items_with_negative_quantity_raises_error()\n3. Use pytest.raises for Exception Testing\nTo verify that code correctly raises exceptions, use the pytest.raises context manager.\n```python import pytest\ndef test_divide_by_zero_raises_exception(): with pytest.raises(ZeroDivisionError): divide(10, 0) ```", "source": "testing_guidelines.md", "metadata": { "tags": [], @@ -18,7 +18,7 @@ } }, { - "text": "Error Handling Principles\n1. Prefer Specific Exceptions\nAlways catch the most specific exception type possible. Avoid using except Exception: unless absolutely necessary. Bad: python try: # some code except Exception as e: log.error(\"An error occurred\")\nGood: python try: # some code except FileNotFoundError as e: log.error(f\"File not found: {e}\") except (KeyError, ValueError) as e: log.warning(f\"Data error: {e}\")\n2. Use Custom Exceptions\nFor errors specific to your application's domain logic, create your own exception classes.", + "text": "Error Handling Principles\n1. Prefer Specific Exceptions\nAlways catch the most specific exception type possible. Avoid using except Exception: unless absolutely necessary.\nBad: python try: # some code except Exception as e: log.error(\"An error occurred\")\nGood: python try: # some code except FileNotFoundError as e: log.error(f\"File not found: {e}\") except (KeyError, ValueError) as e: log.warning(f\"Data error: {e}\")\n2. Use Custom Exceptions", "source": "error_handling.md", "metadata": { "tags": [], @@ -27,7 +27,7 @@ } }, { - "text": "This makes the code more readable and allows calling code to handle specific failures precisely. ```python class InsufficientBalanceError(Exception): \"\"\"Exception raised when the account balance is too low.\"\"\" pass\ndef withdraw(amount): if amount > current_balance: raise InsufficientBalanceError(\"Insufficient funds in the account\") ```\n3. Log Errors Correctly\nWhen catching an exception, be sure to log the full information, including the stack trace, to simplify debugging. ```python import logging\ntry: # ... except Exception as e: logging.error(\"An unexpected error occurred\", exc_info=True)", + "text": "For errors specific to your application's domain logic, create your own exception classes. This makes the code more readable and allows calling code to handle specific failures precisely.\n```python class InsufficientBalanceError(Exception): \"\"\"Exception raised when the account balance is too low.\"\"\" pass\ndef withdraw(amount): if amount > current_balance: raise InsufficientBalanceError(\"Insufficient funds in the account\") ```\n3. Log Errors Correctly\nWhen catching an exception, be sure to log the full information, including the stack trace, to simplify debugging.\n```python import logging", "source": "error_handling.md", "metadata": { "tags": [], @@ -36,43 +36,43 @@ } }, { - "text": "API Design Principles\nResources: Use plural nouns for endpoint naming (e.g., /users, /products). HTTP Methods: Use the correct HTTP verbs for actions:\nGET for retrieving data. POST for creating new resources. PUT / PATCH for updating. DELETE for deleting. Versioning: Include the API version in the URL (e.g., /api/v1/users).", - "source": "api_design.md", + "text": "try: # ... except Exception as e: logging.error(\"An unexpected error occurred\", exc_info=True)", + "source": "error_handling.md", "metadata": { "tags": [], - "full_path": "knowledge/api_design.md", - "chunk_id": "api_design.md_0" + "full_path": "knowledge/error_handling.md", + "chunk_id": "error_handling.md_2" } }, { - "text": "Python Style Guide\nNaming: Use snake_case for variables and functions. Class names should use CamelCase. Constants should be in UPPER_SNAKE_CASE. Line Length: The maximum line length is 99 characters. Docstrings: All public modules, functions, classes, and methods must have Google-style docstrings. Imports: Group imports in the following order: standard library, third-party libraries, local application. String Formatting\nf-strings: Always prefer f-strings for formatting instead of str.format() or the % operator.", - "source": "python_style_guide.md", + "text": "API Design Principles\nResources: Use plural nouns for endpoint naming (e.g., /users, /products).\nHTTP Methods: Use the correct HTTP verbs for actions:\nGET for retrieving data.\nPOST for creating new resources.\nPUT / PATCH for updating.\nDELETE for deleting.\nVersioning: Include the API version in the URL (e.g., /api/v1/users).", + "source": "api_design.md", "metadata": { "tags": [], - "full_path": "knowledge/python_style_guide.md", - "chunk_id": "python_style_guide.md_0" + "full_path": "knowledge/api_design.md", + "chunk_id": "api_design.md_0" } }, { - "text": "Good: user_info = f\"User {user.name} with ID {user.id}\"\nBad: user_info = \"User {} with ID {}\".format(user.name, user.id)\nList Comprehensions\nSimplicity: Use list comprehensions to create lists from existing iterables, but only if the logic remains simple and readable. If complex logic or multiple nested loops are required, use a regular for loop.", + "text": "Python Style Guide\nNaming: Use snake_case for variables and functions. Class names should use CamelCase. Constants should be in UPPER_SNAKE_CASE.\nLine Length: The maximum line length is 99 characters.\nDocstrings: All public modules, functions, classes, and methods must have Google-style docstrings.\nImports: Group imports in the following order: standard library, third-party libraries, local application.\nString Formatting\nf-strings: Always prefer f-strings for formatting instead of str.format() or the % operator.\nGood: user_info = f\"User {user.name} with ID {user.id}\"", "source": "python_style_guide.md", "metadata": { "tags": [], "full_path": "knowledge/python_style_guide.md", - "chunk_id": "python_style_guide.md_1" + "chunk_id": "python_style_guide.md_0" } }, { - "text": "Good: squares = [x*x for x in range(10)]\nAvoid (hard to read): complex_list = [x + y for x in range(10) for y in range(5) if x % 2 == 0 if y % 3 == 0]", + "text": "Bad: user_info = \"User {} with ID {}\".format(user.name, user.id)\nList Comprehensions\nSimplicity: Use list comprehensions to create lists from existing iterables, but only if the logic remains simple and readable. If complex logic or multiple nested loops are required, use a regular for loop.\nGood: squares = [x*x for x in range(10)]\nAvoid (hard to read): complex_list = [x + y for x in range(10) for y in range(5) if x % 2 == 0 if y % 3 == 0]", "source": "python_style_guide.md", "metadata": { "tags": [], "full_path": "knowledge/python_style_guide.md", - "chunk_id": "python_style_guide.md_2" + "chunk_id": "python_style_guide.md_1" } }, { - "text": "Agent Tool Creation Guide\nThis document outlines the best practices for creating new tool functions that can be used by AI agents in our system. 1. Tool Function Structure\nEvery tool must be a wrapper around the core logic and accept a single argument of type Dict[str, Any]. This ensures a unified interface for all tools. ```python\nCorrect\ndef my_tool(input_data: Dict[str, Any]) -> str: #... Incorrect\ndef my_tool(param1: str, param2: int) -> str: #... ```\n2. Mandatory Error Handling\nA tool should never crash with an unhandled exception.", + "text": "Agent Tool Creation Guide\nThis document outlines the best practices for creating new tool functions that can be used by AI agents in our system.\n1. Tool Function Structure\nEvery tool must be a wrapper around the core logic and accept a single argument of type Dict[str, Any]. This ensures a unified interface for all tools.\n```python\nCorrect\ndef my_tool(input_data: Dict[str, Any]) -> str: #...\nIncorrect\ndef my_tool(param1: str, param2: int) -> str: #... ```\n2. Mandatory Error Handling", "source": "tool_creation_guide.md", "metadata": { "tags": [ @@ -83,7 +83,7 @@ } }, { - "text": "Always use try-except and return an informative error message as a string. python def substring_tool(input_data: Dict[str, Any]) -> str: try: text = input_data['text'] start = input_data['start'] # ... core logic ... return result except KeyError as e: return f\"Error: Missing required key {e} in input_data.\" except Exception as e: return f\"An unexpected error occurred: {e}\"\n3. Detailed Docstrings\nAlways write detailed docstrings in Google-style. Describe the function's purpose, all keys in the input_data dictionary, and the return value. 4.", + "text": "A tool should never crash with an unhandled exception. Always use try-except and return an informative error message as a string.\npython def substring_tool(input_data: Dict[str, Any]) -> str: try: text = input_data['text'] start = input_data['start'] # ... core logic ... return result except KeyError as e: return f\"Error: Missing required key {e} in input_data.\" except Exception as e: return f\"An unexpected error occurred: {e}\"\n3. Detailed Docstrings", "source": "tool_creation_guide.md", "metadata": { "tags": [ @@ -94,7 +94,7 @@ } }, { - "text": "Tool Definition (_tool_def)\nEvery tool must have a corresponding _tool_def definition. This is a dictionary that describes the function's signature for the OpenAI API, allowing the agent to understand how to call your tool.", + "text": "Always write detailed docstrings in Google-style. Describe the function's purpose, all keys in the input_data dictionary, and the return value.\n4. Tool Definition (_tool_def)\nEvery tool must have a corresponding _tool_def definition. This is a dictionary that describes the function's signature for the OpenAI API, allowing the agent to understand how to call your tool.", "source": "tool_creation_guide.md", "metadata": { "tags": [ diff --git a/db/embeddings.npy b/db/embeddings.npy index d192fff..d3dae64 100644 Binary files a/db/embeddings.npy and b/db/embeddings.npy differ diff --git a/db/memory.db b/db/memory.db index 551fff0..e4164da 100644 Binary files a/db/memory.db and b/db/memory.db differ diff --git a/main.py b/main.py index 0335b92..ab3df99 100644 --- a/main.py +++ b/main.py @@ -28,8 +28,8 @@ def main(): model = "gpt-4o-mini" - print("--- Multi-Agent System Initialized ---") - print("Enter your request. Type 'exit' to close.") + logging.info("--- Multi-Agent System Initialized ---") + logging.info("Enter your request. Type 'exit' to close.") orchestrator = Orchestrator(api_key=api_key, model=model) @@ -37,7 +37,7 @@ def main(): while True: user_query = input("\033[94mYou > \033[0m") if user_query.lower() in ['exit', 'quit']: - print("Shutting down...") + logging.info("Shutting down...") break if not user_query.strip(): @@ -46,10 +46,10 @@ def main(): # The orchestrator handles the entire process final_result = orchestrator.run(user_query) - print(f"\033[92mFinal Answer >\033[0m {final_result}") + logging.info(f"\033[92mFinal Answer >\033[0m {final_result}") except (KeyboardInterrupt, EOFError): - print("\nShutting down...") + logging.error("\nShutting down...") if __name__ == "__main__": main() \ No newline at end of file diff --git a/scripts/build_knowledge_base.py b/scripts/build_knowledge_base.py index acf4692..19c09b8 100644 --- a/scripts/build_knowledge_base.py +++ b/scripts/build_knowledge_base.py @@ -14,7 +14,7 @@ from openai import OpenAI from unstructured.partition.md import partition_md from rank_bm25 import BM25Okapi -from semchunk.chunker import SemanticChunker +import semchunk # --- Configuration --- load_dotenv() @@ -38,18 +38,10 @@ raise ValueError("OPENAI_API_KEY is not set in the environment variables.") client = OpenAI(api_key=api_key) +# Initialize the semantic chunker from isaacus-dev/semchunk +# This chunker works with token counts, not embedding models. tokenizer = tiktoken.get_encoding("cl100k_base") -# Initialize the semantic chunker -semantic_chunker = SemanticChunker( - embed_model=client, - model_name=EMBEDDING_MODEL, - max_chunk_size=TEXT_CHUNK_MAX_TOKENS, - # The 'breakpoint_percentile_threshold' is a key parameter to tune. - # It determines how different two sentences must be to create a split. - # Lower value = more splits, higher value = fewer splits. - # Let's start with a value that often works well. - breakpoint_percentile_threshold=90 -) +chunker = semchunk.chunkerify(tokenizer, TEXT_CHUNK_MAX_TOKENS) def load_and_partition_documents(directory: Path) -> List[Dict[str, Any]]: """ @@ -90,7 +82,7 @@ def semantic_chunk_document(doc: Dict[str, Any]) -> List[Dict[str, Any]]: logging.info(f"Semantically chunking '{doc['source']}'...") try: # The chunker returns a list of text strings - chunk_texts = semantic_chunker.chunk(doc["text"]) + chunk_texts = chunker(doc["text"]) chunks_with_metadata = [] for i, chunk_text in enumerate(chunk_texts): diff --git a/temp_test_run_data.json b/temp_test_run_data.json deleted file mode 100644 index c19e7ec..0000000 --- a/temp_test_run_data.json +++ /dev/null @@ -1 +0,0 @@ -{"dict_test_cases": {}, "testCases": [], "metricScores": [], "configurations": {}} \ No newline at end of file diff --git a/tests/test_tools.py b/tests/test_tools.py deleted file mode 100644 index e69de29..0000000