8. Самописный поисковый агент на основе deep agents from scratch: solution.py
This commit is contained in:
+198
-79
@@ -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()
|
||||
Reference in New Issue
Block a user