Error Handling & Tool Selection

1from langchain_core.tools import tool
2
3@tool
4def search_flights(departure: str, destination: str, date: str) -> str:
5    """Tìm chuyến bay.
6    
7    Args:
8        departure: Mã sân bay đi (VD: HAN, SGN)
9        destination: Mã sân bay đến
10        date: Ngày bay (YYYY-MM-DD)
11    """
12    try:
13        # Validate input
14        if len(departure) != 3 or len(destination) != 3:
15            return "Error: Airport code phải là 3 ký tự (VD: HAN, SGN)"
16        
17        # Call API
18        results = call_flight_api(departure, destination, date)
19        
20        if not results:
21            return f"Không tìm thấy chuyến bay {departure} -> {destination} ngày {date}"
22        
23        return format_flights(results)
24        
25    except TimeoutError:
26        return "Error: Flight search timed out. Vui lòng thử lại."
27    except ConnectionError:
28        return "Error: Không thể kết nối flight API. Kiểm tra network."
29    except Exception as e:
30        return f"Error: Unexpected error - {str(e)}"

1import time
2import functools
3
4def retry_tool(max_retries=3, backoff_factor=2):
5    """Decorator to add retry logic to tools."""
6    def decorator(func):
7        @functools.wraps(func)
8        def wrapper(*args, **kwargs):
9            last_error = None
10            
11            for attempt in range(max_retries):
12                try:
13                    return func(*args, **kwargs)
14                except Exception as e:
15                    last_error = e
16                    if attempt < max_retries - 1:
17                        wait = backoff_factor ** attempt
18                        print(f"Retry {attempt + 1}/{max_retries} after {wait}s...")
19                        time.sleep(wait)
20            
21            return f"Error after {max_retries} retries: {str(last_error)}"
22        
23        return wrapper
24    return decorator
25
26# Usage
27@tool
28@retry_tool(max_retries=3, backoff_factor=2)
29def call_external_api(query: str) -> str:
30    """Call external API with auto-retry.
31    
32    Args:
33        query: Search query
34    """
35    # This will automatically retry on failure
36    response = requests.get(f"https://api.example.com/search?q={query}", timeout=10)
37    response.raise_for_status()
38    return response.json()

1from langgraph.graph import StateGraph, END
2from typing import TypedDict, Annotated
3import operator
4
5class AgentState(TypedDict):
6    messages: list
7    error_count: int
8    last_error: str
9
10def tool_node(state):
11    """Execute tool with error tracking."""
12    try:
13        # Execute tool
14        result = execute_tool(state["messages"][-1])
15        return {
16            "messages": state["messages"] + [result],
17            "error_count": 0,
18            "last_error": ""
19        }
20    except Exception as e:
21        return {
22            "messages": state["messages"],
23            "error_count": state.get("error_count", 0) + 1,
24            "last_error": str(e)
25        }
26
27def should_retry(state):
28    """Decide: retry, fallback, or give up."""
29    if state.get("error_count", 0) == 0:
30        return "continue"
31    elif state["error_count"] < 3:
32        return "retry"
33    else:
34        return "fallback"
35
36def fallback_node(state):
37    """Handle when all retries fail."""
38    error = state.get("last_error", "Unknown error")
39    message = f"Xin lỗi, tôi gặp lỗi: {error}. Vui lòng thử cách khác."
40    return {
41        "messages": state["messages"] + [{"role": "assistant", "content": message}]
42    }
43
44# Build graph with error handling
45graph = StateGraph(AgentState)
46graph.add_node("agent", agent_node)
47graph.add_node("tools", tool_node)
48graph.add_node("fallback", fallback_node)
49
50graph.add_conditional_edges("tools", should_retry, {
51    "continue": "agent",
52    "retry": "tools",
53    "fallback": "fallback"
54})

