from typing import Dict, List, Any from tools_registry import Tool from huggingface_hub import InferenceClient import os import json from tools import convert_currency, show_my_memory class Agent: def __init__(self, tools: list[Tool]): """ For more information on `huggingface_hub` Inference API support, please check the docs: https://huggingface.co/docs/huggingface_hub/v0.22.2/en/guides/inference Inference API: a service that allows you to run accelerated inference on Hugging Face’s infrastructure for free. This service is a fast way to get started, test different models, and prototype AI products. """ # self.client = InferenceClient("HuggingFaceH4/zephyr-7b-beta") self.client = InferenceClient("Qwen/Qwen2.5-Coder-32B-Instruct") # self.client = InferenceClient("deepseek-ai/DeepSeek-V3-0324") """Initialize Agent with empty tool registry.""" self.tools: Dict[str, Tool] = {} for tool in tools: self.add_tool(tool) self.messages = [ {"role": "system", "content": self.create_system_prompt()}, ] def add_message(self, query_message: str) -> None: """Store the new message to the memory/state of the agent.""" self.messages.append({"role": "user", "content": query_message}) def add_tool(self, tool: Tool) -> None: """Register a new tool with the agent.""" self.tools[tool.name] = tool def get_available_tools(self) -> List[str]: """Get list of available tool descriptions.""" return [f"{tool.name}: {tool.description}" for tool in self.tools.values()] def use_tool(self, tool_name: str, **kwargs: Any) -> str: """Execute a specific tool with given arguments.""" if tool_name not in self.tools: raise ValueError(f"Tool '{tool_name}' not found. Available tools: {list(self.tools.keys())}") tool = self.tools[tool_name] return tool.func(**kwargs) def create_system_prompt(self) -> str: """Create the system prompt for the LLM with available tools.""" tools_json = { "role": "AI Assistant", "capabilities": [ "Using provided tools to help users when necessary", "Responding directly without tools for questions that don't require tool usage", "Planning efficient tool usage sequences" ], "instructions": [ "Use tools only when they are necessary for the task", "If a query can be answered directly, respond with a simple message instead of using tools", "When tools are needed, plan their usage efficiently to minimize tool calls" ], "tools": [ { "name": tool.name, "description": tool.description, "parameters": { name: { "type": info["type"], "description": info["description"] } for name, info in tool.parameters.items() } } for tool in self.tools.values() ], "response_format": { "type": "json", "schema": { "requires_tools": { "type": "boolean", "description": "whether tools are needed for this query" }, "direct_response": { "type": "string", "description": "response when no tools are needed", "optional": True }, "thought": { "type": "string", "description": "reasoning about how to solve the task (when tools are needed)", "optional": True }, "plan": { "type": "array", "items": {"type": "string"}, "description": "steps to solve the task (when tools are needed)", "optional": True }, "tool_calls": { "type": "array", "items": { "type": "object", "properties": { "tool": { "type": "string", "description": "name of the tool" }, "args": { "type": "object", "description": "parameters for the tool" } } }, "description": "tools to call in sequence (when tools are needed)", "optional": True } }, "examples": [ { "query": "Convert 100 USD to EUR", "response": { "requires_tools": True, "thought": "I need to use the currency conversion tool to convert USD to EUR", "plan": [ "Use convert_currency tool to convert 100 USD to EUR", "Return the conversion result" ], "tool_calls": [ { "tool": "convert_currency", "args": { "amount": 100, "from_currency": "USD", "to_currency": "EUR" } } ] } }, { "query": "What's 500 Japanese Yen in British Pounds?", "response": { "requires_tools": True, "thought": "I need to convert JPY to GBP using the currency converter", "plan": [ "Use convert_currency tool to convert 500 JPY to GBP", "Return the conversion result" ], "tool_calls": [ { "tool": "convert_currency", "args": { "amount": 500, "from_currency": "JPY", "to_currency": "GBP" } } ] } }, { "query": "What currency does Japan use?", "response": { "requires_tools": False, "direct_response": "Japan uses the Japanese Yen (JPY) as its official currency. This is common knowledge that doesn't require using the currency conversion tool." } } ] } } return f"""You are an AI assistant that helps users by providing direct answers or using tools when necessary. Configuration, instructions, and available tools are provided in JSON format below: {json.dumps(tools_json, indent=2)} Always respond with a JSON object following the response_format schema above. Remember to use tools only when they are actually needed for the task.""" def plan(self, user_query: str) -> Dict: """Use LLM to create a plan for tool usage.""" self.add_message(user_query) response = self.client.chat_completion( self.messages, max_tokens=512, temperature=0, top_p=0.95, ) print(response.choices[0]) try: return json.loads(response.choices[0].message.content) except json.JSONDecodeError: raise ValueError("Failed to parse LLM response as JSON") def execute(self, user_query: str) -> str: """Execute the full pipeline: plan and execute tools.""" try: plan = self.plan(user_query) if not plan.get("requires_tools", True): return plan["direct_response"] # Execute each tool in sequence results = [] for tool_call in plan["tool_calls"]: tool_name = tool_call["tool"] tool_args = tool_call["args"] result = self.use_tool(tool_name, **tool_args) results.append(result) # Combine results return f"""Thought: {plan['thought']} Plan: {'. '.join(plan['plan'])} Results: {'. '.join(results)}""" except Exception as e: return f"Error executing plan: {str(e)}" def main(): agent = Agent([convert_currency, show_my_memory]) query_list = ["I am traveling to Japan from Serbia, I have 1500 of local currency, how much of Japaese currency will I be able to get?"] for query in query_list: print(f"\nQuery: {query}") result = agent.execute(query) print(result) if __name__ == "__main__": main()