Bài 25: Tools và Toolkits — gắn function vào agent

Sau bài này bạn sẽ:

• ✅ Hiểu Tool là abstraction gì và tại sao LLM cần schema để chọn tool
• ✅ Tạo Tool bằng 3 cách: @tool decorator, StructuredTool, BaseTool subclass
• ✅ Bind tool vào LLM và đọc tool_calls từ response
• ✅ Viết vòng lặp agent thủ công để execute tool và lấy kết quả cuối
• ✅ Biết khi nào dùng Toolkits có sẵn từ langchain_community
• ✅ Handle error trong tool và tránh các pitfall phổ biến

LLM chỉ sinh text. Nếu muốn nó thực sự tính toán, truy vấn database hay gọi API, bạn cần tool calling: LLM không thực thi trực tiếp mà trả về một "yêu cầu gọi hàm" có cấu trúc, ứng dụng của bạn thực thi, rồi trả kết quả lại cho LLM.

Trong LangChain 0.3.x, một Tool là wrapper trên function Python gồm 4 thành phần:

• name — tên duy nhất, LLM dùng để tham chiếu khi muốn gọi.
• description — LLM dựa vào đây để quyết định gọi tool này hay không. Mô tả sơ sài là nguyên nhân hàng đầu khiến LLM chọn sai tool.
• args_schema — Pydantic model hoặc JSON Schema mô tả arguments. LLM điền vào đây khi gọi.
• func / coroutine — implementation Python thực sự chạy.

Luồng hoạt động cơ bản:

User message
    │
    ▼
LLM nhận [message + tool list (name + description + schema)]
    │
    ├─ (quyết định không cần tool) → trả text bình thường
    │
    └─ (cần tool) → trả AIMessage với tool_calls:
         [{"name": "tên_tool", "args": {...}, "id": "call_xxx"}]
              │
              ▼
         Ứng dụng execute tool
              │
              ▼
         ToolMessage(content=kết_quả, tool_call_id="call_xxx")
              │
              ▼
         LLM nhận kết quả → sinh text trả lời cuối

Khác với các bài function calling thuần (raw OpenAI / Anthropic SDK trong Series 4), LangChain cung cấp abstraction Tool nhất quán, dùng được với mọi provider hỗ trợ function calling (OpenAI, Anthropic, Google Gemini, Mistral, Llama 3.1+).

Nhanh nhất. LangChain tự generate schema từ type hint và docstring.

from langchain_core.tools import tool

@tool
def add(a: int, b: int) -> int:
    """Cộng 2 số nguyên."""
    return a + b

@tool
def search_docs(query: str, top_k: int = 5) -> list[str]:
    """Tìm tài liệu nội bộ theo query. Trả về list snippet.

    Args:
        query: câu hỏi hoặc từ khóa cần tìm
        top_k: số kết quả tối đa, mặc định 5
    """
    # implementation thực tế gọi vector DB
    return [f"Kết quả {i+1} cho '{query}'" for i in range(top_k)]

# Kiểm tra schema được generate
print(add.name)           # "add"
print(add.description)    # "Cộng 2 số nguyên."
print(add.args_schema)    # Pydantic model tự generate
print(add.args)           # {"a": {"type": "integer"}, "b": {"type": "integer"}}

Lưu ý quan trọng:

• Docstring đầu tiên (trước Args:) trở thành description. Viết cụ thể: mô tả khi nào nên dùng tool này và nó làm gì chính xác.
• Type hint bắt buộc cho tất cả parameters. Thiếu type hint → schema sai → LLM không biết truyền gì.
• Default value trong signature được map sang Field(default=...).
• Decorator @tool cũng nhận tham số: @tool(name="custom_name", return_direct=True). return_direct=True để agent trả kết quả tool thẳng cho user, không qua LLM lần nữa.

@tool(name="web_search", return_direct=False)
def search_web(query: str) -> str:
    """Tìm kiếm thông tin trên web khi cần dữ liệu thời sự hoặc thông tin ngoài training data.
    Dùng khi câu hỏi liên quan đến sự kiện gần đây, giá cả, thời tiết, tin tức."""
    # call actual search API
    return f"Kết quả web cho: {query}"

