1from openai import OpenAI
2
3client = OpenAI()
4
5# Define tools
6tools = [
7    {
8        "type": "function",
9        "function": {
10            "name": "get_weather",
11            "description": "Lấy thông tin thời tiết hiện tại",
12            "parameters": {
13                "type": "object",
14                "properties": {
15                    "city": {
16                        "type": "string",
17                        "description": "Tên thành phố"
18                    },
19                    "unit": {
20                        "type": "string",
21                        "enum": ["celsius", "fahrenheit"],
22                        "description": "Đơn vị nhiệt độ"
23                    }
24                },
25                "required": ["city"]
26            }
27        }
28    }
29]
30
31# Call with tools
32response = client.chat.completions.create(
33    model="gpt-4o",
34    messages=[{"role": "user", "content": "Thời tiết Hà Nội thế nào?"}],
35    tools=tools,
36    tool_choice="auto"  # Tự động quyết định
37)
38
39# Check if tool was called
40if response.choices[0].message.tool_calls:
41    tool_call = response.choices[0].message.tool_calls[0]
42    print(f"Function: {tool_call.function.name}")
43    print(f"Arguments: {tool_call.function.arguments}")

1from langchain.tools import tool
2
3@tool
4def search_web(query: str) -> str:
5    """
6    Tìm kiếm thông tin trên web.
7    
8    Args:
9        query: Từ khóa tìm kiếm
10        
11    Returns:
12        Kết quả tìm kiếm
13    """
14    # Implementation
15    return f"Results for: {query}"
16
17@tool
18def calculate(expression: str) -> str:
19    """
20    Tính toán biểu thức toán học.
21    
22    Args:
23        expression: Biểu thức (ví dụ: "2 + 2 * 3")
24        
25    Returns:
26        Kết quả
27    """
28    try:
29        return str(eval(expression))
30    except:
31        return "Error: Invalid expression"

1from langchain.tools import StructuredTool
2from pydantic import BaseModel, Field
3
4class EmailInput(BaseModel):
5    to: str = Field(description="Email người nhận")
6    subject: str = Field(description="Tiêu đề email")
7    body: str = Field(description="Nội dung email")
8    cc: list[str] | None = Field(default=None, description="CC list")
9
10def send_email(to: str, subject: str, body: str, cc: list[str] = None) -> str:
11    """Gửi email"""
12    # Implementation
13    return f"Email sent to {to}"
14
15email_tool = StructuredTool.from_function(
16    func=send_email,
17    name="send_email",
18    description="Gửi email đến người nhận",
19    args_schema=EmailInput
20)

1from langchain.tools import BaseTool
2from pydantic import BaseModel, Field
3from typing import Type
4
5class DatabaseQueryInput(BaseModel):
6    query: str = Field(description="SQL query")
7    database: str = Field(default="main", description="Database name")
8
9class DatabaseQueryTool(BaseTool):
10    name: str = "query_database"
11    description: str = "Execute SQL query on database"
12    args_schema: Type[BaseModel] = DatabaseQueryInput
13    
14    def _run(self, query: str, database: str = "main") -> str:
15        """Execute the query"""
16        # Implement database query
17        return f"Query executed on {database}"
18    
19    async def _arun(self, query: str, database: str = "main") -> str:
20        """Async version"""
21        return self._run(query, database)

1# Search
2from langchain_community.tools import DuckDuckGoSearchRun
3search = DuckDuckGoSearchRun()
4
5# Wikipedia
6from langchain_community.tools import WikipediaQueryRun
7from langchain_community.utilities import WikipediaAPIWrapper
8wiki = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper())
9
10# Python REPL
11from langchain_experimental.tools import PythonREPLTool
12python_repl = PythonREPLTool()
13
14# File operations
15from langchain_community.tools import ReadFileTool, WriteFileTool
16read_file = ReadFileTool()
17write_file = WriteFileTool()
18
19# Requests
20from langchain_community.tools import RequestsGetTool
21requests_tool = RequestsGetTool()

