8. Самописный поисковый агент на основе deep agents from scratch: добавлен solution.py
This commit is contained in:
lonpatovaadelina committed
1 parent
8a2d31be37
commit
07c67470ed
1 file changed
+199
-1
+199
-1
@@ -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[\"ouLine truncated
|
||||
# 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: <name>\nPARAMS: <json>"""
|
||||
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\": \"<URL>\",\n \"filename\": \"<имя файла>\"\n}\n"
|
||||
"3) write_file: создать файл с заданным содержимым. Параметры: {\n \"filename\": \"<имя файла>\",\n \"content\": \"<текст>\"\n}\n"
|
||||
"\n\nТекущие файлы в виртуальной файловой системе: {files}\n\nЗадача: {task}\n\nОтвет в формате: ACTION: <действие>\nPARAMS: <json>"
|
||||
)
|
||||
|
||||
agent = DeepAgent(prompt_template=prompt_template, vfs=vfs)
|
||||
|
||||
# Пример задачи: найти последние новости о Python и сохранить первую найденную статью
|
||||
task_description = "Найти последние новости о Python и сохранить первую найденную статью в файл news.html"
|
||||
agent.run(task_description)
|
||||
|
||||
# Выгружаем виртуальные файлы в реальную файловую систему
|
||||
vfs.dump_to_disk()
|
||||
Reference in new issue
Block a user