Dùng khi muốn tách schema rõ ràng khỏi implementation, hoặc khi function đã có sẵn và không muốn sửa docstring.

from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field

# 1. Định nghĩa schema riêng với Field để mô tả từng arg
class SearchInput(BaseModel):
    query: str = Field(description="câu hỏi hoặc từ khóa cần tìm trong tài liệu nội bộ")
    top_k: int = Field(default=5, description="số kết quả tối đa, 1-20")
    filter_tag: str | None = Field(default=None, description="lọc theo tag nếu cần, ví dụ 'policy' hoặc 'faq'")

# 2. Implementation function (tách biệt hoàn toàn với schema)
def search_impl(query: str, top_k: int = 5, filter_tag: str | None = None) -> list[str]:
    # gọi vector DB thực
    return [f"doc_{i}" for i in range(top_k)]

# 3. Wrap thành Tool
search_tool = StructuredTool.from_function(
    func=search_impl,
    name="search_docs",
    description="Tìm tài liệu nội bộ theo nội dung. Dùng khi cần thông tin từ knowledge base của công ty.",
    args_schema=SearchInput,
)

# Async variant — thêm coroutine
async def search_async(query: str, top_k: int = 5, filter_tag: str | None = None) -> list[str]:
    # async implementation
    return [f"doc_{i}" for i in range(top_k)]

search_tool_async = StructuredTool.from_function(
    func=search_impl,          # sync fallback
    coroutine=search_async,    # async implementation
    name="search_docs_async",
    description="...",
    args_schema=SearchInput,
)

StructuredTool phù hợp hơn khi:

• Function đã có sẵn (legacy code, thư viện bên thứ ba).
• Schema cần validation phức tạp (custom validator, field aliasing).
• Muốn dùng cả sync lẫn async implementation riêng biệt.

Dùng khi tool cần giữ state (connection pool, session, cache) hoặc cần override hành vi bên trong invoke/ainvoke.

from typing import Type
from pydantic import BaseModel, Field
from langchain_core.tools import BaseTool

class DBQueryInput(BaseModel):
    sql: str = Field(description="câu SQL SELECT để chạy, chỉ đọc, không UPDATE/DELETE")
    limit: int = Field(default=10, description="giới hạn số row trả về, tối đa 100")

class DatabaseQueryTool(BaseTool):
    name: str = "database_query"
    description: str = (
        "Chạy câu SQL SELECT trên database nội bộ. "
        "Dùng khi cần tổng hợp số liệu, đếm record hoặc join bảng. "
        "Không dùng cho câu hỏi thông thường — chỉ khi cần data chính xác từ DB."
    )
    args_schema: Type[BaseModel] = DBQueryInput

    # state — connection pool giữ nguyên giữa các lần gọi
    connection_string: str = "postgresql://localhost/mydb"
    _conn: object = None  # private, không vào schema

    def _run(self, sql: str, limit: int = 10) -> str:
        """Synchronous execution."""
        # validate không có destructive SQL
        if any(kw in sql.upper() for kw in ["UPDATE", "DELETE", "DROP", "INSERT"]):
            raise ValueError(f"Chỉ cho phép SELECT. Câu SQL không hợp lệ: {sql[:80]}")
        # thực thi
        return f"[mock] {limit} rows từ: {sql}"

    async def _arun(self, sql: str, limit: int = 10) -> str:
        """Async execution — dùng async DB driver."""
        # asyncpg hoặc SQLAlchemy async
        return self._run(sql, limit)

db_tool = DatabaseQueryTool(connection_string="postgresql://prod-host/analytics")

Khi BaseTool là lựa chọn đúng:

• Tool cần inject dependency (DB connection, HTTP client, API key) vào constructor.
• Cần pre/post-processing bên trong tool (logging, metrics, rate limiting riêng).
• Cần override invoke để bắt exception theo cách đặc biệt trước khi raise.

bind_tools là method trên ChatModel (bất kỳ provider nào implement BaseChatModel). Nó sinh ra một Runnable mới — schema của tool được gửi kèm mọi request.

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

@tool
def add(a: int, b: int) -> int:
    """Cộng 2 số nguyên."""
    return a + b

