142 lines
5.1 KiB
Python
142 lines
5.1 KiB
Python
"""
|
|
Lab 2.8: Multi-Tool Agent -- SOLUTION
|
|
|
|
Agent with file operations AND web search capabilities.
|
|
Uses Anthropic API with three tools, or mock for offline execution.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '_shared'))
|
|
try:
|
|
from mock_llm import MockAnthropic as Anthropic
|
|
client = Anthropic()
|
|
IS_MOCK = True
|
|
except ImportError:
|
|
from anthropic import Anthropic
|
|
client = Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
|
|
IS_MOCK = False
|
|
|
|
MAX_ITERATIONS = 15
|
|
|
|
TOOLS = [
|
|
{
|
|
"name": "read_file",
|
|
"description": "Read the contents of a file at the given path",
|
|
"input_schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"path": {"type": "string", "description": "Path to the file"},
|
|
"reasoning": {"type": "string", "description": "Why are you reading this?"}
|
|
},
|
|
"required": ["path", "reasoning"]
|
|
}
|
|
},
|
|
{
|
|
"name": "search_web",
|
|
"description": "Search the web for current information. Returns a summary of results.",
|
|
"input_schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"query": {"type": "string", "description": "The search query"},
|
|
"reasoning": {"type": "string", "description": "Why are you searching?"}
|
|
},
|
|
"required": ["query", "reasoning"]
|
|
}
|
|
},
|
|
{
|
|
"name": "write_file",
|
|
"description": "Write content to a file at the given path",
|
|
"input_schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"path": {"type": "string", "description": "Path to write to"},
|
|
"content": {"type": "string", "description": "Content to write"},
|
|
"reasoning": {"type": "string", "description": "Why are you writing this?"}
|
|
},
|
|
"required": ["path", "content", "reasoning"]
|
|
}
|
|
}
|
|
]
|
|
|
|
|
|
def execute_tool(tool_name: str, tool_args: dict) -> str:
|
|
if tool_name == "read_file":
|
|
path = tool_args["path"]
|
|
try:
|
|
with open(path, "r") as f:
|
|
return f"Contents of {path}:\n{f.read()}"
|
|
except FileNotFoundError:
|
|
return f"Error: File not found at {path}"
|
|
except Exception as e:
|
|
return f"Error: {str(e)}"
|
|
|
|
elif tool_name == "search_web":
|
|
query = tool_args["query"]
|
|
try:
|
|
import urllib.request
|
|
import urllib.parse
|
|
encoded = urllib.parse.quote(query)
|
|
url = f"https://api.duckduckgo.com/?q={encoded}&format=json"
|
|
with urllib.request.urlopen(url, timeout=10) as resp:
|
|
data = json.loads(resp.read())
|
|
summary = data.get("AbstractText", "")
|
|
results = data.get("RelatedTopics", [])[:3]
|
|
result_texts = [r.get("Text", "") for r in results if isinstance(r, dict)]
|
|
parts = [f"Summary: {summary}" if summary else ""] + result_texts
|
|
return "\n".join(parts) if any(parts) else "No results found."
|
|
except Exception as e:
|
|
return f"Search failed: {str(e)}. Try manual search."
|
|
|
|
elif tool_name == "write_file":
|
|
path = tool_args["path"]
|
|
content = tool_args["content"]
|
|
try:
|
|
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
|
|
with open(path, "w") as f:
|
|
f.write(content)
|
|
return f"Successfully wrote {len(content)} bytes to {path}"
|
|
except Exception as e:
|
|
return f"Error writing file: {str(e)}"
|
|
|
|
return f"Unknown tool: {tool_name}"
|
|
|
|
|
|
def run_agent(prompt: str) -> str:
|
|
"""Run multi-tool agent loop. Works with real API or mock."""
|
|
|
|
system_prompt = """You are a helpful assistant with access to file operations and web search.
|
|
You can read files, search the web, and write files.
|
|
Combine tools as needed to fulfill the user's request."""
|
|
|
|
messages = [{"role": "user", "content": prompt}]
|
|
|
|
for _ in range(MAX_ITERATIONS):
|
|
response = client.messages.create(
|
|
model="claude-sonnet-4-20260501" if not IS_MOCK else "mock-model",
|
|
max_tokens=4096,
|
|
system=system_prompt,
|
|
messages=messages,
|
|
tools=TOOLS
|
|
)
|
|
|
|
if response.stop_reason == "tool_use":
|
|
for block in response.content:
|
|
if block.type == "tool_use":
|
|
result = execute_tool(block.name, block.input)
|
|
messages.append({"role": "assistant", "content": response.content})
|
|
messages.append({"role": "user", "content": result})
|
|
elif response.stop_reason == "end_turn":
|
|
return "".join(b.text for b in response.content if b.type == "text")
|
|
else:
|
|
return f"Unexpected: {response.stop_reason}"
|
|
|
|
return "Max iterations reached."
|
|
|
|
|
|
if __name__ == "__main__":
|
|
prompt = " ".join(sys.argv[1:]) if len(sys.argv) > 1 else "Search for recent AI agent news and save the results to research.txt"
|
|
print(run_agent(prompt))
|