From 07c67470ed041e0a487cddf49f14e896f7147ba0 Mon Sep 17 00:00:00 2001 From: lonpatovaadelina Date: Mon, 25 May 2026 12:10:50 +0000 Subject: [PATCH] =?UTF-8?q?8.=20=D0=A1=D0=B0=D0=BC=D0=BE=D0=BF=D0=B8=D1=81?= =?UTF-8?q?=D0=BD=D1=8B=D0=B9=20=D0=BF=D0=BE=D0=B8=D1=81=D0=BA=D0=BE=D0=B2?= =?UTF-8?q?=D1=8B=D0=B9=20=D0=B0=D0=B3=D0=B5=D0=BD=D1=82=20=D0=BD=D0=B0=20?= =?UTF-8?q?=D0=BE=D1=81=D0=BD=D0=BE=D0=B2=D0=B5=20deep=20agents=20from=20s?= =?UTF-8?q?cratch:=20=D0=B4=D0=BE=D0=B1=D0=B0=D0=B2=D0=BB=D0=B5=D0=BD=20so?= =?UTF-8?q?lution.py?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../solution.py | 200 +++++++++++++++++- 1 file changed, 199 insertions(+), 1 deletion(-) diff --git a/solutions/69de7223f309a98be0007e09_8__Самописный_поисковый_агент_на_основе_/solution.py b/solutions/69de7223f309a98be0007e09_8__Самописный_поисковый_агент_на_основе_/solution.py index 9f384f6..bbbf015 100644 --- a/solutions/69de7223f309a98be0007e09_8__Самописный_поисковый_агент_на_основе_/solution.py +++ b/solutions/69de7223f309a98be0007e09_8__Самописный_поисковый_агент_на_основе_/solution.py @@ -1 +1,199 @@ -{"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[\"output_dir\"]\n os.makedirs(output_dir, exist_ok=True)\n written_files = []\n for filename, content in self.file_store.items():\n path = os.path.join(output_dir, filename)\n with open(path, \"w\", encoding=\"utf-8\") as f:\n f.write(content)\n written_files.append(path)\n return f\"Выгружено файлов: {', '.join(written_files)}\"\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# 4. Создание агента\n# =====================\nclass DeepAgent:\n \"\"\"Класс, реализующий deep agent.\"\"\"\n\n def __init__(self, llm: Any, output_dir: str = \"output_files\"):\n self.llm = llm\n self.output_dir = output_dir\n self.virtual_file_store: Dict[str, str] = {}\n self.tools = [\n DuckDuckGoSearchTool(),\n VirtualFileCreatorTool(self.virtual_file_store),\n DumpVirtualFilesTool(self.virtual_file_store),\n ]\n self.agent_executor = self._create_agent_executor()\n\n def _create_agent_executor(self) -> AgentExecutor:\n # Шаблон для агента\n prefix = (\n \"Вы - поисковый агент. Ваша задача: найти информацию в интернете, \"\n \"создать виртуальные файлы с результатами и в конце выгрузить их в реальную файловую систему. \"\n \"Вы можете использовать следующие инструменты:\"\n )\n suffix = \"Когда закончите, используйте DumpVirtualFiles, чтобы выгрузить файлы.\"\n format_instructions = \"\\n".join([f\"{tool.name}: {tool.description}\" for tool in self.tools])\n prompt = PromptTemplate(\n input_variables=[\"input\"],\n template=prefix + \"\\n\" + format_instructions + \"\\n\" + suffix + \"\\n\\nВход: {input}\\nОтвет:\"\n )\n llm_chain = LLMChain(llm=self.llm, prompt=prompt)\n agent = ZeroShotAgent(llm_chain=llm_chain, tools=self.tools, verbose=True)\n return AgentExecutor.from_agent_and_tools(agent=agent, tools=self.tools, verbose=True)\n\n def run(self, task: str) -> str:\n # Запускаем агента\n result = self.agent_executor.run(task)\n return result\n\n# =====================\n# 5. Точка входа\n# =====================\nif __name__ == \"__main__\":\n # Проверяем наличие ключа OpenAI\n if not os.getenv(\"OPENAI_API_KEY\"):\n print(\"Пожалуйста, установите переменную окружения OPENAI_API_KEY\")\n sys.exit(1)\n\n # Создаём LLM\n llm = OpenAI(temperature=0, model_name=\"gpt-3.5-turbo\")\n\n agent = DeepAgent(llm=llm, output_dir=\"output_files\")\n\n # Пример задачи: собрать информацию о \"Python asyncio\" и сохранить в файл\n task = (\n \"Найди информацию о Python asyncio, создай виртуальный файл \"\n \"asyncio_info.txt с найденными данными, и выгрузить файлы в директорию output_files.\"\n )\n print(\"Запуск агента...\")\n agent.run(task)\n print(\"Задача завершена.\")"} ,{"filename":"requirements.txt","content":"langchain\nopenai\nrequests\n"} ,{"filename":"README.md","content":"# Deep Agent – Самописный поисковый агент\n\n## Описание\n\nДанный проект реализует **deep agent** на основе LangChain, который умеет:\n\n1. Искать информацию в интернете (через DuckDuckGo API).\n2. Создавать виртуальные файлы в памяти.\n3. В конце работы выгружать все виртуальные файлы в реальную файловую систему.\n\n## Установка\n\n```bash\n# Клонируйте репозиторий\ngit clone \ncd <папка>\n\n# Создайте виртуальное окружение (рекомендуется)\npython -m venv venv\nsource venv/bin/activate # Windows: venv\\Scripts\\activate\n\n# Установите зависимости\npip install -r requirements.txt\n```\n\n## Настройка\n\nДля работы с LLM необходимо задать переменную окружения `OPENAI_API_KEY`:\n\n```bash\nexport OPENAI_API_KEY=sk-... # Linux/macOS\nset OPENAI_API_KEY=sk-... # Windows\n```\n\n## Запуск\n\n```bash\npython solution.py\n```\n\nПосле выполнения агент создаст директорию `output_files` и запишет туда виртуальные файлы.\n\n## Пример задачи\n\nВ примере `solution.py` задаётся задача собрать информацию о `Python asyncio` и сохранить её в файл `asyncio_info.txt`.\n\n## Как добавить новые инструменты\n\n1. Создайте класс, наследующий `BaseTool`.\n2. Реализуйте методы `_run` и `_arun`.\n3. Добавьте экземпляр инструмента в список `self.tools` в классе `DeepAgent`.\n4. Обновите шаблон подсказки, если нужно.\n\n---\n\n**Важно:**\n- Убедитесь, что ваш инструмент корректно обрабатывает входные данные.\n- Для больших файлов используйте потоковую запись, чтобы не загромождать память.\n"}],"main_file":"solution.py"} \ No newline at end of file +# solution.py +# -*- coding: utf-8 -*- +# Автор: ChatGPT +# Задача: реализовать простого deep‑agent, который может искать в интернете, создавать виртуальные файлы и в конце выгружать их в реальную файловую систему. + +import os +import json +import re +import requests +from pathlib import Path +from typing import Dict, List + +# Для работы с LLM будем использовать langchain, но чтобы не зависеть от конкретного провайдера, реализуем простую обёртку +# В реальном проекте можно заменить на OpenAI, Anthropic и т.д. +try: + from langchain.chat_models import ChatOpenAI + from langchain.schema import HumanMessage, AIMessage +except Exception as e: + # Если langchain не установлен, пользователь получит понятное сообщение + raise ImportError("Не удалось импортировать langchain. Установите пакет, выполнив 'pip install -r requirements.txt'") + +# ----------------------------- +# Конфигурация +# ----------------------------- +LLM_MODEL = "gpt-4o-mini" # можно заменить на любой доступный LLM +MAX_RETRIES = 3 +TIMEOUT = 10 # сек. + +# ----------------------------- +# Виртуальная файловая система +# ----------------------------- +class VirtualFileSystem: + """Простейшая виртуальная файловая система. + Хранит файлы в словаре {path: content}.""" + + def __init__(self): + self.files: Dict[str, str] = {} + + def write(self, path: str, content: str) -> None: + """Записывает файл в виртуальную файловую систему.""" + self.files[path] = content + + def read(self, path: str) -> str: + return self.files.get(path, "") + + def list_files(self) -> List[str]: + return list(self.files.keys()) + + def dump_to_disk(self, base_dir: str = "output") -> None: + """Выгружает все виртуальные файлы в реальную файловую систему.""" + base_path = Path(base_dir) + base_path.mkdir(parents=True, exist_ok=True) + for path, content in self.files.items(): + file_path = base_path / path + file_path.parent.mkdir(parents=True, exist_ok=True) + file_path.write_text(content, encoding="utf-8") + print(f"Виртуальные файлы выгружены в каталог '{base_path.resolve()}'") + +# ----------------------------- +# Инструменты для агента +# ----------------------------- +class WebSearchTool: + """Инструмент для поиска в интернете. Возвращает список URL.""" + + def __init__(self, query: str, max_results: int = 5): + self.query = query + self.max_results = max_results + + def run(self) -> List[str]: + # Для простоты используем DuckDuckGo Instant Answer API (без ключа) + url = "https://api.duckduckgo.com/" + params = { + "q": self.query, + "format": "json", + "no_redirect": 1, + "no_html": 1, + "skip_disambig": 1, + } + try: + resp = requests.get(url, params=params, timeout=TIMEOUT) + resp.raise_for_status() + data = resp.json() + results = [] + for link in data.get("RelatedTopics", []): + if isinstance(link, dict) and "FirstURL" in link: + results.append(link["FirstURL"]) + if len(results) >= self.max_results: + break + return results + except Exception as e: + print(f"Ошибка при поиске: {e}") + return [] + +class FetchPageTool: + """Инструмент для скачивания содержимого веб‑страницы.""" + + def __init__(self, url: str): + self.url = url + + def run(self) -> str: + try: + resp = requests.get(self.url, timeout=TIMEOUT) + resp.raise_for_status() + return resp.text + except Exception as e: + print(f"Ошибка при скачивании {self.url}: {e}") + return "" + +# ----------------------------- +# Основной агент +# ----------------------------- +class DeepAgent: + """Простой deep‑agent, использующий LLM для планирования действий.""" + + def __init__(self, prompt_template: str, vfs: VirtualFileSystem): + self.prompt_template = prompt_template + self.vfs = vfs + self.llm = ChatOpenAI(model=LLM_MODEL, temperature=0.2) + + def _extract_action(self, text: str) -> str: + """Извлекает действие из ответа LLM. Ожидается формат: ACTION: \nPARAMS: """ + match = re.search(r"ACTION:\s*(\w+)\s*PARAMS:\s*(\{.*\})", text, re.S) + if match: + action = match.group(1).strip() + params = json.loads(match.group(2)) + return action, params + else: + raise ValueError("Не удалось распознать действие из ответа LLM") + + def run(self, task: str) -> None: + """Запускает агента для выполнения задачи.""" + # Шаг 1: планирование действий + prompt = self.prompt_template.format(task=task, files=self.vfs.list_files()) + messages = [HumanMessage(content=prompt)] + for _ in range(MAX_RETRIES): + try: + response = self.llm(messages) + break + except Exception as e: + print(f"Ошибка LLM: {e}") + else: + print("Не удалось получить ответ от LLM.") + return + + # Шаг 2: обработка действий + try: + action, params = self._extract_action(response.content) + except Exception as e: + print(f"Ошибка при разборе ответа LLM: {e}") + return + + if action == "search": + query = params.get("query", "") + tool = WebSearchTool(query) + urls = tool.run() + self.vfs.write("search_results.json", json.dumps(urls, indent=2)) + print(f"Найдено {len(urls)} ссылок по запросу '{query}'.") + elif action == "fetch": + url = params.get("url", "") + tool = FetchPageTool(url) + content = tool.run() + filename = params.get("filename", "page.html") + self.vfs.write(filename, content) + print(f"Содержимое страницы {url} сохранено в {filename}.") + elif action == "write_file": + filename = params.get("filename", "output.txt") + content = params.get("content", "") + self.vfs.write(filename, content) + print(f"Файл {filename} создан в виртуальной файловой системе.") + else: + print(f"Неизвестное действие: {action}") + + # После выполнения одного действия агент завершает работу. + # Для более сложных сценариев можно добавить цикл с повторным вызовом LLM. + +# ----------------------------- +# Пример использования +# ----------------------------- +if __name__ == "__main__": + # Создаём виртуальную файловую систему + vfs = VirtualFileSystem() + + # Шаблон промпта для LLM + prompt_template = ( + "Вы агент, который может выполнять следующие действия:\n" + "1) search: выполнить поиск в интернете. Параметры: {\n \"query\": \"<запрос>\"\n}\n" + "2) fetch: скачать содержимое страницы. Параметры: {\n \"url\": \"\",\n \"filename\": \"<имя файла>\"\n}\n" + "3) write_file: создать файл с заданным содержимым. Параметры: {\n \"filename\": \"<имя файла>\",\n \"content\": \"<текст>\"\n}\n" + "\n\nТекущие файлы в виртуальной файловой системе: {files}\n\nЗадача: {task}\n\nОтвет в формате: ACTION: <действие>\nPARAMS: " + ) + + agent = DeepAgent(prompt_template=prompt_template, vfs=vfs) + + # Пример задачи: найти последние новости о Python и сохранить первую найденную статью + task_description = "Найти последние новости о Python и сохранить первую найденную статью в файл news.html" + agent.run(task_description) + + # Выгружаем виртуальные файлы в реальную файловую систему + vfs.dump_to_disk()