@tool
def multiply(a: int, b: int) -> int:
    """Nhân 2 số nguyên."""
    return a * b

@tool
def lookup_rate(currency: str) -> float:
    """Lấy tỷ giá USD của đồng tiền chỉ định. Ví dụ: 'VND', 'EUR', 'JPY'."""
    rates = {"VND": 25400, "EUR": 0.92, "JPY": 157.3}
    return rates.get(currency.upper(), 0.0)

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
llm_with_tools = llm.bind_tools([add, multiply, lookup_rate])

# Gọi bình thường — LLM quyết định có dùng tool không
response = llm_with_tools.invoke("Tính 5 + 7")
print(type(response))       # AIMessage
print(response.tool_calls)  # [{"name": "add", "args": {"a": 5, "b": 7}, "id": "call_abc"}]
print(response.content)     # "" (rỗng khi LLM chọn gọi tool, chưa có câu trả lời)

# Câu hỏi không cần tool
response2 = llm_with_tools.invoke("Python ra đời năm nào?")
print(response2.tool_calls)  # []
print(response2.content)     # câu trả lời bình thường

Một số tham số hữu ích của bind_tools:

• tool_choice="auto" (mặc định) — LLM tự quyết định.
• tool_choice="required" — buộc LLM phải gọi ít nhất một tool.
• tool_choice={"type": "function", "function": {"name": "add"}} — buộc gọi đúng tool chỉ định.
• parallel_tool_calls=False — tắt parallel tool calls (OpenAI); dùng khi thứ tự quan trọng.

# Ví dụ: parallel tool calls tắt
llm_sequential = llm.bind_tools(
    [add, multiply],
    parallel_tool_calls=False,
)

Sau khi LLM trả về AIMessage có tool_calls, ứng dụng phải:

1. Tra tool theo name.
2. Gọi tool.invoke(args).
3. Wrap kết quả vào ToolMessage với đúng tool_call_id.
4. Gửi toàn bộ conversation (kể cả AIMessage + ToolMessage) cho LLM để lấy câu trả lời cuối.

import json
from langchain_core.messages import HumanMessage, ToolMessage

# Build tool map để lookup nhanh
tools = [add, multiply, lookup_rate]
tool_map = {t.name: t for t in tools}

# Bước 1: gửi message ban đầu
messages = [HumanMessage("Tỷ giá EUR là bao nhiêu? Và tính 12 * 8.")]
response = llm_with_tools.invoke(messages)
messages.append(response)  # giữ AIMessage trong history

# Bước 2: LLM có thể trả về nhiều tool_calls trong 1 response (parallel)
for tc in response.tool_calls:
    tool = tool_map[tc["name"]]
    try:
        result = tool.invoke(tc["args"])
        # ToolMessage.content phải là string
        content = json.dumps(result, default=str) if not isinstance(result, str) else result
    except Exception as e:
        content = f"Error: {e}"

    messages.append(
        ToolMessage(content=content, tool_call_id=tc["id"])
    )

# Bước 3: final response
final = llm_with_tools.invoke(messages)
print(final.content)
# "Tỷ giá EUR là 0.92 USD. 12 × 8 = 96."

Một số điểm cần nhớ:

• Mỗi ToolMessage phải có tool_call_id khớp với tc["id"]. Sai id → OpenAI trả lỗi 400.
• ToolMessage.content phải là str. Object/dict không serializable → dùng json.dumps(result, default=str).
• LLM có thể trả về nhiều tool_calls trong một response (parallel). Phải lặp qua tất cả, không chỉ lấy phần tử đầu.

Thực tế, LLM đôi khi cần gọi tool nhiều lượt (kết quả lượt 1 dùng để quyết định lượt 2). Pattern đơn giản nhất là vòng lặp: gọi LLM → nếu có tool_calls thì execute rồi gọi lại, đến khi không còn tool_calls.

import json
from langchain_core.messages import HumanMessage, ToolMessage

