diff --git a/solutions/69de7223f309a98be0007e09_8__Самописный_поисковый_агент_на_основе_/solution.py b/solutions/69de7223f309a98be0007e09_8__Самописный_поисковый_агент_на_основе_/solution.py index 524f0f9..cb7f8cf 100644 --- a/solutions/69de7223f309a98be0007e09_8__Самописный_поисковый_агент_на_основе_/solution.py +++ b/solutions/69de7223f309a98be0007e09_8__Самописный_поисковый_агент_на_основе_/solution.py @@ -1,100 +1,219 @@ -<|channel|>final solution.py<|message|>import os -from typing import List +import os +import json +import re +from typing import Dict, List -# Библиотеки для работы с языковой моделью и векторным хранилищем -from langchain.embeddings.openai import OpenAIEmbeddings -from langchain.vectorstores.qdrant import Qdrant -from langchain.llms import OpenAI -from langchain.chains import RetrievalQA - -# Библиотека для вывода красивых сообщений в консоль +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 rich.console import Console from rich.table import Table -console = Console() -def init_vector_store(qdrant_url: str, collection_name: str): +# --------------------------------------------------------------------------- # +# Виртуальная файловая система +# --------------------------------------------------------------------------- # +class VirtualFileSystem: """ - Инициализирует подключение к Qdrant и возвращает объект vectorstore. - Если коллекция не существует – создаётся новая с использованием OpenAIEmbeddings. + Хранит файлы в памяти. Позволяет создавать и экспортировать их. """ - embeddings = OpenAIEmbeddings() - try: - # Попытка получить существующую коллекцию - vector_store = Qdrant( - client=None, - collection_name=collection_name, - embedding_function=embeddings.embed_query, - url=qdrant_url, - ) - except Exception: - # Если коллекция отсутствует – создаём её - vector_store = Qdrant.from_texts( - texts=[], - embeddings=embeddings, - client=None, - collection_name=collection_name, - url=qdrant_url, - ) - return vector_store -def build_agent(vector_store): + def __init__(self): + self.files: Dict[str, str] = {} + + def write(self, filename: str, content: str) -> None: + """Создаёт или перезаписывает файл.""" + self.files[filename] = content + + def read(self, filename: str) -> str | None: + """Возвращает содержимое файла, если он существует.""" + return self.files.get(filename) + + def list_files(self) -> List[str]: + """Список всех файлов.""" + return list(self.files.keys()) + + def export_all(self, directory: str = ".") -> None: + """ + Записывает все файлы в указанную директорию. + Если файл с таким именем уже существует – он будет перезаписан. + """ + os.makedirs(directory, exist_ok=True) + for name, content in self.files.items(): + path = os.path.join(directory, name) + with open(path, "w", encoding="utf-8") as f: + f.write(content) + + +# --------------------------------------------------------------------------- # +# Инструменты для агента +# --------------------------------------------------------------------------- # +def create_search_tool() -> Tool: """ - Создаёт цепочку RetrievalQA, которая использует заданный векторный store - и модель OpenAI для генерации ответов. + Возвращает инструмент поиска в интернете. + Использует DuckDuckGo (без API‑ключа). """ - llm = OpenAI(temperature=0.2) - qa_chain = RetrievalQA.from_chain_type( - llm=llm, - chain_type="stuff", - retriever=vector_store.as_retriever(search_kwargs={"k": 5}), + + 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=( + "Используется для поиска информации в интернете. " + "Вход: строка запроса. Выход: список ссылок и заголовков." + ), ) - return qa_chain -def add_documents(vector_store, documents: List[str]): - """ - Добавляет новые документы в коллекцию Qdrant. - """ - vector_store.add_texts(documents) -def query_agent(agent, question: str): +def create_file_tool(vfs: VirtualFileSystem) -> Tool: """ - Выполняет запрос к агенту и выводит результат в красивом формате. + Возвращает инструмент создания/обновления виртуального файла. """ - answer = agent.run(question) - table = Table(title="Ответ агента") - table.add_column("Вопрос", style="cyan") - table.add_column("Ответ", style="magenta") - table.add_row(question, answer) + + 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=( + "Выгружает все виртуальные файлы на диск. " + "Вход: путь к директории (можно оставить пустым)." + ), + ) + + +# --------------------------------------------------------------------------- # +# Основной агент +# --------------------------------------------------------------------------- # +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, + ) + return agent_executor + + +# --------------------------------------------------------------------------- # +# Точка входа +# --------------------------------------------------------------------------- # +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) + + 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") + for fname in vfs.list_files(): + table.add_row(fname) console.print(table) -def main(): - # Конфигурация - QDRANT_URL = os.getenv("QDRANT_URL", "http://localhost:6333") - COLLECTION_NAME = os.getenv("COLLECTION_NAME", "deep_agent_collection") + # Экспортируем файлы в текущую директорию + export_result = agent.run("ExportFiles") + console.print("\n[green]Результат экспорта:[/]") + console.print(export_result) - # Инициализация хранилища и агента - vector_store = init_vector_store(QDRANT_URL, COLLECTION_NAME) - agent = build_agent(vector_store) - - # Пример добавления документов (можно заменить на чтение из файлов) - sample_docs = [ - "Python – это язык программирования высокого уровня.", - "FastAPI позволяет быстро создавать веб‑приложения.", - "LangChain упрощает работу с LLM и векторными хранилищами." - ] - add_documents(vector_store, sample_docs) - - # Пример взаимодействия - console.print("[bold green]Deep Agent готов к работе![/]") - while True: - try: - question = console.input("\nВведите ваш вопрос (или 'выход' для завершения): ") - if question.lower() in ("выход", "exit", "quit"): - break - query_agent(agent, question) - except KeyboardInterrupt: - break if __name__ == "__main__": main() \ No newline at end of file