Add utils

utils.py

@@ -0,0 +1,58 @@
+import os
+import json
+import requests
+from typing import List, Dict
+
+# Simple web search using Perplexity API (or fallback to DuckDuckGo)
+PERPLEXITY_API_URL = "https://api.perplexity.ai/chat/completions"
+API_KEY = os.getenv("PERPLEXITY_API_KEY")
+
+
+def web_search(query: str, max_results: int = 3) -> List[Dict[str, str]]:
+    """Return a list of search results with title and url.
+    Uses Perplexity API if key is set, otherwise DuckDuckGo.
+    """
+    if API_KEY:
+        headers = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}
+        payload = {
+            "model": "llama-3.1",
+            "messages": [{"role": "user", "content": f"Search the web for: {query}. Return top {max_results} results with title and url."}],
+            "temperature": 0,
+        }
+        resp = requests.post(PERPLEXITY_API_URL, headers=headers, json=payload)
+        resp.raise_for_status()
+        data = resp.json()
+        # Parse the assistant's content assuming JSON list
+        try:
+            results = json.loads(data["choices"][0]["message"]["content"])
+            return results[:max_results]
+        except Exception:
+            pass
+    # Fallback DuckDuckGo instant answer API (no key required)
+    ddg_url = f"https://api.duckduckgo.com/?q={query}&format=json"
+    r = requests.get(ddg_url)
+    r.raise_for_status()
+    data = r.json()
+    results = []
+    for link in data.get("RelatedTopics", [])[:max_results]:
+        if "Text" in link and "FirstURL" in link:
+            results.append({"title": link["Text"], "url": link["FirstURL"]})
+    return results
+
+
+def create_virtual_file(name: str, content: str) -> Dict[str, str]:
+    """Return a dict representing a virtual file."""
+    return {"name": name, "content": content}
+
+
+def export_files(files: List[Dict[str, str]], out_dir: str = "."):
+    os.makedirs(out_dir, exist_ok=True)
+    for f in files:
+        path = os.path.join(out_dir, f["name"])
+        with open(path, "w", encoding="utf-8") as fp:
+            fp.write(f["content"])
+
+# Simple logger
+
+def log(msg: str):
+    print("[DeepAgent]", msg)