📖 7 min read

Tool Calling & Tavily Search#

This page explains how LLMs decide when and how to invoke external tools using structured function calling, with practical examples using the Tavily Search API to give agents access to real-time web information.

Learning Objectives#

Understand Tool/Function Calling in LLMs and how it works
Integrate external tools with agents effectively
Use Tavily Search API for optimized web search
Build agent with multiple tools and handle tool orchestration

1. What is Tool Calling?#

1.1 Concept#

Tool Calling (or Function Calling) is the ability of LLM to:

LLM decides when to use tools: Model automatically decides when to call tool based on user query
Structured tool invocation: Call tool with parameters formatted in standard format (JSON schema)
Parse tool results: Receive and process results returned from tool
Continue reasoning: Continue reasoning with new information to create final response

Basic Flow:

User Query → LLM analyzes → Decides to use tool → Calls tool with params
→ Tool executes → Returns result → LLM processes → Final response

1.2 Why Tool Calling?#

Extend LLM capabilities: Overcome knowledge cutoff and training data limitations
Access real-time data: Get real-time information (weather, stock prices, news)
Perform actions: Perform real actions (send email, create ticket, update database)
Integrate with APIs: Connect with external services and third-party APIs

1.3 Function Calling vs Tool Use#

Function Calling	Tool Use
OpenAI terminology	LangChain/Anthropic terminology
JSON schema for functions	Tool interface with description
Returns a function call object	Returns a tool invocation
Commonly used for OpenAI models	Framework-agnostic approach

2. OpenAI Function Calling#

2.1 API overview#

Define functions with JSON schema:

functions = [
    {
        "name": "search_web",
        "description": "Search the web for current information about a topic",
        "parameters": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "The search query"
                },
                "num_results": {
                    "type": "integer",
                    "description": "Number of results to return",
                    "default": 5
                }
            },
            "required": ["query"]
        }
    }
]

2.2 Request#

import openai

response = openai.ChatCompletion.create(
    model="gpt-4",
    messages=[
        {"role": "user", "content": "What's the weather in Hanoi?"}
    ],
    functions=functions,
    function_call="auto"  # "auto", "none", hoặc {"name": "function_name"}
)

2.3 Response#

{
    "choices": [{
        "message": {
            "role": "assistant",
            "content": null,
            "function_call": {
                "name": "search_web",
                "arguments": '{"query": "weather Hanoi today"}'
            }
        }
    }]
}

2.4 Executing Function#

import json

# Parse function call
function_call = response.choices[0].message.function_call
function_name = function_call.name
arguments = json.loads(function_call.arguments)

# Execute function
if function_name == "search_web":
    result = search_web(**arguments)

2.5 Continuation#

# Send result back to model
messages.append({
    "role": "function",
    "name": function_name,
    "content": json.dumps(result)
})

final_response = openai.ChatCompletion.create(
    model="gpt-4",
    messages=messages
)

3. LangChain Tools#

3.1 Tool Interface#

from langchain.tools import Tool

def search_function(query: str) -> str:
    """Search implementation"""
    # Call actual search API
    results = tavily_client.search(query)
    return str(results)

search_tool = Tool(
    name="WebSearch",
    func=search_function,
    description="Useful for searching the web for current information. Input should be a search query string."
)

3.2 @tool Decorator#

from langchain.tools import tool

@tool
def calculator(expression: str) -> str:
    """Useful for performing mathematical calculations.
    Input should be a valid Python mathematical expression."""
    try:
        result = eval(expression)
        return f"Result: {result}"
    except Exception as e:
        return f"Error: {str(e)}"

@tool
def get_current_time(timezone: str = "UTC") -> str:
    """Get current time in specified timezone.
    Input should be timezone string like 'UTC', 'Asia/Ho_Chi_Minh'."""
    from datetime import datetime
    import pytz
    tz = pytz.timezone(timezone)
    return datetime.now(tz).strftime("%Y-%m-%d %H:%M:%S %Z")

3.3 Built-in Tools#

# DuckDuckGo Search
from langchain_community.tools import DuckDuckGoSearchResults
search = DuckDuckGoSearchResults()

# Wikipedia
from langchain_community.tools import WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper
wikipedia = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper())

# Python REPL
from langchain_community.tools import PythonREPLTool
python_repl = PythonREPLTool()

# File Management
from langchain_community.tools import ReadFileTool, WriteFileTool
read_file = ReadFileTool()
write_file = WriteFileTool()

4. Tavily Search API#

4.1 Introduction to Tavily#

Tavily is a search engine optimized for AI:

AI-optimized search engine: Results are pre-formatted for LLMs
Designed for LLMs and RAG: Easy integration with AI workflows
Clean, relevant results: Filters out noise, only returns quality content
Real-time web search: Information updated in real-time

4.2 Features#

Web search: Search entire web with optimized ranking
News search: Specialized for latest news
Answer mode: Returns direct answers instead of a list of links
Context optimization: Optimize context for RAG applications

4.3 Getting Started#

# 1. Sign up at tavily.com
# 2. Get API key from dashboard
# 3. Install SDK
pip install tavily-python

4.4 Basic Usage#

from tavily import TavilyClient

# Initialize client
client = TavilyClient(api_key="tvly-xxxxxxxxxxxxx")

