Tại sao Function Calling Thúc đẩy Các Agent Phức tạp

Gọi hàm của Claude—chính thức là "tool use" trong Anthropic API—cho phép các agent AI tương tác động với các công cụ bên ngoài, API và codebase. Không giống như các chat completions đơn giản, nó cho phép Claude suy luận từng bước, gọi nhiều công cụ song song và tích hợp kết quả vào chuỗi suy nghĩ của nó. Điều này hoàn hảo để xây dựng các agent tự động hóa workflow như nghiên cứu, phân tích dữ liệu hoặc tích hợp với các dịch vụ như Zapier hoặc n8n.

Đối với các lập trình viên, các cuộc gọi công cụ song song (giới thiệu trong Claude 3.5 Sonnet) có nghĩa là thực thi nhanh hơn mà không có nút thắt thứ tự. Trong hướng dẫn này, chúng ta sẽ xây dựng từ cơ bản đến một orchestrator agent đầy đủ, tập trung vào Python với Anthropic SDK.

Thiết lập Môi trường Của Bạn

Bắt đầu bằng cách cài đặt Anthropic SDK:

pip install anthropic

Lấy API key của bạn từ Anthropic Console (console.anthropic.com). Đặt nó làm biến môi trường:

export ANTHROPIC_API_KEY='your-api-key-here'

Khởi tạo client:

import os
import json
from anthropic import Anthropic, HUMAN_PROMPT, AI_PROMPT

client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

Chúng ta sẽ sử dụng Claude 3.5 Sonnet (claude-3-5-sonnet-20240620) nhờ hỗ trợ công cụ song song vượt trội của nó.

Định nghĩa Công cụ Với JSON Schema

Các công cụ được định nghĩa dưới dạng JSON schema. Claude phân tích chúng và tạo ra các cuộc gọi với các đối số chính xác.

Ví dụ các công cụ cho một agent nghiên cứu:

tools = [
    {
        "name": "web_search",
        "description": "Search the web for current information.",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Search query"}
            },
            "required": ["query"]
        }
    },
    {
        "name": "summarize_text",
        "description": "Summarize given text.",
        "input_schema": {
            "type": "object",
            "properties": {
                "text": {"type": "string", "description": "Text to summarize"},
                "max_length": {"type": "integer", "description": "Max summary length", "default": 200}
            },
            "required": ["text"]
        }
    },
    {
        "name": "get_weather",
        "description": "Get current weather for a location.",
        "input_schema": {
            "type": "object",
            "properties": {
                "location": {"type": "string", "description": "City or location"}
            },
            "required": ["location"]
        }
    }
]

Những công cụ giả lập này mô phỏng các tích hợp thực tế (ví dụ: web_search có thể gọi SerpAPI, get_weather qua OpenWeatherMap).

Ví dụ Gọi Hàm Đơn

Gửi một tin nhắn với các công cụ:

message = client.messages.create(
    model="claude-3-5-sonnet-20240620",
    max_tokens=1024,
    tools=tools,
    messages=[{"role": "user", "content": "What's the weather in NYC?"}]
)

print(message.content)

Claude trả lời với một khối tool_use:

[{"type": "tool_use", "id": "toolu_01...", "name": "get_weather", "input": {"location": "New York City"}}]

Thực thi công cụ và trả lời:

def execute_tool(tool_name, tool_input):
    if tool_name == "get_weather":
        return "Sunny, 75°F in NYC."
    # Add more logic
    return "Mock result"

tool_input = message.content[0].input
result = execute_tool(message.content[0].name, tool_input)

tool_result = {
    "tool_use_id": message.content[0].id,
    "type": "tool_result",
    "content": result
}

final_message = client.messages.create(
    model="claude-3-5-sonnet-20240620",
    max_tokens=1024,
    tools=tools,
    messages=[
        {"role": "user", "content": "What's the weather in NYC?"},
        {"role": "assistant", "content": [{"type": "tool_use", "id": message.content[0].id, "name": message.content[0].name, "input": tool_input}]},
        {"role": "user", "content": [tool_result]}
    ]
)

print(final_message.content[0].text)  # "The weather in NYC is sunny, 75°F."

Các Cuộc Gọi Hàm Song Song

Claude có thể yêu cầu nhiều công cụ cùng lúc. Phát hiện và thực thi đồng thời bằng cách sử dụng concurrent.futures:

import concurrent.futures

def handle_parallel_tools(message):
    tool_uses = [block for block in message.content if block.type == "tool_use"]
    futures = []

    with concurrent.futures.ThreadPoolExecutor() as executor:
        for tool_use in tool_uses:
            future = executor.submit(execute_tool, tool_use.name, tool_use.input)
            futures.append((tool_use.id, future))

    tool_results = []
    for tool_id, future in futures:
        result = future.result()
        tool_results.append({
            "tool_use_id": tool_id,
            "type": "tool_result",
            "content": result
        })
    return tool_results

Gợi ý cho Claude: "Compare weather in NYC and SF, and summarize recent AI news."