def run_agent(user_input: str, llm_with_tools, tool_map, max_iterations: int = 10) -> str:
    """
    Vòng lặp agent đơn giản. Trả content string của response cuối cùng.
    Raise RuntimeError nếu vượt max_iterations.
    """
    messages = [HumanMessage(user_input)]

    for iteration in range(max_iterations):
        response = llm_with_tools.invoke(messages)
        messages.append(response)

        # Không có tool_calls → LLM đã có câu trả lời cuối
        if not response.tool_calls:
            return response.content

        # Execute tất cả tool calls trong response này
        for tc in response.tool_calls:
            tool = tool_map.get(tc["name"])
            if tool is None:
                content = f"Error: tool '{tc['name']}' không tồn tại."
            else:
                try:
                    result = tool.invoke(tc["args"])
                    content = json.dumps(result, default=str) if not isinstance(result, str) else result
                except Exception as e:
                    content = f"Error: {e}"

            messages.append(ToolMessage(content=content, tool_call_id=tc["id"]))

    raise RuntimeError(f"Agent vượt {max_iterations} iterations mà chưa kết thúc.")


# Sử dụng
result = run_agent(
    "Tính (5 + 7) rồi nhân với 3",
    llm_with_tools=llm_with_tools,
    tool_map=tool_map,
    max_iterations=10,
)
print(result)  # "36"

Vòng lặp thủ công đủ dùng cho use case đơn giản, nhưng có giới hạn:

• Không có state persistence — mỗi lần gọi run_agent là conversation mới.
• Không có branching logic — không thể routing dựa trên kết quả tool.
• Khó debug — không có built-in tracing hay checkpoint.
• Human-in-the-loop không có cơ chế pause/resume.

LangGraph (Module 5) giải quyết những giới hạn này bằng state machine có checkpoint. Đây là một trong những lý do chính để chuyển sang LangGraph khi agent phức tạp hơn.

Toolkit là nhóm các tool liên quan đến cùng một hệ thống hay domain, đóng gói trong một class. Thay vì tự viết từng tool để tương tác với SQL database hoặc GitHub, bạn khởi tạo một Toolkit rồi lấy list tool ra.

# Ví dụ: SQLDatabaseToolkit
from langchain_community.agent_toolkits import SQLDatabaseToolkit
from langchain_community.utilities import SQLDatabase
from langchain_openai import ChatOpenAI

db = SQLDatabase.from_uri("sqlite:///./sales.db")
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

toolkit = SQLDatabaseToolkit(db=db, llm=llm)
tools = toolkit.get_tools()

# tools gồm:
# - sql_db_list_tables: liệt kê tất cả bảng
# - sql_db_schema: lấy schema của bảng
# - sql_db_query: chạy SQL SELECT
# - sql_db_query_checker: dùng LLM verify câu SQL trước khi chạy

print([t.name for t in tools])
# ['sql_db_list_tables', 'sql_db_schema', 'sql_db_query', 'sql_db_query_checker']

# Bind vào LLM như bình thường
llm_with_tools = llm.bind_tools(tools)

Các Toolkit phổ biến trong langchain_community (tính đến v0.3):

• SQLDatabaseToolkit — list tables, get schema, query SQL. Cần cài langchain-community sqlalchemy.
• GitHubToolkit — đọc file, tạo issue, comment, tạo pull request. Cần pygithub.
• GmailToolkit — send email, tìm email, đọc thread. Cần Google API credentials.
• JiraToolkit — tạo/đọc/update issue Jira. Cần atlassian-python-api.
• PlayWrightBrowserToolkit — điều khiển browser: navigate, click, extract text. Cần playwright.
• FileManagementToolkit — đọc/ghi/xóa file trong sandbox directory.

Khi nào dùng Toolkit thay vì tự viết Tool:

• Hệ thống đích đã có Toolkit sẵn → tiết kiệm thời gian, schema đã test kỹ.
• Cần nhiều tool liên quan (ví dụ 4 tool SQL) → quản lý gọn hơn.
• Vẫn cần tự viết nếu Toolkit không có sẵn hoặc cần logic nghiệp vụ riêng (validate, transform, auth).

Lưu ý: langchain_community không có trong langchain-core. Cài riêng:

pip install langchain-community
# Thêm dependency cho từng toolkit
pip install sqlalchemy       # SQLDatabaseToolkit
pip install pygithub         # GitHubToolkit

LangChain có hai cơ chế để agent chọn tool:

