Rewrite DeepAgent without external libs

This commit is contained in:
2026-05-28 13:21:11 +00:00
parent 2c2d72bc24
commit 0f41ec1a28
+26 -19
View File
@@ -1,41 +1,48 @@
""" """
DeepAgent implementation using deepagents from scratch. DeepAgent implementation from scratch.
The agent searches the web via Perplexity API and writes results to virtual files, The agent searches the web via Perplexity API, writes results to virtual files,
then exports them to real filesystem. and exports them to real filesystem.
""" """
import os import os
import shutil import shutil
from pathlib import Path from pathlib import Path
from typing import List, Dict import json
import requests
from deepagents import AgentBuilder, Tool, LLMWrapper
from deepagents.tools.perplexity import PerplexityTool
# Configuration
BASE_DIR = Path("/tmp/deepagent") BASE_DIR = Path("/tmp/deepagent")
OUTPUT_DIR = Path("./output_files") OUTPUT_DIR = Path("./output_files")
class DeepAgent: class DeepAgent:
def __init__(self, api_key: str): def __init__(self, api_key: str):
self.api_key = api_key self.api_key = api_key
# Build agent with Perplexity tool
perplexity_tool = Tool(
name="perplexity",
description="Search the web using Perplexity API.",
func=lambda query: PerplexityTool(api_key=api_key).search(query),
)
self.agent = AgentBuilder().with_tools([perplexity_tool]).build()
BASE_DIR.mkdir(parents=True, exist_ok=True) BASE_DIR.mkdir(parents=True, exist_ok=True)
OUTPUT_DIR.mkdir(parents=True, exist_ok=True) OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
def _search(self, query: str) -> dict:
url = "https://api.perplexity.ai/chat/completions"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
}
payload = {
"model": "perplexity-llama-3.1-mini-128k-chat",
"messages": [{"role": "user", "content": query}],
"temperature": 0,
}
resp = requests.post(url, headers=headers, json=payload)
resp.raise_for_status()
data = resp.json()
content = data["choices"][0]["message"]["content"]
# Expect JSON
return json.loads(content)
def search_and_save(self, query: str, filename: str): def search_and_save(self, query: str, filename: str):
# Run agent to get result result = self._search(query)
result = self.agent.run(query)
file_path = BASE_DIR / filename file_path = BASE_DIR / filename
with open(file_path, "w", encoding="utf-8") as f: with open(file_path, "w", encoding="utf-8") as f:
f.write(result) json.dump(result, f, ensure_ascii=False, indent=2)
return file_path return file_path
def export(self): def export(self):
@@ -50,6 +57,6 @@ if __name__ == "__main__":
agent = DeepAgent(api_key=args.api_key) agent = DeepAgent(api_key=args.api_key)
queries = ["Python async programming", "LangChain deep agents"] queries = ["Python async programming", "LangChain deep agents"]
for i, q in enumerate(queries, 1): for i, q in enumerate(queries, 1):
agent.search_and_save(q, f"result_{i}.txt") agent.search_and_save(q, f"result_{i}.json")
agent.export() agent.export()
print("Exported files to", OUTPUT_DIR) print("Exported files to", OUTPUT_DIR)