1from langchain.agents import create_tool_calling_agent, AgentExecutor
2from langchain_openai import ChatOpenAI
3
4llm = ChatOpenAI(model="gpt-4o")
5
6tools = [search, calculate, get_weather]
7
8agent = create_tool_calling_agent(llm, tools, prompt)
9executor = AgentExecutor(agent=agent, tools=tools)
10
11# LLM tự quyết định dùng tool nào
12result = executor.invoke({"input": "Thời tiết Tokyo và tính 100 USD sang JPY"})

1response = llm.invoke(
2    messages,
3    tools=tools,
4    tool_choice={"type": "function", "function": {"name": "get_weather"}}
5)

1response = llm.invoke(
2    messages,
3    tools=tools,
4    tool_choice="required"
5)

1@tool
2def get_stock_price(symbol: str) -> str:
3    """Lấy giá cổ phiếu"""
4    # API call
5    return f"{symbol}: $150"
6
7# Query cần multiple tools
8query = "Giá của AAPL, GOOGL và MSFT là bao nhiêu?"
9
10# GPT-4 sẽ generate parallel tool calls
11response = llm.invoke([HumanMessage(query)], tools=[get_stock_price])
12
13# Response có thể chứa multiple tool calls
14for tool_call in response.tool_calls:
15    print(f"Calling {tool_call['name']} with {tool_call['args']}")

1@tool
2def risky_operation(data: str) -> str:
3    """Operation có thể fail"""
4    try:
5        result = process(data)
6        return f"Success: {result}"
7    except ValidationError as e:
8        return f"Validation error: {e}. Please check input format."
9    except TimeoutError:
10        return "Operation timed out. Please try again later."
11    except Exception as e:
12        return f"Unexpected error: {e}. Please contact support."

1from tenacity import retry, stop_after_attempt, wait_exponential
2
3@tool
4@retry(
5    stop=stop_after_attempt(3),
6    wait=wait_exponential(min=1, max=10)
7)
8def api_call(endpoint: str) -> str:
9    """Call external API với retry"""
10    response = requests.get(endpoint, timeout=5)
11    response.raise_for_status()
12    return response.json()

1from pydantic import BaseModel, Field, validator
2
3class SQLQueryInput(BaseModel):
4    query: str = Field(description="SQL query to execute")
5    
6    @validator('query')
7    def validate_query(cls, v):
8        # Prevent SQL injection
9        dangerous_keywords = ['DROP', 'DELETE', 'TRUNCATE', 'UPDATE']
10        for keyword in dangerous_keywords:
11            if keyword in v.upper():
12                raise ValueError(f"Dangerous keyword '{keyword}' not allowed")
13        return v
14
15@tool(args_schema=SQLQueryInput)
16def safe_sql_query(query: str) -> str:
17    """Execute safe SQL query"""
18    # Only SELECT queries allowed
19    return execute_query(query)

1import aiohttp
2
3@tool
4async def async_search(query: str) -> str:
5    """Async web search"""
6    async with aiohttp.ClientSession() as session:
7        async with session.get(f"https://api.search.com?q={query}") as response:
8            return await response.text()
9
10# Use with async agent
11result = await agent_executor.ainvoke({"input": "Search for Python tutorials"})

1# Starter code
2from langchain.tools import tool
3from pydantic import BaseModel, Field
4
5class ProductSearchInput(BaseModel):
6    query: str = Field(description="Từ khóa tìm kiếm")
7    category: str | None = Field(default=None, description="Danh mục")
8    max_price: float | None = Field(default=None, description="Giá tối đa")
9
10@tool(args_schema=ProductSearchInput)
11def search_products(query: str, category: str = None, max_price: float = None) -> str:
12    """Tìm kiếm sản phẩm"""
13    # TODO: Implement
14    pass
15
16# TODO: Add more tools