ReAct prompt-based agent (create_react_agent cũ) được giữ lại trong LangChain để tương thích ngược, nhưng hầu hết LLM hiện đại đã hỗ trợ function calling native. Với model mới, tool calling đáng tin cậy hơn đáng kể.

Ngoài ra, LangChain đã deprecate AgentExecutor (wrapper cao cấp quản lý vòng lặp agent). Pattern thay thế là tự viết loop thủ công (như mục 8) hoặc dùng LangGraph cho trường hợp phức tạp.

Khi tool raise exception, có 3 cách xử lý:

Cách A — Catch trong vòng lặp agent (khuyến nghị)

@tool
def divide(a: float, b: float) -> float:
    """Chia a cho b. Không dùng khi b = 0."""
    if b == 0:
        raise ValueError("Mẫu số không được bằng 0")
    return a / b

# Trong vòng lặp agent
for tc in response.tool_calls:
    tool = tool_map[tc["name"]]
    try:
        result = tool.invoke(tc["args"])
        content = str(result)
    except Exception as e:
        # Trả error description cho LLM — nó sẽ thử cách khác hoặc giải thích
        content = f"Tool error: {type(e).__name__}: {e}"

    messages.append(ToolMessage(content=content, tool_call_id=tc["id"]))

Cách B — handle_tool_error trên StructuredTool

def _divide(a: float, b: float) -> float:
    if b == 0:
        raise ValueError("Mẫu số không được bằng 0")
    return a / b

divide_tool = StructuredTool.from_function(
    func=_divide,
    name="divide",
    description="Chia a cho b.",
    handle_tool_error=True,  # convert exception thành ToolMessage error string tự động
)

# Khi LLM gọi divide(a=5, b=0), tool trả về chuỗi lỗi thay vì raise
# ToolMessage.content = "ValueError: Mẫu số không được bằng 0"

handle_tool_error cũng nhận callable: handle_tool_error=lambda e: f"Lỗi xử lý: {e}".

Cách C — Validate schema trước khi execute

from pydantic import ValidationError

for tc in response.tool_calls:
    tool = tool_map.get(tc["name"])
    if tool is None:
        content = f"Tool '{tc['name']}' không tồn tại trong tool_map."
        messages.append(ToolMessage(content=content, tool_call_id=tc["id"]))
        continue

    # Validate args trước
    try:
        tool.args_schema.model_validate(tc["args"])
    except ValidationError as e:
        content = f"Tham số không hợp lệ: {e.errors()}"
        messages.append(ToolMessage(content=content, tool_call_id=tc["id"]))
        continue

    # Execute sau khi validate xong
    result = tool.invoke(tc["args"])
    messages.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))

Khi LLM nhận ToolMessage chứa error description, nó thường sẽ:

• Thử lại với args khác (tự sửa).
• Gọi tool khác phù hợp hơn.
• Trả lời cho user biết không thể thực hiện thao tác đó.

Hành vi phụ thuộc vào model. GPT-4o và Claude 3.5+ xử lý error recovery tốt hơn các model nhỏ.

Khi dùng .stream(), LLM yield AIMessageChunk từng phần. Tool args được stream incremental dưới dạng tool_call_chunks — mỗi chunk chứa fragment của JSON args.

from langchain_core.messages import AIMessageChunk

chunks = []
for chunk in llm_with_tools.stream("Tính 100 + 200"):
    chunks.append(chunk)
    # In text content nếu có (phase trả lời, không phải phase tool call)
    if chunk.content:
        print(chunk.content, end="", flush=True)

# Sau khi stream xong, aggregate tất cả chunks
final_message = chunks[0]
for chunk in chunks[1:]:
    final_message = final_message + chunk  # AIMessageChunk.__add__ merge tool_call_chunks

# final_message.tool_calls đã đầy đủ — dùng bình thường
if final_message.tool_calls:
    for tc in final_message.tool_calls:
        tool = tool_map[tc["name"]]
        result = tool.invoke(tc["args"])
        print(f"\n[tool: {tc['name']}] → {result}")

Điểm thực tế:

