diff --git a/main.py b/main.py index 00d383d..1bb8c0a 100644 --- a/main.py +++ b/main.py @@ -1,36 +1,10 @@ -""" -Deep Agent — поисковый агент с виртуальной ФС на основе deepagents. - -Агент умеет: - 1. Искать информацию в интернете (DuckDuckGo, без API-ключа) - 2. Создавать и редактировать файлы в виртуальной ФС через write_file / edit_file - 3. По завершении работы выгружает все виртуальные файлы в реальную ФС (./output/) - -Использование: - python main.py - python main.py "что такое LangGraph" -""" - -import asyncio import os -import shutil -import sys -from pathlib import Path - -from dotenv import load_dotenv -from langchain.tools import tool -from langchain_core.messages import HumanMessage +import json from langchain_openai import ChatOpenAI +from langchain.tools import tool +from duckduckgo_search import DDGS -from deepagents import create_deep_agent -from deepagents.backends import CompositeBackend, FilesystemBackend, LocalShellBackend - -load_dotenv() - -# --------------------------------------------------------------------------- -# LLM -# --------------------------------------------------------------------------- - +# LLM configuration llm = ChatOpenAI( model="openai/gpt-oss-20b:free", base_url="https://platform.brojs.ru/jrnl-bh/api/inference/v1", @@ -38,149 +12,83 @@ llm = ChatOpenAI( temperature=0.5, ) -# --------------------------------------------------------------------------- -# Workspace dirs -# --------------------------------------------------------------------------- +# Simple virtual file system +class VirtualFileSystem: + def __init__(self): + self.files = {} -VIRTUAL_ROOT = Path("./workspace") -REAL_OUT = Path("./output") -VIRTUAL_ROOT.mkdir(parents=True, exist_ok=True) -REAL_OUT.mkdir(parents=True, exist_ok=True) + def write(self, path: str, content: str): + self.files[path] = content -# --------------------------------------------------------------------------- -# Tools -# --------------------------------------------------------------------------- + def list(self): + return list(self.files.keys()) + def dump_to_real_fs(self, base_dir: str = "."): + for path, content in self.files.items(): + full_path = os.path.join(base_dir, path) + os.makedirs(os.path.dirname(full_path), exist_ok=True) + with open(full_path, "w", encoding="utf-8") as f: + f.write(content) +# Tool for web search using duckduckgo @tool def web_search(query: str) -> str: - """Ищет информацию в интернете и возвращает краткие сниппеты. - - Args: - query: поисковый запрос на любом языке - """ + """Search the web for a query and return top 3 results as a formatted string.""" try: - from duckduckgo_search import DDGS with DDGS() as ddgs: - results = list(ddgs.text(query, max_results=7)) + results = list(ddgs.text(query, max_results=3)) if not results: - return "Ничего не найдено." - lines = [] - for r in results: - lines.append(f"### {r.get('title', '')}") - lines.append(r.get("body", "")) - lines.append(f"URL: {r.get('href', '')}") - lines.append("") - return " -".join(lines) - except Exception as exc: - return f"Ошибка поиска: {exc}" + return "No results found." + formatted = [] + for i, r in enumerate(results, 1): + formatted.append(f"{i}. {r['title']}\n{r['body']}\nURL: {r['href']}") + return "\n\n".join(formatted) + except Exception as e: + return f"Search error: {e}" +# Deep agent implementation +class DeepAgent: + def __init__(self, llm, tools): + self.llm = llm + self.tools = {t.__name__: t for t in tools} + self.vfs = VirtualFileSystem() -@tool -def export_virtual_files() -> str: - """Копирует все файлы из виртуальной ФС (workspace/) в реальную папку output/. - - Вызывай этот инструмент в самом конце, после того как все нужные файлы созданы. - """ - try: - copied = [] - for src in VIRTUAL_ROOT.rglob("*"): - if src.is_file(): - rel = src.relative_to(VIRTUAL_ROOT) - dst = REAL_OUT / rel - dst.parent.mkdir(parents=True, exist_ok=True) - shutil.copy2(src, dst) - copied.append(str(rel)) - if copied: - return f"Выгружено {len(copied)} файл(ов) в {REAL_OUT}:\n" + "\n".join(f" {f}" for f in copied) - return f"Виртуальная ФС пуста — нет файлов для выгрузки." - except Exception as exc: - return f"Ошибка экспорта: {exc}" - - -# --------------------------------------------------------------------------- -# Backend — виртуальная ФС поверх workspace/ -# --------------------------------------------------------------------------- - -_shell_backend = LocalShellBackend( - root_dir=str(VIRTUAL_ROOT), - virtual_mode=True, - inherit_env=True, -) - -_fs_backend = FilesystemBackend( - root_dir=str(VIRTUAL_ROOT), - virtual_mode=True, -) - -backend = CompositeBackend( - default=_shell_backend, - routes={"/fs/": _fs_backend}, -) - -# --------------------------------------------------------------------------- -# Agent -# --------------------------------------------------------------------------- - -SYSTEM_PROMPT = """\ -Ты — исследовательский агент с виртуальной файловой системой. - -## Доступные инструменты -- web_search(query) — ищет информацию в интернете, возвращает сниппеты -- write_file(path, content) — создаёт/перезаписывает файл в виртуальной ФС -- edit_file(path, old, new) — редактирует файл -- read_file(path) — читает файл -- ls(path) — список файлов -- export_virtual_files() — копирует все файлы из виртуальной ФС в ./output/ - -## Порядок работы -1. Найди информацию по запросу через web_search -2. Обработай и сохрани результаты в один или несколько файлов через write_file -3. Проверь сохранённые файлы через read_file -4. В самом конце вызови export_virtual_files() чтобы выгрузить файлы в реальную ФС - -## Требования к файлам -- Сохраняй информацию в структурированном виде (markdown, JSON или plain text) -- Имена файлов — осмысленные (например results.md, summary.txt) -- Обязательно вызови export_virtual_files() в конце -""" - -agent = create_deep_agent( - model=llm, - tools=[web_search, export_virtual_files], - backend=backend, - system_prompt=SYSTEM_PROMPT, -) - -# --------------------------------------------------------------------------- -# Main -# --------------------------------------------------------------------------- - - -async def run(query: str) -> None: - print(f"Запрос: {query}\n") - config = {"configurable": {"thread_id": f"search-{hash(query) & 0xFFFF:04x}"}} - result = await agent.ainvoke( - {"messages": [HumanMessage(content=query)]}, - config, - ) - last = result["messages"][-1] - print("\n=== Ответ агента ===") - print(getattr(last, "content", str(last))) - print("\n=== Файлы в output/ ===") - out_files = list(REAL_OUT.rglob("*")) - if out_files: - for f in out_files: - if f.is_file(): - print(f" {f.relative_to(REAL_OUT)} ({f.stat().st_size} байт)") - else: - print(" (нет файлов)") - + def run(self, prompt: str): + # Initial system message + system_msg = "You are a helpful assistant that can search the web and create virtual files." + # Build conversation + messages = [system_msg, prompt] + # Simple loop: ask for tool usage + while True: + # Ask LLM for next action + response = self.llm.invoke(messages) + text = response.content.strip() + if text.lower().startswith("search:"): + query = text[7:].strip() + result = web_search(query) + self.vfs.write(f"search_{query.replace(' ', '_')}.txt", result) + messages.append(f"Search result for '{query}' written to virtual file.") + elif text.lower().startswith("create file:"): + parts = text[12:].strip().split("::", 1) + if len(parts) == 2: + path, content = parts + self.vfs.write(path.strip(), content.strip()) + messages.append(f"File '{path.strip()}' created.") + else: + messages.append("Invalid create file syntax. Use 'create file: path::content'.") + elif text.lower() == "done": + break + else: + messages.append(text) + # Dump virtual files to real filesystem + self.vfs.dump_to_real_fs() + return "Agent finished. Files written to disk." if __name__ == "__main__": - query = " ".join(sys.argv[1:]) if len(sys.argv) > 1 else ( - "Найди информацию о фреймворке LangGraph: что это, основные концепции, " - "примеры использования. Сохрани результаты в файл langgraph_research.md" + agent = DeepAgent(llm, tools=[web_search]) + # Example usage: search for LangChain and create a file + user_prompt = ( + "Search for 'LangChain deep agent' and create a file named 'summary.txt' with the first search result." ) - asyncio.run(run(query)) + result = agent.run(user_prompt) + print(result)