Claude có thể gọi get_weather hai lần (các vị trí khác nhau) và web_search song song—thực thi trong <1s so với thứ tự.

Xây dựng Một Orchestrator Agent Đa Bước

Các agent cần vòng lặp cho reasoning, action, observation (mô hình ReAct). Đây là một orchestrator đầy đủ:

def agent_loop(user_query, max_steps=10):
    messages = [{"role": "user", "content": user_query}]

    for step in range(max_steps):
        response = client.messages.create(
            model="claude-3-5-sonnet-20240620",
            max_tokens=1024,
            tools=tools,
            messages=messages
        )

        # Append assistant response
        messages.append({"role": "assistant", "content": response.content})

        # Check for tool uses
        tool_uses = [block for block in response.content if block.type == "tool_use"]
        if not tool_uses:
            return response.content[0].text  # Final answer

        # Execute in parallel
        tool_results = handle_parallel_tools(response)
        messages.append({"role": "user", "content": tool_results})

    return "Max steps reached."

# Usage
result = agent_loop("Plan a trip to NYC: check weather, search hotels, summarize attractions.")
print(result)

Vòng lặp này tiếp tục cho đến khi Claude ngừng gọi công cụ, xây dựng ngữ cảnh qua các lượt.

Xử lý Lỗi và Thử Lại

Các công cụ có thể thất bại (lỗi API, đầu vào không hợp lệ). Trả về lỗi có cấu trúc:

def execute_tool(tool_name, tool_input):
    try:
        if tool_name == "web_search":
            # Simulate API call
            if "error" in tool_input.get("query", ""):
                raise ValueError("Invalid query")
            return "Results: Claude 3.5 excels in coding."
        # ...
    except Exception as e:
        return f"Tool error: {str(e)}. Please retry or adjust."

Cải thiện vòng lặp với thử lại:

max_retries = 3
for retry in range(max_retries):
    try:
        tool_results = handle_parallel_tools(response)
        break
    except Exception as e:
        error_result = [{"type": "tool_result", "content": f"Execution failed: {e}"}]
        messages.append({"role": "user", "content": error_result})

Claude suy luận qua lỗi: "Công cụ thất bại do đầu vào không hợp lệ; hãy để tôi tinh chỉnh."

Thêm timeout:

future = executor.submit(execute_tool, ...)
result = future.result(timeout=30)

Ví dụ Thực tế: Trợ lý Nghiên cứu Tự động

Kết hợp thành một agent nghiên cứu tích hợp tìm kiếm Serper.dev giả lập và tóm tắt:

Định nghĩa các công cụ thực tế hơn:

# Mock Serper integration
def real_web_search(query):
    # In production: requests.post('https://google.serper.dev/search', json={'q': query})
    return f"Top results for '{query}': 1. Anthropic releases Claude 3.5. 2. Python SDK v0.10."

tools[0]['input_schema']  # As before

# Run agent
research = agent_loop("Research latest on Claude API function calling: search updates, summarize changes, check Python examples.")
print(research)
# Output: Claude 3.5 Sonnet supports parallel tools up to 10x faster. Key changes: ...

Số từ cho đến nay: ~1200. Mở rộng cho production:

• Logging: Sử dụng structlog để theo dõi các cuộc gọi công cụ.
• Quản lý Trạng thái: Lưu trữ tin nhắn trong SQLite cho các agent chạy lâu.
• Giới hạn Tốc độ: Triển khai backoff hàm mũ với tenacity.
• Xác thực: Pydantic cho đầu vào/đầu ra công cụ.

from pydantic import BaseModel

class SearchInput(BaseModel):
    query: str

# In execute_tool
input_model = SearchInput(**tool_input)

Mở rộng quy mô đến enterprise: Tích hợp với MCP servers cho các công cụ mở rộng hoặc n8n cho workflow.

Các Thực hành Tốt nhất và Mẹo

• Prompt Engineering: Sử dụng thẻ XML: <thinking>Reason first</thinking><tool_calls>...</tool_calls>.
• Lựa chọn Model: Sonnet cho tốc độ/cân bằng; Opus cho suy luận phức tạp.
• Hiệu quả Token: Giới hạn công cụ ở 10-20; tóm tắt lịch sử.
• Bảo mật: Xác thực/làm sạch đầu vào công cụ để ngăn chặn injection.
• Kiểm thử: Unit test execute_tool; mock phản hồi API.
• Giám sát: Theo dõi tỷ lệ thành công công cụ, độ trễ.
• So sánh: Claude vượt trội GPT-4o về độ chính xác công cụ (theo benchmark của Anthropic).

Kết thúc

Với orchestrator Python này, bạn đã sẵn sàng xây dựng các agent Claude cấp production. Thử nghiệm với các công cụ của bạn—mở rộng đến Slack bots, tự động hóa Zapier hoặc pipeline AI đầy đủ. Kiểm tra tài liệu Anthropic để cập nhật: docs.anthropic.com/en/docs/tool-use.

Fork code trên GitHub (link trong comments) và chia sẻ agent của bạn trên diễn đàn Claude Directory!

(~1450 từ)