• Trong phase tool calling, chunk.content thường rỗng — LLM chưa có gì để trả lời.
• Sau khi aggregate, final_message.tool_calls có cấu trúc giống AIMessage.tool_calls thông thường.
• Streaming chủ yếu có giá trị ở phase final response (sau khi tool đã execute), để text trả lời hiện ra dần.

1. Description sơ sài

LLM chọn tool dựa vào description, không phải tên. Description ngắn, chung chung dẫn đến LLM chọn sai hoặc không chọn khi cần.

# Tệ — LLM không biết khi nào dùng
@tool
def search(query: str) -> str:
    """Tìm kiếm."""
    ...

# Tốt — rõ phạm vi và use case
@tool
def search_internal_docs(query: str) -> str:
    """Tìm kiếm trong tài liệu nội bộ của công ty (wiki, policy, SOP).
    Dùng khi câu hỏi liên quan đến quy trình, chính sách hoặc hướng dẫn nội bộ.
    Không dùng cho câu hỏi thông tin ngoài (tin tức, web search)."""
    ...

2. Quên handle parallel tool_calls

LLM (đặc biệt OpenAI) có thể trả về nhiều tool_calls trong một response. Nếu chỉ xử lý response.tool_calls[0], các tool còn lại bị bỏ qua → history thiếu ToolMessage → API lỗi ở request tiếp theo.

# Sai — chỉ xử lý tool đầu tiên
tc = response.tool_calls[0]
result = tool_map[tc["name"]].invoke(tc["args"])

# Đúng — lặp qua tất cả
for tc in response.tool_calls:
    result = tool_map[tc["name"]].invoke(tc["args"])
    messages.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))

3. Tool có side-effect không idempotent

Tool như "gửi email" hay "tạo order" không thể retry an toàn. Nếu lỗi xảy ra sau khi tool đã execute nhưng trước khi ToolMessage được ghi, retry sẽ gửi email lần 2.

Giải pháp: kiểm tra idempotency key, dùng deduplication ở tầng API, hoặc tách "tool preview" (chỉ trả JSON plan) khỏi "tool commit" (thực sự thực thi).

4. ToolMessage.content không phải string

# Sai — content là dict
ToolMessage(content={"result": 42}, tool_call_id=tc["id"])

# Đúng
import json
ToolMessage(content=json.dumps({"result": 42}), tool_call_id=tc["id"])
# hoặc đơn giản
ToolMessage(content=str(result), tool_call_id=tc["id"])

5. Tên tool trùng

bind_tools raise ValueError nếu hai tool có cùng name. Kiểm tra trước khi bind:

names = [t.name for t in tools]
assert len(names) == len(set(names)), f"Tên tool trùng: {names}"

6. Schema field trùng từ khóa Python

Pydantic raise nếu field name là built-in Python keyword (type, class, id). Dùng Field(alias="type") hoặc đổi tên field.

from pydantic import BaseModel, Field

# Sai
class BadInput(BaseModel):
    type: str  # 'type' là shadowed name trong Pydantic

# Đúng
class GoodInput(BaseModel):
    item_type: str = Field(description="loại item: 'invoice' hoặc 'receipt'")

• Tool = wrapper trên function Python với name, description, args_schema, func.
• 3 cách tạo: @tool decorator (nhanh), StructuredTool.from_function (schema tách biệt), BaseTool subclass (state + custom logic).
• description là yếu tố quan trọng nhất để LLM chọn đúng tool.
• llm.bind_tools([...]) sinh Runnable mới, gửi schema kèm mọi request.
• Execute tool call: loop qua response.tool_calls, invoke, append ToolMessage với đúng tool_call_id.
• Vòng lặp agent thủ công đơn giản, nhưng không có state persistence hay branching — LangGraph giải quyết điều này.
• Toolkit (langchain_community) cung cấp 30+ bộ tool domain-specific (SQL, GitHub, Gmail...).
• Dùng tool calling native thay vì ReAct prompt-based với model hiện đại.
• ToolMessage.content phải là string; luôn loop qua tất cả tool_calls, không chỉ phần tử đầu.

Bài 26: Vì sao cần LangGraph khi đã có LangChain — vòng lặp agent thủ công ở bài này có các giới hạn cụ thể; bài tiếp theo phân tích chính xác những điểm đó và giải thích kiến trúc state machine của LangGraph.