Building Custom Tools

1from langchain_core.tools import tool
2
3@tool
4def search_product(query: str) -> str:
5    """Tìm kiếm sản phẩm trong database.
6    
7    Args:
8        query: Tên hoặc mô tả sản phẩm cần tìm
9    """
10    # Simulate database search
11    products = {
12        "laptop": "MacBook Air M3 - 25.990.000 VND",
13        "phone": "iPhone 15 - 22.990.000 VND",
14        "tablet": "iPad Air - 16.990.000 VND",
15    }
16    
17    for key, value in products.items():
18        if key in query.lower():
19            return value
20    
21    return "Không tìm thấy sản phẩm."
22
23# Tool properties
24print(search_product.name)         # "search_product"
25print(search_product.description)  # From docstring
26print(search_product.args_schema)  # Auto-generated from type hints

1@tool
2def calculate_shipping(
3    weight_kg: float,
4    destination: str,
5    express: bool = False
6) -> str:
7    """Tính phí vận chuyển dựa trên cân nặng và địa chỉ.
8    
9    Args:
10        weight_kg: Cân nặng kiện hàng (kg)
11        destination: Thành phố đích (VD: "Hà Nội", "TP.HCM")
12        express: Giao hàng nhanh (True/False)
13    """
14    base_rate = 15000  # VND/kg
15    
16    city_multiplier = {
17        "Hà Nội": 1.0,
18        "TP.HCM": 1.0,
19        "Đà Nẵng": 1.2,
20    }
21    
22    multiplier = city_multiplier.get(destination, 1.5)
23    cost = weight_kg * base_rate * multiplier
24    
25    if express:
26        cost *= 1.5
27    
28    return f"Phí ship {weight_kg}kg đến {destination}: {cost:,.0f} VND"

1from pydantic import BaseModel, Field
2from langchain_core.tools import StructuredTool
3from typing import Optional
4from datetime import date
5
6class BookingInput(BaseModel):
7    """Input schema for hotel booking."""
8    city: str = Field(description="Thành phố (VD: Đà Nẵng)")
9    check_in: str = Field(description="Ngày check-in (YYYY-MM-DD)")
10    check_out: str = Field(description="Ngày check-out (YYYY-MM-DD)")
11    guests: int = Field(default=2, description="Số khách")
12    max_price: Optional[int] = Field(default=None, description="Giá tối đa/đêm (VND)")
13
14def search_hotels(
15    city: str,
16    check_in: str,
17    check_out: str,
18    guests: int = 2,
19    max_price: int = None
20) -> str:
21    """Tìm kiếm khách sạn."""
22    results = f"Found 5 hotels in {city} for {guests} guests "
23    results += f"from {check_in} to {check_out}"
24    if max_price:
25        results += f" under {max_price:,} VND/night"
26    return results
27
28hotel_tool = StructuredTool.from_function(
29    func=search_hotels,
30    name="search_hotels",
31    description="Tìm kiếm và so sánh khách sạn",
32    args_schema=BookingInput
33)

1class WeatherResult(BaseModel):
2    city: str
3    temperature: float
4    condition: str
5    humidity: int
6
7@tool(response_format="content_and_artifact")
8def get_weather(city: str) -> tuple:
9    """Lấy thông tin thời tiết.
10    
11    Args:
12        city: Tên thành phố
13    """
14    # Simulate API call
15    data = WeatherResult(
16        city=city,
17        temperature=28.5,
18        condition="Partly cloudy",
19        humidity=75
20    )
21    
22    content = f"{city}: {data.temperature}°C, {data.condition}, Humidity: {data.humidity}%"
23    return content, data.model_dump()

