diff --git a/cookbook/pocketflow-advanced-tool-use/README.md b/cookbook/pocketflow-advanced-tool-use/README.md new file mode 100644 index 00000000..5e426826 --- /dev/null +++ b/cookbook/pocketflow-advanced-tool-use/README.md @@ -0,0 +1,96 @@ +# PocketFlow Advanced Tool Calling Demo + +This project shows how to build an agent that performs 2 advanced methods of tool calling (brought by Claude). + +1st is to encode tool metadata that could be search with embedding, instead of loading all tool info into context. **Progressively disclose** most matching tool info into context window, hence tool info could be more rich like adding best examples. + +2nd is (quote) "**Programmatic Tool Calling (PTC)** enables Claude to orchestrate tools **through code** rather than **through individual API round-trips**. Instead of Claude requesting tools one at a time with each result being returned to its context, Claude writes code that calls multiple tools, processes their outputs, and controls what information actually enters its context window." + +This implementation is based on : + +- the article: [Introducing advanced tool use on the Claude Developer Platform](https://www.anthropic.com/engineering/advanced-tool-use) + +- the cookbook: [Programmatic Tool Calling (PTC)](https://github.com/anthropics/claude-cookbooks/blob/main/tool_use/programmatic_tool_calling_ptc.ipynb) + +- the cookbook: [Tool Search With Embeddings](https://github.com/anthropics/claude-cookbooks/blob/main/tool_use/tool_search_with_embeddings.ipynb) + +## Features + +- Search relatively a big number of tools based on embedding +- Match with similairy score (easy to scale), progressively loading tool info into context +- Support both single tool calling based on PocketFlow +- Support multiple tool calling (by generated code using provided tools) + +## How to Run + +1. Set your API key: + ```bash + export OPENAI_API_KEY="your-api-key-here" + ``` + Or update it directly in `utils.py` + +2. Install and run: + ```bash + pip install -r requirements.txt + python main.py + ``` + +## An Easier Way for Multiple Tool Calling and Save Lots of Tokens + +### Typical Tool Calling Flow + +- User input a complex question (like a sequence of compound SQL statements) that needed rounds of tool calling (depending on how smart of the model either). + +- **Model Loaded all tool info into context** + +- Model run the **1st call and get the 1st result** . + +- Pass result into next round of prompt . + +- Start the **2nd round , and then 3rd , 4th**.... + +- According Claude's article , this will cause 10x more token usage + +### New Method + +- User input a complex question that needed rounds of tool calling. + +- Model search most-relevant tools and get tool name , params, input schema into context . + +- Model knew the tool could be called with python scripting , so generate python code to run . + +- Code finished running in sandbox and return results back to Model . + +- Then Model will process all the results in next round . + + +## How It Works + +```mermaid +flowchart LR + reason[ReasonNode] -->|search| tools[ToolsNode] + tools[ToolsNode] -->|reason| reason[ReasonNode] + reason -->|execute| execute[ExecuteToolNode] + reason -->|execute_coding| execute[ExecuteToolNode] + execute -->|reason| reason + reason -->|answer| answer[AnswerQuestion] +``` + + reason_node - "tool_search" >> tools_node + tools_node - "reason" >> reason_node + reason_node - "tool_execute" >> exec_node + reason_node - "tool_execute_coding" >> exec_node + exec_node - "reason" >> reason_node + reason_node - "answer" >> answer_node + +The agent uses PocketFlow to create a workflow where: +1. ReasonNode takes user input about Stock Tickers +2. ReasonNode choose what string to search with Tools (embeddings) +3. ToolsNode returned search results (similarity score, name, parameters) +4. ReasonNode choose to make single tool calling , or generate python code for multiple rounds of tool calling in one shot (which saved A LOT of tokens!) +5. AnswerNode craft final answers based on ReasonNode's context + +## Files + +- [`main.py`](./main.py): Implementation of nodes and flow assembling +- [`utils.py`](./utils.py): Helper functions tool calling and tool coding running (minimal unsafe way of using subprocess, only for demo purpose) diff --git a/cookbook/pocketflow-advanced-tool-use/main.py b/cookbook/pocketflow-advanced-tool-use/main.py new file mode 100644 index 00000000..e24b1a0f --- /dev/null +++ b/cookbook/pocketflow-advanced-tool-use/main.py @@ -0,0 +1,222 @@ +from pocketflow import Node, Flow +from utils import TOOL_LIBRARY, TOOL_SEARCH_DEFINITION, handle_tool_search, handle_tool_exec, call_llm, create_all_tools_embedding, run_user_code +import yaml +import sys +from datetime import datetime + +class ToolsNode(Node): + def prep(self, shared): + """Initialize and get tools""" + # The question is now passed from main via shared + tool_embeddings = create_all_tools_embedding() + return shared["query"], shared["top_k"], tool_embeddings + + def exec(self, inputs): + """Retrieve tools from the MCP server""" + tool_query, top_k, tool_embeddings = inputs + res = handle_tool_search(tool_query, top_k, tool_embeddings) + return res + + def post(self, shared, prep_res, exec_res): + """Store tools and process to decision node""" + tools_search_result = exec_res + shared["tools_search_result"] = tools_search_result + return "reason" + +class ReasonNode(Node): + def prep(self, shared): + """Prepare the prompt for LLM to process the question""" + question = shared["question"] + tools_search_result = shared.get("tools_search_result","no tool list") + tools_exec_result = shared.get("tools_exec_result","no tool execute result") + date = datetime.now().strftime("%Y-%m-%d") + + # Now is the time to combine with PocketFlow + # refer from #77 of https://github.com/anthropics/claude-cookbooks/blob/main/tool_use/tool_search_with_embeddings.ipynb + prompt = f""" +#### PROMPT START + +#### CONTEXT +You are a reasoning assistant. +CURRENT DATE:{date} +Question: {question} +Tool Recall List: {tools_search_result} +Tool Execute Result : {tools_exec_result} + +#### ACTION SPACE +[1] tool_search +{TOOL_SEARCH_DEFINITION} + +[2] tool_execute +Based on Tool Recall List and Question, use tool with proper parameters to answer Question. + +[3] tool_execute_coding +When single tool calling is not enough to answer the Question, follow provided tool input schema to generate python code. +All your generated python code will be executed within single python file. +ONLY ALLOW to call provided tool, tool calling result will be provided with Tool Execute Result of next round. + +[4] answer +Answer Question with proper reason + +#### NEXT ACTION +Decide the next action based on the context and available actions. +Return your response in this format: + +```yaml +thinking: | + +action: tool_search or tool_execute or tool_execute_coding or answer +reason: +query: +top_k: +tool_name: +tool_param: +code_block: +conclusion: +``` +IMPORTANT: Make sure to: +1. Use proper indentation (4 spaces) for all multi-line fields +2. Use the | character for multi-line text fields +3. Keep single-line fields without the | character + +#### END OF PROMPT +""" + return prompt + + def exec(self, prompt): + """Call LLM to process the question and decide which tool to use""" + print("πŸ€” Analyzing question and deciding which tool to use...") + response = call_llm(prompt) + return response + + def post(self, shared, prep_res, exec_res): + """Extract decision from YAML and save to shared context""" + try: + yaml_str = exec_res.split("```yaml")[1].split("```")[0].strip() + decision = yaml.safe_load(yaml_str) + shared["action"] = decision["action"] + except Exception as e: + print(f"❌ Error parsing LLM response: {e}") + print("Raw response:", exec_res) + exit(1) + if decision["action"] == "tool_search": + shared["reason"] = decision["reason"] + shared["query"] = decision["query"] + shared["top_k"] = decision["top_k"] + print(f"πŸ’‘ Reason Node Decide to query tool with string: {decision['query']}") + return "tool_search" + elif decision["action"] == "tool_execute": + shared["tool_name"] = decision["tool_name"] + shared["tool_param"] = decision["tool_param"] + print(f"πŸ’‘ Reason Node Decide to use tool. \n πŸ’‘NAME: {decision['tool_name']} , PARAMS : {decision['tool_param']}") + return "tool_execute" + elif decision["action"] == "tool_execute_coding": + shared["code_block"] = decision["code_block"] + print(f"πŸ’‘ Reason Node Decide to use tool coding \n πŸ’‘CODE : {decision['code_block']}") + return "tool_execute_coding" + elif decision["action"] == "answer": + print(" 🟒 Reason Node Decide to answer ") + shared["context"] = "Latest Reasonning : \n" + decision["thinking"] + "\n Tool Calling Result : \n" + decision["conclusion"] + return "answer" + else : + print("Action NOT DEFINED !! ") + exit(1) + +class ExecuteToolNode(Node): + def prep(self, shared): + """Prepare tool execution parameters""" + if shared["action"] == "tool_execute" : + return shared["action"], shared["tool_name"], shared["tool_param"] + elif shared["action"] == "tool_execute_coding" : + print(" exec node : coding ") + return shared["action"], shared["code_block"] + else: + print(" WRONG INPUT FOR EXEC NODE ") + exit(1) + + def exec(self, inputs): + """Execute the chosen tool""" + if inputs[0] == "tool_execute" : + intent, tool_name, parameters = inputs + result = handle_tool_exec(tool_name, parameters) + elif inputs[0] == "tool_execute_coding" : + intent, code_block = inputs + result = run_user_code(code_block) + else : + print("❌ no tool chosen or code generated") + exit(1) + + return result + + def post(self, shared, prep_res, exec_res): + print(f"\nβœ… Tool Execution Result is : {exec_res}") + shared["tools_exec_result"] = exec_res + return "reason" + + +class AnswerQuestion(Node): + def prep(self, shared): + """Get the question and context for answering.""" + return shared["question"], shared.get("context", "") + + def exec(self, inputs): + """Call the LLM to generate a final answer.""" + question, context = inputs + + print(f"✍️ Crafting final answer...") + + # Create a prompt for the LLM to answer the question + prompt = f""" +### CONTEXT +Based on the following information, answer the question. +Question: {question} +Conclusion: {context} + +## YOUR ANSWER: +Provide a comprehensive answer using the research results. +""" + # Call the LLM to generate an answer + answer = call_llm(prompt) + return answer + + def post(self, shared, prep_res, exec_res): + """Save the final answer and complete the flow.""" + # Save the answer in the shared store + shared["answer"] = exec_res + + print(f"βœ… Answer generated successfully") + + # We're done - no need to continue the flow + return "done" + +if __name__ == "__main__": + # Default question + default_question = "I want to know the latest price of NVDA/MSFT/QCOM" + + # Get question from command line if provided with -- + question = default_question + for arg in sys.argv[1:]: + if arg.startswith("--"): + question = arg[2:] + break + + print(f"πŸ€” Processing question: {question}") + + # Create nodes + reason_node = ReasonNode() + tools_node = ToolsNode() + exec_node = ExecuteToolNode() + answer_node = AnswerQuestion() + + # Connect nodes + reason_node - "tool_search" >> tools_node + tools_node - "reason" >> reason_node + reason_node - "tool_execute" >> exec_node + reason_node - "tool_execute_coding" >> exec_node + exec_node - "reason" >> reason_node + reason_node - "answer" >> answer_node + + # Create and run flow + flow = Flow(start=reason_node) + shared = {"question": question} + flow.run(shared) diff --git a/cookbook/pocketflow-advanced-tool-use/requirements.txt b/cookbook/pocketflow-advanced-tool-use/requirements.txt new file mode 100644 index 00000000..e6bde509 --- /dev/null +++ b/cookbook/pocketflow-advanced-tool-use/requirements.txt @@ -0,0 +1,5 @@ +pocketflow>=0.0.1 +openai>=1.0.0 +fastmcp +pyyaml +numpy \ No newline at end of file diff --git a/cookbook/pocketflow-advanced-tool-use/utils.py b/cookbook/pocketflow-advanced-tool-use/utils.py new file mode 100644 index 00000000..47558373 --- /dev/null +++ b/cookbook/pocketflow-advanced-tool-use/utils.py @@ -0,0 +1,430 @@ +from openai import OpenAI +import os + +def call_llm(prompt): + client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY", "your-api-key")) + r = client.chat.completions.create( + model="gpt-4o", + messages=[{"role": "user", "content": prompt}] + ) + return r.choices[0].message.content + +def get_embeddings(text_list : list) -> list: +# # DUMMY EMBEDDING API +# return + +# VIBE CODING WARNING: +# This is a minimal prototype for controlled/educational use. +# It is NOT a complete security boundary. Do not run untrusted code in production. +# Reference Article : https://www.anthropic.com/engineering/advanced-tool-use +# Reference Notebook : https://github.com/anthropics/claude-cookbooks/blob/main/tool_use/tool_search_with_embeddings.ipynb + +import yaml +import numpy as np + +# Define our tool library with 2 domains +TOOL_LIBRARY = [ + # Weather Tools + { + "name": "get_weather", + "description": "Get the current weather in a given location", + "input_schema": { + "type": "object", + "properties": { + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + }, + "unit": { + "type": "string", + "enum": ["celsius", "fahrenheit"], + "description": "The unit of temperature", + }, + }, + "required": ["location"], + }, + }, + { + "name": "get_forecast", + "description": "Get the weather forecast for multiple days ahead", + "input_schema": { + "type": "object", + "properties": { + "location": { + "type": "string", + "description": "The city and state", + }, + "days": { + "type": "number", + "description": "Number of days to forecast (1-10)", + }, + }, + "required": ["location", "days"], + }, + }, + { + "name": "get_timezone", + "description": "Get the current timezone and time for a location", + "input_schema": { + "type": "object", + "properties": { + "location": { + "type": "string", + "description": "City name or timezone identifier", + } + }, + "required": ["location"], + }, + }, + { + "name": "get_air_quality", + "description": "Get current air quality index and pollutant levels for a location", + "input_schema": { + "type": "object", + "properties": { + "location": { + "type": "string", + "description": "City name or coordinates", + } + }, + "required": ["location"], + }, + }, + # Finance Tools + { + "name": "get_stock_price", + "description": "Get the stock price for a given ticker symbol at a specified timestamp (UTC+8) .", + "input_schema": { + "type": "object", + "properties": { + "ticker": { + "type": "string", + "description": "Stock ticker symbol (e.g., AAPL, GOOGL)", + }, + "timestamp": { + "type": "string", + "description": "to locate the ticket price with exact timestamp from NASDAQ database", + }, + }, + "required": ["ticker", "timestamp"], + }, + }, + { + "name": "convert_currency", + "description": "Convert an amount from one currency to another using current exchange rates", + "input_schema": { + "type": "object", + "properties": { + "amount": { + "type": "number", + "description": "Amount to convert", + }, + "from_currency": { + "type": "string", + "description": "Source currency code (e.g., USD)", + }, + "to_currency": { + "type": "string", + "description": "Target currency code (e.g., EUR)", + }, + }, + "required": ["amount", "from_currency", "to_currency"], + }, + }, + { + "name": "calculate_compound_interest", + "description": "Calculate compound interest for investments over time", + "input_schema": { + "type": "object", + "properties": { + "principal": { + "type": "number", + "description": "Initial investment amount", + }, + "rate": { + "type": "number", + "description": "Annual interest rate (as percentage)", + }, + "years": {"type": "number", "description": "Number of years"}, + "frequency": { + "type": "string", + "enum": ["daily", "monthly", "quarterly", "annually"], + "description": "Compounding frequency", + }, + }, + "required": ["principal", "rate", "years"], + }, + }, + { + "name": "get_market_news", + "description": "Get recent financial news and market updates for a specific company or sector", + "input_schema": { + "type": "object", + "properties": { + "query": { + "type": "string", + "description": "Company name, ticker symbol, or sector", + }, + "limit": { + "type": "number", + "description": "Maximum number of news articles to return", + }, + }, + "required": ["query"], + }, + }, +] + + +# Line #71 of Notebook +def tool_to_text(tool) -> str: + """ + Convert a tool definition into a text representation for embedding. + Combines the tool name, description, and parameter information. + """ + text_parts = [ + f"Tool: {tool['name']}", + f"Description: {tool['description']}", + ] + + # Add parameter information + if "input_schema" in tool and "properties" in tool["input_schema"]: + params = tool["input_schema"]["properties"] + param_descriptions = [] + for param_name, param_info in params.items(): + param_desc = param_info.get("description", "") + param_type = param_info.get("type", "") + param_descriptions.append(f"{param_name} ({param_type}): {param_desc}") + + if param_descriptions: + text_parts.append("Parameters: " + ", ".join(param_descriptions)) + + return "\n".join(text_parts) + + +def create_all_tools_embedding(): + """ + Create tools text embedding in TOOL_LIBRARY + + Args: + None + + Returns: + List of tool embedding numpy array + """ + # Create embeddings for all tools + print("Creating embeddings for all tools...") + + tool_texts = [tool_to_text(tool) for tool in TOOL_LIBRARY] + + # ⚠️⚠️ API NOTICE + # Please adapt to your embedding APIs + tool_embeddings = [] + temp_obj = get_embeddings(tool_texts) + for item in temp_obj.data: + t_array = np.array(item.embedding) + tool_embeddings.append(t_array) + + print(f"βœ… Defined {len(TOOL_LIBRARY)} tools in the library") + + return tool_embeddings + + + +# Refer from +# Line #73 of : https://github.com/anthropics/claude-cookbooks/blob/main/tool_use/tool_search_with_embeddings.ipynb +def search_tools(query: str, top_k: int = 5, tool_embeddings:list=[]) -> list[dict]: + """ + Search for tools using semantic similarity. + + Args: + query: Natural language description of what tool is needed + top_k: Number of top tools to return + + Returns: + List of tool definitions most relevant to the query + """ + + # ⚠️⚠️ API NOTICE + # Please adapt to your embedding APIs + query_embedding = get_embeddings(query) + + q_array = np.array(query_embedding.data[0].embedding) + similarities = [] + for index, value in enumerate(tool_embeddings): + tmp = np.dot(value, q_array) + similarities.append(tmp) + + top_indices = np.argsort(similarities)[-top_k:][::-1] + + results = [] + for idx in top_indices: + results.append({"tool": TOOL_LIBRARY[idx], "similarity_score": float(similarities[idx])}) + + return results + + + +# The tool_search tool definition +TOOL_SEARCH_DEFINITION = { + "name": "tool_search", + "description": "Search for available tools that can help with a task. Returns tool definitions for matching tools. Use this when you need a tool but don't have it available yet.", + "input_schema": { + "type": "object", + "properties": { + "query": { + "type": "string", + "description": "Natural language description of what kind of tool you need (e.g., 'weather information', 'currency conversion', 'stock prices')", + }, + "top_k": { + "type": "number", + "description": "Number of tools to return (default: 5)", + }, + }, + "required": ["query"], + }, +} + +def handle_tool_exec(tool_name:str, tool_params:dict): + if tool_name == "get_weather" : + return "NYC : sunny" + elif tool_name == "get_forecast" : + return "NYC : tommorrow : sunny" + else : + return "tool executed" + +# in Line #75 of https://github.com/anthropics/claude-cookbooks/blob/main/tool_use/tool_search_with_embeddings.ipynb +# there is no input schema for tool search, adding it for my local experiment +def handle_tool_search(query: str, top_k: int = 5, tool_embeddings:list=[]) -> list[dict[str, any]]: + """ + Handle a tool_search invocation and return tool references. + + Returns a list of tool_reference content blocks for discovered tools. + """ + # Search for relevant tools + results = search_tools(query, top_k, tool_embeddings) + + # Create tool_reference objects instead of full definitions + tool_references = [ + {"type": "tool_reference", "tool_name": result["tool"]["name"], "similarity_score":f"{result['similarity_score']:.3f}","input_schema": result["tool"]["input_schema"]} for result in results + ] + + # the print is only for debug , not changing tool_references contents + print(f"\nπŸ” Tool search: '{query}'") + print(f" Found {len(tool_references)} tools:") + for i, result in enumerate(results, 1): + print(f" {i}. {result['tool']['name']} (similarity: {result['similarity_score']:.3f})") + + return tool_references + + +import time, tempfile, os, textwrap, sys, subprocess + +def run_user_code( + code: str, + input_data: str = "", +): + """ + Execute user Python code in a restricted subprocess. + Captures stdout/stderr and enforces time/memory limits. + + Returns: + dict: { + stdout (str), + stderr (str), + returncode (int) + } + """ + start = time.time() + with tempfile.TemporaryDirectory() as tmpdir: + user_code_path = os.path.join(tmpdir, "user_code.py") + + body = textwrap.dedent(code).strip("\n") + body = textwrap.indent(code, " ") + + # assemble tool code code + tmp_code = f""" + +def get_stock_price(ticker, timestamp): + if ticker == "NVDA": + return ticker + ": 888.88" + elif ticker == "MSFT": + return ticker + ": 666.66" + elif ticker == "QCOM": + return ticker + ": 444.44" + else: + return "Wrong Ticket or Timestamp" + +def __entry__(): +{body} + +if __name__ == "__main__": + __entry__() + + """ + + print(tmp_code) + + # Write user code to a temp file + with open(user_code_path, "w", encoding="utf-8") as f: + f.write(tmp_code) + + # Run Python in a more isolated mode: + # -I: isolated mode (ignores user site-packages and env vars) + # -S: do not import site automatically + # -u: unbuffered I/O for reliable output capture + cmd = [sys.executable, "-I", "-S", "-u", user_code_path] + + try: + result = subprocess.run( + cmd, + input=input_data, + capture_output=True, + text=True, + cwd=tmpdir, # confined working dir + env={}, # empty environment for fewer side effects + start_new_session=True, # separate session for termination control + ) + # duration = time.time() - start + return { + "stdout": result.stdout, + "stderr": result.stderr, + "returncode": result.returncode, + } + except subprocess.TimeoutExpired as e: + # On timeout, report and mark as timed_out + # duration = time.time() - start + return { + "stdout": e.stdout or "", + "stderr": (e.stderr or "") , + "returncode": -1, + # "duration": duration, + "timed_out": True, + } + + +if __name__ == "__main__": + print("βœ… Tool search definition created") + + # Test with one tool + sample_text = tool_to_text(TOOL_LIBRARY[0]) + print("Sample tool text representation:") + print(sample_text) + + # create tool embeddings + tool_embeddings = create_all_tools_embedding() + + test_query = "I need to check the weather" + test_results = search_tools(test_query, top_k=3, tool_embeddings=[]) + + print(f"➑️ Test Search Tool API, Query: '{test_query}'\n") + print("πŸ” Top 3 matching tools:") + for i, result in enumerate(test_results, 1): + tool_name = result["tool"]["name"] + score = result["similarity_score"] + print(f"{i}. {tool_name} (similarity: {score:.3f})") + + res = handle_tool_search("stock market data", top_k=3, tool_embeddings=[]) + print(f"\nReturned {len(res)} tool references:") + for ref in res: + print(f"{ref}") \ No newline at end of file