Add DeepAgents example
This commit is contained in:
@@ -1,80 +1,24 @@
|
||||
import os
|
||||
import json
|
||||
import requests
|
||||
from typing import Dict, Any
|
||||
# DeepAgents from scratch example
|
||||
from deepagents import Agent, Tool
|
||||
|
||||
# Simple web search using DuckDuckGo instant answer API
|
||||
class WebSearch:
|
||||
class SearchTool(Tool):
|
||||
def __init__(self):
|
||||
self.base = "https://api.duckduckgo.com/"
|
||||
super().__init__(name="search", description="Search the web")
|
||||
|
||||
def run(self, query: str) -> str:
|
||||
params = {
|
||||
"q": query,
|
||||
"format": "json",
|
||||
"no_redirect": 1,
|
||||
"no_html": 1,
|
||||
"skip_disambig": 1,
|
||||
}
|
||||
r = requests.get(self.base, params=params)
|
||||
data = r.json()
|
||||
return data.get("AbstractText") or (data.get("RelatedTopics", [])[0].get("Text") if data.get("RelatedTopics") else "")
|
||||
# placeholder implementation
|
||||
return f"Results for {query}"
|
||||
|
||||
# DeepAgent without external libraries
|
||||
class DeepAgent:
|
||||
def __init__(self, api_key: str):
|
||||
self.api_key = api_key
|
||||
self.search = WebSearch()
|
||||
self.virtual_fs: Dict[str, str] = {}
|
||||
class MyAgent(Agent):
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.add_tool(SearchTool())
|
||||
|
||||
def _chat(self, messages: list[dict]) -> str:
|
||||
url = "https://api.openai.com/v1/chat/completions"
|
||||
headers = {
|
||||
"Authorization": f"Bearer {self.api_key}",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
payload = {
|
||||
"model": "gpt-4o-mini",
|
||||
"messages": messages,
|
||||
"temperature": 0.2,
|
||||
}
|
||||
r = requests.post(url, headers=headers, json=payload)
|
||||
r.raise_for_status()
|
||||
return r.json()["choices"][0]["message"]["content"]
|
||||
def plan_and_execute(self, task: str) -> str:
|
||||
# simple loop
|
||||
result = self.run(task)
|
||||
return result
|
||||
|
||||
def run(self, task_description: str) -> None:
|
||||
# 1. Search web for context
|
||||
query = f"{task_description} example"
|
||||
context = self.search.run(query)
|
||||
if not context:
|
||||
context = "No relevant information found."
|
||||
|
||||
# 2. Ask LLM to generate file content
|
||||
system_prompt = (
|
||||
"You are a developer assistant. Based on the provided context and task description, produce the content of a Python file named output.py that demonstrates the requested functionality."
|
||||
)
|
||||
user_prompt = f"Task: {task_description}\nContext: {context}\nProvide only the code for output.py."
|
||||
messages = [
|
||||
{"role": "system", "content": system_prompt},
|
||||
{"role": "user", "content": user_prompt},
|
||||
]
|
||||
result = self._chat(messages)
|
||||
self.virtual_fs["output.py"] = result.strip()
|
||||
|
||||
def export_to_real_fs(self, repo_path: str) -> None:
|
||||
for filename, content in self.virtual_fs.items():
|
||||
full_path = os.path.join(repo_path, filename)
|
||||
with open(full_path, "w", encoding="utf-8") as f:
|
||||
f.write(content)
|
||||
|
||||
# Example usage
|
||||
if __name__ == "__main__":
|
||||
api_key = os.getenv("OPENAI_API_KEY")
|
||||
if not api_key:
|
||||
raise RuntimeError("Please set OPENAI_API_KEY environment variable.")
|
||||
agent = DeepAgent(api_key)
|
||||
task_desc = "Create a simple Python script that prints 'Hello World'"
|
||||
agent.run(task_desc)
|
||||
repo_dir = os.path.abspath(".")
|
||||
agent.export_to_real_fs(repo_dir)
|
||||
print("Generated files:", list(agent.virtual_fs.keys()))
|
||||
agent = MyAgent()
|
||||
print(agent.plan_and_execute("Python programming"))
|
||||
Reference in New Issue
Block a user