1import sqlite3
2
3@tool
4def query_database(sql_query: str) -> str:
5    """Execute SQL query trên database sản phẩm.
6    
7    Args:
8        sql_query: SQL query (SELECT only). Tables: products, orders, customers.
9    """
10    # Safety check
11    if not sql_query.strip().upper().startswith("SELECT"):
12        return "Error: Only SELECT queries are allowed."
13    
14    dangerous_keywords = ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER"]
15    for keyword in dangerous_keywords:
16        if keyword in sql_query.upper():
17            return f"Error: {keyword} is not allowed."
18    
19    try:
20        conn = sqlite3.connect("store.db")
21        cursor = conn.cursor()
22        cursor.execute(sql_query)
23        results = cursor.fetchall()
24        columns = [desc[0] for desc in cursor.description]
25        conn.close()
26        
27        if not results:
28            return "No results found."
29        
30        # Format as table
31        output = " | ".join(columns) + "\n"
32        output += "-" * 50 + "\n"
33        for row in results:
34            output += " | ".join(str(v) for v in row) + "\n"
35        
36        return output
37    except Exception as e:
38        return f"Query error: {str(e)}"

1import json
2
3@tool
4def search_web(query: str, num_results: int = 3) -> str:
5    """Tìm kiếm thông tin trên web.
6    
7    Args:
8        query: Search query
9        num_results: Số kết quả trả về (1-5)
10    """
11    # Using SerpAPI or similar
12    # In production, use actual API
13    import requests
14    
15    params = {
16        "q": query,
17        "num": min(num_results, 5),
18    }
19    
20    # Simulated response
21    results = [
22        {"title": "Result 1", "snippet": "Description...", "link": "https://example.com"},
23    ]
24    
25    output = ""
26    for i, r in enumerate(results, 1):
27        output += f"{i}. {r['title']}\n   {r['snippet']}\n   {r['link']}\n\n"
28    
29    return output

1import os
2
3@tool
4def read_document(filepath: str) -> str:
5    """Đọc nội dung file văn bản.
6    
7    Args:
8        filepath: Đường dẫn file (chỉ .txt, .md, .csv)
9    """
10    allowed_extensions = [".txt", ".md", ".csv"]
11    ext = os.path.splitext(filepath)[1].lower()
12    
13    if ext not in allowed_extensions:
14        return f"Error: Only {allowed_extensions} files are supported."
15    
16    if not os.path.exists(filepath):
17        return f"Error: File not found: {filepath}"
18    
19    try:
20        with open(filepath, "r", encoding="utf-8") as f:
21            content = f.read()
22        
23        # Limit output size
24        if len(content) > 3000:
25            content = content[:3000] + "\n... (truncated)"
26        
27        return content
28    except Exception as e:
29        return f"Error reading file: {str(e)}"

1import asyncio
2import aiohttp
3
4@tool
5async def fetch_url(url: str) -> str:
6    """Fetch content from a URL.
7    
8    Args:
9        url: URL to fetch
10    """
11    async with aiohttp.ClientSession() as session:
12        async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
13            if resp.status == 200:
14                text = await resp.text()
15                return text[:2000]  # Truncate
16            return f"Error: HTTP {resp.status}"
17
18@tool
19async def parallel_search(queries: list) -> str:
20    """Search multiple queries in parallel.
21    
22    Args:
23        queries: List of search queries
24    """
25    async def search_one(q):
26        # Simulate async search
27        await asyncio.sleep(0.1)
28        return f"Results for '{q}': ..."
29    
30    tasks = [search_one(q) for q in queries]
31    results = await asyncio.gather(*tasks)
32    
33    return "\n".join(results)

1from langchain_openai import ChatOpenAI
2from langgraph.prebuilt import create_react_agent
3
4# Collect all tools
5tools = [
6    search_product,
7    calculate_shipping,
8    hotel_tool,
9    get_weather,
10    query_database,
11    search_web,
12    read_document,
13]
14
15# Create agent with tools
16llm = ChatOpenAI(model="gpt-4o-mini")
17agent = create_react_agent(llm, tools)
18
19# Run
20result = agent.invoke({
21    "messages": [("user", "Tìm laptop giá tốt và tính ship về Đà Nẵng 2kg")]
22})
23
24for msg in result["messages"]:
25    print(f"{msg.type}: {msg.content[:200]}")