8. Самописный поисковый агент на основе deep agents from scratch: solution.py

solutions/69de7223f309a98be0007e09_8__Самописный_поисковый_агент_на_основе_/solution.py

@@ -0,0 +1 @@
+{"files":[{"filename":"solution.py","content":"# solution.py\n# Полный код самописного поискового агента на основе LangChain\n# Агент умеет:\n#   1) Поискать информацию в интернете (через DuckDuckGo API)\n#   2) Создавать виртуальные файлы в памяти\n#   3) В конце работы выгружать все виртуальные файлы в реальную файловую систему\n\nimport os\nimport json\nimport re\nimport sys\nfrom typing import Dict, Any, List\n\n# LangChain и связанные библиотеки\nfrom langchain import OpenAI, LLMChain, PromptTemplate\nfrom langchain.agents import Tool, AgentExecutor, ZeroShotAgent\nfrom langchain.tools import BaseTool\nfrom langchain.callbacks.manager import CallbackManager\nfrom langchain.schema import AgentAction, AgentFinish\n\n# Для поиска в интернете будем использовать DuckDuckGo (без API ключа)\nimport requests\n\n# =====================\n# 1. Инструмент поиска\n# =====================\nclass DuckDuckGoSearchTool(BaseTool):\n    \"\"\"Инструмент для поиска в интернете через DuckDuckGo.\"\"\"\n\n    name: str = \"DuckDuckGoSearch\"\n    description: str = (\n        \"Используйте этот инструмент для поиска информации в интернете. \"\n        \"Вход: строка запроса. Выход: строка с результатами поиска.\"\n    )\n\n    def _run(self, query: str) -> str:\n        # Простой запрос к DuckDuckGo API (без ключа)\n        url = \"https://api.duckduckgo.com/\"\n        params = {\n            \"q\": query,\n            \"format\": \"json\",\n            \"no_redirect\": 1,\n            \"no_html\": 1,\n            \"skip_disambig\": 1,\n        }\n        try:\n            response = requests.get(url, params=params, timeout=10)\n            response.raise_for_status()\n            data = response.json()\n            # Собираем заголовки и описания из результатов\n            results = []\n            for i in range(1, 6):  # первые 5 результатов\n                title = data.get(f\"RelatedTopics[{i-1}].Text\")\n                if title:\n                    results.append(title)\n            if not results:\n                return \"Ничего не найдено.\"\n            return \"\\n\".join(results)\n        except Exception as e:\n            return f\"Ошибка при поиске: {e}\"\n\n    def _arun(self, query: str) -> str:  # pragma: no cover\n        raise NotImplementedError(\"Async не реализован\")\n\n# =====================\n# 2. Инструмент создания виртуального файла\n# =====================\nclass VirtualFileCreatorTool(BaseTool):\n    \"\"\"Создаёт виртуальный файл в памяти.\"\"\"\n\n    name: str = \"CreateVirtualFile\"\n    description: str = (\n        \"Создаёт виртуальный файл. Вход: JSON строка с полями 'filename' и 'content'. \"\n        \"Возвращает подтверждение.\"\n    )\n\n    def __init__(self, file_store: Dict[str, str]):\n        super().__init__()\n        self.file_store = file_store\n\n    def _run(self, input_str: str) -> str:\n        try:\n            data = json.loads(input_str)\n            filename = data[\"filename\"]\n            content = data[\"content\"]\n            self.file_store[filename] = content\n            return f\"Файл {filename} успешно создан в виртуальном хранилище.\"\n        except Exception as e:\n            return f\"Ошибка при создании файла: {e}\"\n\n    def _arun(self, input_str: str) -> str:  # pragma: no cover\n        raise NotImplementedError(\"Async не реализован\")\n\n# =====================\n# 3. Инструмент выгрузки файлов в реальную файловую систему\n# =====================\nclass DumpVirtualFilesTool(BaseTool):\n    \"\"\"Выгружает все виртуальные файлы в указанную директорию.\"\"\"\n\n    name: str = \"DumpVirtualFiles\"\n    description: str = (\n        \"Выгружает все виртуальные файлы в реальную файловую систему. \"\n        \"Вход: JSON строка с полем 'output_dir'. Возвращает список записанных файлов.\"\n    )\n\n    def __init__(self, file_store: Dict[str, str]):\n        super().__init__()\n        self.file_store = file_store\n\n    def _run(self, input_str: str) -> str:\n        try:\n            data = json.loads(input_str)\n            output_dir = data[\"ouLine truncated