95 lines
3.4 KiB
Python
95 lines
3.4 KiB
Python
import os
|
|
import json
|
|
from langchain_openai import ChatOpenAI
|
|
from langchain.tools import tool
|
|
from duckduckgo_search import DDGS
|
|
|
|
# LLM configuration
|
|
llm = ChatOpenAI(
|
|
model="openai/gpt-oss-20b:free",
|
|
base_url="https://platform.brojs.ru/jrnl-bh/api/inference/v1",
|
|
api_key=os.getenv("JOURNAL_MCP_PAT"),
|
|
temperature=0.5,
|
|
)
|
|
|
|
# Simple virtual file system
|
|
class VirtualFileSystem:
|
|
def __init__(self):
|
|
self.files = {}
|
|
|
|
def write(self, path: str, content: str):
|
|
self.files[path] = content
|
|
|
|
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:
|
|
"""Search the web for a query and return top 3 results as a formatted string."""
|
|
try:
|
|
with DDGS() as ddgs:
|
|
results = list(ddgs.text(query, max_results=3))
|
|
if not results:
|
|
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()
|
|
|
|
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__":
|
|
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."
|
|
)
|
|
result = agent.run(user_prompt)
|
|
print(result)
|