diff --git a/solutions/69de7223f309a98be0007e09_8__Самописный_поисковый_агент_на_основе_/solution.py b/solutions/69de7223f309a98be0007e09_8__Самописный_поисковый_агент_на_основе_/solution.py index cb7f8cf..d0758ad 100644 --- a/solutions/69de7223f309a98be0007e09_8__Самописный_поисковый_агент_на_основе_/solution.py +++ b/solutions/69de7223f309a98be0007e09_8__Самописный_поисковый_агент_на_основе_/solution.py @@ -5,9 +5,7 @@ from typing import Dict, List import requests from bs4 import BeautifulSoup -from langchain.agents import Tool, AgentExecutor, initialize_agent, AgentType -from langchain.llms.openai import OpenAI -from langchain.schema import AgentAction, AgentFinish +from openai import OpenAI from rich.console import Console from rich.table import Table @@ -48,160 +46,102 @@ class VirtualFileSystem: # --------------------------------------------------------------------------- # -# Инструменты для агента +# Инструменты # --------------------------------------------------------------------------- # -def create_search_tool() -> Tool: + +def web_search(query: str) -> str: """ - Возвращает инструмент поиска в интернете. - Использует DuckDuckGo (без API‑ключа). + Выполняет поиск в интернете через DuckDuckGo и возвращает первые 3 результата. """ + url = "https://duckduckgo.com/html/" + params = {"q": query} + try: + resp = requests.get(url, params=params, timeout=10) + resp.raise_for_status() + except Exception as e: + return f"Ошибка при поиске: {e}" - def search(query: str) -> str: - """Ищет query и возвращает первые 3 ссылки с заголовками.""" - url = "https://duckduckgo.com/html/" - params = {"q": query} - try: - resp = requests.get(url, params=params, timeout=10) - resp.raise_for_status() - except Exception as e: - return f"Ошибка при поиске: {e}" - - soup = BeautifulSoup(resp.text, "html.parser") - results = [] - for a in soup.select("a.result__a")[:3]: - title = a.get_text(strip=True) - href = a["href"] - # DuckDuckGo иногда оборачивает ссылку в redirect - match = re.search(r"uddg=([^&]+)", href) - if match: - href = requests.utils.unquote(match.group(1)) - results.append(f"{title}\n{href}") - return "\n\n".join(results) or "Ничего не найдено." - - return Tool( - name="InternetSearch", - func=search, - description=( - "Используется для поиска информации в интернете. " - "Вход: строка запроса. Выход: список ссылок и заголовков." - ), - ) + soup = BeautifulSoup(resp.text, "html.parser") + results = [] + for a in soup.select("a.result__a")[:3]: + title = a.get_text(strip=True) + href = a["href"] + match = re.search(r"uddg=([^&]+)", href) + if match: + href = requests.utils.unquote(match.group(1)) + results.append(f"{title}\n{href}") + return "\n\n".join(results) or "Ничего не найдено." -def create_file_tool(vfs: VirtualFileSystem) -> Tool: +def create_file(vfs: VirtualFileSystem, filename: str, content: str) -> str: """ - Возвращает инструмент создания/обновления виртуального файла. + Создаёт или обновляет виртуальный файл. """ - - def write_file(args: str) -> str: - """ - Ожидает строку в формате: - filename=<имя> - content= - <текст> - - Пример: - filename=summary.txt - content= - Это пример содержания. - """ - try: - parts = args.split("content=", 1) - if len(parts) != 2: - return "Неверный формат. Используйте: filename=<имя>\ncontent=\n<текст>" - header, body = parts - filename_match = re.search(r"filename=([^\n]+)", header.strip()) - if not filename_match: - return "Не найдено имя файла (filename=...)." - filename = filename_match.group(1).strip() - content = body.strip() - vfs.write(filename, content) - return f"Файл '{filename}' успешно создан/обновлён." - except Exception as e: - return f"Ошибка при записи файла: {e}" - - return Tool( - name="CreateFile", - func=write_file, - description=( - "Создаёт или обновляет виртуальный файл. " - "Вход: строка с 'filename=<имя>' и 'content=' + текст." - ), - ) - - -def create_export_tool(vfs: VirtualFileSystem) -> Tool: - """ - Инструмент для выгрузки всех файлов в реальную файловую систему. - """ - - def export(args: str) -> str: - """Выгружает файлы в директорию, указанную в args (по умолчанию текущая).""" - directory = args.strip() or "." - try: - vfs.export_all(directory) - return f"Все файлы выгружены в '{os.path.abspath(directory)}'." - except Exception as e: - return f"Ошибка при экспорте файлов: {e}" - - return Tool( - name="ExportFiles", - func=export, - description=( - "Выгружает все виртуальные файлы на диск. " - "Вход: путь к директории (можно оставить пустым)." - ), - ) + vfs.write(filename, content) + return f"Файл '{filename}' успешно создан/обновлён." # --------------------------------------------------------------------------- # -# Основной агент +# Основной логика # --------------------------------------------------------------------------- # -def build_agent(vfs: VirtualFileSystem) -> AgentExecutor: - """ - Создаёт и возвращает готовый агент. - """ - llm = OpenAI(temperature=0, model_name="gpt-3.5-turbo") # или другой LLM - tools = [ - create_search_tool(), - create_file_tool(vfs), - create_export_tool(vfs), - ] - - agent_executor = initialize_agent( - tools, - llm, - agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION, - verbose=True, - handle_parsing_errors=True, +def run_agent(task: str, vfs: VirtualFileSystem) -> str: + """ + Выполняет задачу с помощью OpenAI LLM. + """ + client = OpenAI() + # Формируем запрос к модели + prompt = ( + f"Задача: {task}\n" + "Используйте web_search для получения информации.\n" + "Сформируйте краткое резюме (не более 300 слов) и сохраните его в файл summary.txt." ) - return agent_executor + # Получаем ответ модели + try: + response = client.chat.completions.create( + model="gpt-3.5-turbo", + temperature=0, + messages=[{"role": "user", "content": prompt}], + ) + except Exception as e: + return f"Ошибка при вызове OpenAI: {e}" + + summary_text = response.choices[0].message.content.strip() + # Сохраняем в виртуальный файл + create_file(vfs, "summary.txt", summary_text) + return summary_text + + +def flush_virtual_fs(vfs: VirtualFileSystem) -> str: + """ + Экспортирует все файлы из виртуальной файловой системы на диск. + """ + try: + vfs.export_all(".") + return f"Все файлы выгружены в '{os.path.abspath('.')}'." + except Exception as e: + return f"Ошибка при экспорте файлов: {e}" # --------------------------------------------------------------------------- # # Точка входа # --------------------------------------------------------------------------- # + def main(): console = Console() vfs = VirtualFileSystem() - # Пример задачи: собрать информацию о новых возможностях Python 3.11 и сохранить в файл. task = ( "Найди последние новости и обновления по версии Python 3.11, " "создай краткое резюме (не более 300 слов) и сохрани его в файле summary.txt." ) console.print("[bold cyan]Запуск агента...[/]") - agent = build_agent(vfs) - - # Запускаем агент - result = agent.run(task) + result = run_agent(task, vfs) console.print("\n[green]Итоговое сообщение от агента:[/]\n") console.print(result) - # Выводим список виртуальных файлов console.print("\n[bold magenta]Виртуальные файлы, созданные агентом:[/]") table = Table(show_header=True, header_style="bold blue") table.add_column("Имя файла", style="dim") @@ -209,8 +149,7 @@ def main(): table.add_row(fname) console.print(table) - # Экспортируем файлы в текущую директорию - export_result = agent.run("ExportFiles") + export_result = flush_virtual_fs(vfs) console.print("\n[green]Результат экспорта:[/]") console.print(export_result)