# Basic search
response = client.search(
    query="LangGraph tutorial 2025"
)

print(response["results"])

4.5 Search Parameters#

response = client.search(
    query="climate change solutions",           # Required
    search_depth="advanced",                    # "basic" or "advanced"
    max_results=10,                             # Max number of results
    include_domains=["edu", "gov"],             # Filter domains
    exclude_domains=["example.com"],            # Block domains
    include_answer=True,                        # Get AI-generated answer
    include_raw_content=False,                  # Include full page content
    include_images=True                         # Include image URLs
)

4.6 Response Structure#

{
  "query": "climate change solutions",
  "follow_up_questions": null,
  "answer": "Several effective climate change solutions include...",
  "images": ["https://...", "https://..."],
  "results": [
    {
      "title": "10 Solutions for Climate Change",
      "url": "https://example.com/article",
      "content": "Clean summary of the page content...",
      "score": 0.98,
      "raw_content": null
    }
  ],
  "response_time": 1.23
}

5. TavilySearchResults Tool#

5.1 LangChain Integration#

from langchain_community.tools.tavily_search import TavilySearchResults
import os

# Set API key
os.environ["TAVILY_API_KEY"] = "tvly-xxxxxxxxxxxxx"

# Create tool
search = TavilySearchResults(
    max_results=5,
    search_depth="advanced",
    include_answer=True,
    include_raw_content=False,
    include_domains=[],
    exclude_domains=[]
)

# Use tool
result = search.invoke("latest AI developments")

6. Advanced Tool Patterns#

6.1 Tool Chaining#

Output of one tool becomes input for another:

@tool
def search_company(company_name: str) -> str:
    """Search for company information"""
    return search.invoke(f"{company_name} official website")

@tool
def get_stock_price(ticker: str) -> str:
    """Get stock price from ticker symbol"""
    # ticker extracted from search_company result
    return f"${price}"

# Chain: company name → search → extract ticker → get price

7. Error Handling#

7.1 API Failures#

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
def call_tavily_with_retry(query):
    try:
        return client.search(query)
    except Exception as e:
        print(f"API call failed: {e}")
        raise

7.2 Timeout Handling#

import asyncio

async def execute_with_timeout(tool_func, timeout=30):
    try:
        result = await asyncio.wait_for(
            tool_func(),
            timeout=timeout
        )
        return result
    except asyncio.TimeoutError:
        return "Tool execution timeout"

8. Optimization#

8.1 Caching Tool Results#

from functools import lru_cache
import hashlib

# Simple LRU cache
@lru_cache(maxsize=100)
def cached_search(query: str):
    return client.search(query)

from datetime import datetime, timedelta

cache = {}
CACHE_TTL = timedelta(hours=1)

def search_with_cache(query):
    cache_key = hashlib.md5(query.encode()).hexdigest()

    if cache_key in cache:
        result, timestamp = cache[cache_key]
        if datetime.now() - timestamp < CACHE_TTL:
            return result

    result = client.search(query)
    cache[cache_key] = (result, datetime.now())
    return result

8.2 Rate Limiting#

from ratelimit import limits, sleep_and_retry

# Max 10 calls per minute
@sleep_and_retry
@limits(calls=10, period=60)
def rate_limited_search(query):
    return client.search(query)

8.3 Batching Requests#

async def batch_search(queries: list[str]):
    """Execute multiple searches efficiently"""
    tasks = [client.search_async(q) for q in queries]
    results = await asyncio.gather(*tasks)
    return results

9. Custom Tools#

9.1 Creating Custom Tool#

from langchain.tools import BaseTool
from pydantic import BaseModel, Field

class SearchInput(BaseModel):
    query: str = Field(description="Search query")
    domain: str = Field(description="Specific domain to search")

class DomainSearchTool(BaseTool):
    name = "domain_search"
    description = "Search within a specific domain"
    args_schema = SearchInput

    def _run(self, query: str, domain: str) -> str:
        """Synchronous implementation"""
        full_query = f"site:{domain} {query}"
        return client.search(full_query)

    async def _arun(self, query: str, domain: str) -> str:
        """Async implementation"""
        full_query = f"site:{domain} {query}"
        return await client.search_async(full_query)

10. Security Considerations#

10.1 API Key Management#

# ❌ Never hardcode
client = TavilyClient(api_key="tvly-xxxxx")

# ✅ Use environment variables
import os
from dotenv import load_dotenv

load_dotenv()
client = TavilyClient(api_key=os.getenv("TAVILY_API_KEY"))

# ✅ Use secret management
from azure.keyvault.secrets import SecretClient
api_key = secret_client.get_secret("tavily-api-key").value

11. Best Practices#

11.1 Tool Descriptions#

# ❌ Vague description
@tool
def search(query: str) -> str:
    """Search the web"""
    pass

# ✅ Clear, detailed description
@tool
def search(query: str) -> str:
    """
    Search the web for current information using Tavily API.
    Best used for: recent news, current events, factual information.
    Input should be a clear, specific search query.
    Returns: Top 5 relevant web results with summaries.
    """
    pass

12. PRACTICE#

From Research Agent in module 02, add another agent called Web Search tool (Tavily) to handle task for web search request AND coordinator must advise user to use web search for research request.
- Web search tool must be able to call in parallel along with other tools