1import asyncio
2
3async def tool_with_timeout(tool_func, args, timeout_seconds=30):
4    """Execute tool with timeout."""
5    try:
6        result = await asyncio.wait_for(
7            tool_func.ainvoke(args),
8            timeout=timeout_seconds
9        )
10        return result
11    except asyncio.TimeoutError:
12        return f"Tool '{tool_func.name}' timed out after {timeout_seconds}s"

1# BAD: Vague description
2@tool
3def tool_a(query: str) -> str:
4    """Process data."""  # LLM doesn't know when to use this
5    ...
6
7# GOOD: Specific description
8@tool
9def search_product_catalog(product_name: str) -> str:
10    """Tìm kiếm sản phẩm trong catalog theo tên.
11    Sử dụng khi user hỏi về giá, thông tin, hoặc tìm sản phẩm.
12    KHÔNG dùng cho đơn hàng hoặc vận chuyển.
13    
14    Args:
15        product_name: Tên sản phẩm (VD: "iPhone 15", "MacBook Air")
16    """
17    ...

1# Organize tools by capability
2search_tools = [search_products, search_orders, search_customers]
3action_tools = [create_order, update_order, cancel_order]
4info_tools = [get_weather, get_exchange_rate, get_shipping_cost]
5
6# System prompt guides tool selection
7system_prompt = """You are a customer support agent.
8
9## Tool Usage Rules:
101. For product questions: use search_products
112. For order status: use search_orders FIRST, then get details
123. For creating orders: ALWAYS confirm with user before using create_order
134. For cancellation: check order status before cancel_order
145. NEVER call create_order and cancel_order in the same turn
15"""

1@tool
2def get_order_status(order_id: str) -> str:
3    """Kiểm tra trạng thái đơn hàng.
4    
5    Preconditions:
6    - order_id phải là format: ORD-XXXXX
7    - Gọi tool này TRƯỚC khi thực hiện bất kỳ action nào với đơn hàng
8    
9    Args:
10        order_id: Mã đơn hàng (format: ORD-XXXXX)
11    """
12    # Validate format
13    if not order_id.startswith("ORD-") or len(order_id) != 9:
14        return "Error: Invalid order ID format. Expected: ORD-XXXXX"
15    
16    # Look up order
17    order = db.get_order(order_id)
18    if not order:
19        return f"Error: Order {order_id} not found"
20    
21    return f"Order {order_id}: Status={order['status']}, Total={order['total']:,} VND"

1class FallbackToolChain:
2    """Try primary tool, fall back to alternatives."""
3    
4    def __init__(self, tools):
5        self.tools = tools  # Ordered by priority
6    
7    def execute(self, query):
8        errors = []
9        
10        for tool in self.tools:
11            try:
12                result = tool.invoke(query)
13                if not result.startswith("Error"):
14                    return result
15                errors.append(f"{tool.name}: {result}")
16            except Exception as e:
17                errors.append(f"{tool.name}: {str(e)}")
18        
19        # All tools failed
20        return "Tất cả sources đều không khả dụng:\n" + "\n".join(errors)
21
22# Usage
23search_chain = FallbackToolChain([
24    google_search,      # Primary
25    bing_search,        # Fallback 1
26    duckduckgo_search,  # Fallback 2
27])

1@tool
2def comprehensive_search(query: str) -> str:
3    """Tìm kiếm thông tin toàn diện.
4    
5    Args:
6        query: Câu hỏi hoặc từ khóa
7    """
8    results = {}
9    
10    # Try each source, collect what works
11    try:
12        results["web"] = web_search(query)
13    except Exception:
14        pass
15    
16    try:
17        results["database"] = db_search(query)
18    except Exception:
19        pass
20    
21    try:
22        results["documents"] = doc_search(query)
23    except Exception:
24        pass
25    
26    if not results:
27        return "Unable to search any source. Please try again later."
28    
29    # Combine available results
30    output = "Found results from: " + ", ".join(results.keys()) + "\n\n"
31    for source, data in results.items():
32        output += f"--- {source} ---\n{data}\n\n"
33    
34    return output