Function Calling实战：构建可执行的任务分解Agent

Trong bài viết này, tôi sẽ chia sẻ kinh nghiệm thực chiến khi xây dựng một Task Decomposition Agent sử dụng Function Calling — hệ thống có thể tự động phân tách công việc phức tạp thành các bước nhỏ có thể thực thi được. Đây là kỹ thuật nền tảng mà tôi đã áp dụng thành công trong nhiều dự án production.

Tại sao Function Calling quan trọng?

Function Calling cho phép LLM gọi các hàm được định nghĩa sẵn, tạo ra chuỗi hành động có thể kiểm soát. Thay vì chỉ trả về text, model có thể:

• Trigger database queries
• Thực thi API calls
• Tính toán và xử lý dữ liệu
• Điều phối sub-agents

So sánh chi phí các Model hỗ trợ Function Calling

Trước khi bắt đầu, hãy xem xét chi phí vận hành. Dưới đây là bảng so sánh giá 2026 đã được xác minh:

Model	Output Price ($/MTok)	10M tokens/tháng
GPT-4.1	$8.00	$80
Claude Sonnet 4.5	$15.00	$150
Gemini 2.5 Flash	$2.50	$25
DeepSeek V3.2	$0.42	$4.20

Với đăng ký tại đây trên HolyShehe AI, bạn được hưởng tỷ giá ¥1 = $1 — tiết kiệm tới 85%+ so với các provider khác. Ngoài ra còn hỗ trợ WeChat/Alipay, độ trễ <50ms, và tín dụng miễn phí khi đăng ký.

Kiến trúc Task Decomposition Agent

┌─────────────────────────────────────────────────────────┐
│                   USER REQUEST                          │
└─────────────────────┬───────────────────────────────────┘
                      ▼
┌─────────────────────────────────────────────────────────┐
│              TASK DECOMPOSER AGENT                       │
│  ┌─────────────────────────────────────────────────┐   │
│  │  1. Nhận yêu cầu phức tạp                       │   │
│  │  2. Phân tách thành sub-tasks                  │   │
│  │  3. Xác định dependencies                      │   │
│  │  4. Gọi Function Execution                      │   │
│  │  5. Tổng hợp kết quả                            │   │
│  └─────────────────────────────────────────────────┘   │
└─────────────────────┬───────────────────────────────────┘
                      ▼
┌─────────────────────────────────────────────────────────┐
│              FUNCTION REGISTRY                          │
│  • search_web()     • calculate()                       │
│  • fetch_data()     • save_result()                     │
│  • validate()       • notify_user()                     │
└─────────────────────────────────────────────────────────┘

Code Implementation đầy đủ

1. Định nghĩa Functions Schema

import openai
import json
from typing import List, Dict, Any, Optional

Cấu hình HolySheep AI - KHÔNG dùng api.openai.com
client = openai.OpenAI(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1"
)

Định nghĩa functions cho task decomposition
FUNCTIONS = [
    {
        "name": "decompose_task",
        "description": "Phân tách công việc phức tạp thành các bước nhỏ",
        "parameters": {
            "type": "object",
            "properties": {
                "task": {"type": "string", "description": "Công việc cần phân tách"},
                "context": {"type": "string", "description": "Bối cảnh bổ sung"}
            },
            "required": ["task"]
        }
    },
    {
        "name": "execute_step",
        "description": "Thực thi một bước cụ thể trong kế hoạch",
        "parameters": {
            "type": "object",
            "properties": {
                "step_id": {"type": "string", "description": "ID của bước"},
                "action": {"type": "string", "description": "Hành động cần thực hiện"},
                "params": {"type": "object", "description": "Tham số cho hành động"}
            },
            "required": ["step_id", "action"]
        }
    },
    {
        "name": "search_information",
        "description": "Tìm kiếm thông tin trên web",
        "parameters": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Câu truy vấn tìm kiếm"},
                "max_results": {"type": "integer", "default": 5}
            },
            "required": ["query"]
        }
    },
    {
        "name": "calculate",
        "description": "Thực hiện phép tính toán",
        "parameters": {
            "type": "object",
            "properties": {
                "operation": {"type": "string", "enum": ["add", "subtract", "multiply", "divide", "custom"]},
                "values": {"type": "array", "items": {"type": "number"}},
                "formula": {"type": "string", "description": "Công thức tùy chỉnh"}
            },
            "required": ["operation", "values"]
        }
    },
    {
        "name": "aggregate_results",
        "description": "Tổng hợp kết quả từ nhiều sub-tasks",
        "parameters": {
            "type": "object",
            "properties": {
                "results": {"type": "array", "description": "Danh sách kết quả"},
                "strategy": {"type": "string", "enum": ["merge", "average", "concatenate", "custom"]}
            },
            "required": ["results"]
        }
    }
]

def execute_function(function_name: str, arguments: dict) -> dict:
    """Thực thi function được gọi từ model"""
    
    if function_name == "decompose_task":
        return decompose_task_handler(arguments)
    elif function_name == "execute_step":
        return execute_step_handler(arguments)
    elif function_name == "search_information":
        return search_information_handler(arguments)
    elif function_name == "calculate":
        return calculate_handler(arguments)
    elif function_name == "aggregate_results":
        return aggregate_results_handler(arguments)
    else:
        return {"error": f"Unknown function: {function_name}"}

def decompose_task_handler(args: dict) -> dict:
    """Xử lý phân tách công việc"""
    task = args["task"]
    context = args.get("context", "")
    
    # Logic phân tách công việc
    steps = [
        {"id": "step_1", "action": "analyze", "description": "Phân tích yêu cầu"},
        {"id": "step_2", "action": "search_information", "params": {"query": task}},
        {"id": "step_3", "action": "calculate", "params": {"operation": "custom"}},
        {"id": "step_4", "action": "aggregate_results", "params": {"strategy": "merge"}}
    ]
    
    return {
        "task": task,
        "context": context,
        "steps": steps,
        "estimated_steps": len(steps)
    }

def execute_step_handler(args: dict) -> dict:
    """Thực thi một bước cụ thể"""
    step_id = args["step_id"]
    action = args["action"]
    params = args.get("params", {})
    
    return {
        "step_id": step_id,
        "status": "completed",
        "action": action,
        "result": f"Step {step_id} executed successfully",
        "output": {"data": "sample_output"}
    }

def search_information_handler(args: dict) -> dict:
    """Tìm kiếm thông tin"""
    query = args["query"]
    max_results = args.get("max_results", 5)
    
    # Mock search results
    return {
        "query": query,
        "results": [
            {"title": f"Result {i}", "snippet": f"Sample snippet {i}"}
            for i in range(min(max_results, 5))
        ],
        "count": min(max_results, 5)
    }

def calculate_handler(args: dict) -> dict:
    """Xử lý phép tính"""
    operation = args["operation"]
    values = args["values"]
    
    if operation == "add":
        result = sum(values)
    elif operation == "subtract":
        result = values[0] - sum(values[1:])
    elif operation == "multiply":
        result = 1
        for v in values:
            result *= v
    elif operation == "divide":
        result = values[0] / values[1] if values[1] != 0 else "Error"
    else:
        result = "Custom operation"
    
    return {"operation": operation, "input": values, "result": result}

def aggregate_results_handler(args: dict) -> dict:
    """Tổng hợp kết quả"""
    results = args["results"]
    strategy = args.get("strategy", "merge")
    
    return {
        "strategy": strategy,
        "aggregated": f"Combined {len(results)} results",
        "final_output": {"status": "success", "data": results}
    }

2. Main Agent Loop với Streaming

import time

class TaskDecompositionAgent:
    def __init__(self, model: str = "gpt-4.1"):
        self.client = client
        self.model = model
        self.conversation_history = []
        self.execution_log = []
    
    def run(self, user_request: str, stream: bool = True) -> dict:
        """Chạy agent với request từ user"""
        
        self.conversation_history.append({
            "role": "user",
            "content": user_request
        })
        
        # Đo thời gian phản hồi
        start_time = time.time()
        
        response = self.client.chat.completions.create(
            model=self.model,
            messages=self.conversation_history,
            tools=[{"type": "function", "function": f} for f in FUNCTIONS],
            tool_choice="auto",
            stream=stream
        )
        
        if stream:
            return self._handle_stream_response(response, start_time)
        else:
            return self._handle_response(response, start_time)
    
    def _handle_stream_response(self, response, start_time: float) -> dict:
        """Xử lý streaming response"""
        full_content = ""
        tool_calls = []
        current_tool_call = None
        
        for chunk in response:
            delta = chunk.choices[0].delta
            
            if delta.content:
                full_content += delta.content
            
            if delta.tool_calls:
                for tc in delta.tool_calls:
                    if current_tool_call is None:
                        current_tool_call = {
                            "id": tc.id,
                            "name": tc.function.name,
                            "arguments": ""
                        }
                    else:
                        current_tool_call["arguments"] += tc.function.arguments
        
        elapsed_ms = (time.time() - start_time) * 1000
        
        return {
            "content": full_content,
            "elapsed_ms": round(elapsed_ms, 2),
            "status": "streaming_complete"
        }
    
    def _handle_response(self, response, start_time: float) -> dict:
        """Xử lý non-streaming response"""
        message = response.choices[0].message
        elapsed_ms = (time.time() - start_time) * 1000
        
        result = {
            "content": message.content,
            "elapsed_ms": round(elapsed_ms, 2),
            "tool_calls": []
        }
        
        # Xử lý function calls
        if message.tool_calls:
            for tool_call in message.tool_calls:
                func_name = tool_call.function.name
                func_args = json.loads(tool_call.function.arguments)
                
                print(f"🔧 Calling function: {func_name}")
                print(f"   Arguments: {json.dumps(func_args, indent=2, ensure_ascii=False)}")
                
                # Thực thi function
                func_result = execute_function(func_name, func_args)
                
                self.execution_log.append({
                    "function": func_name,
                    "arguments": func_args,
                    "result": func_result,
                    "timestamp": time.time()
                })
                
                # Thêm kết quả vào conversation
                self.conversation_history.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": json.dumps(func_result, ensure_ascii=False)
                })
                
                result["tool_calls"].append({
                    "name": func_name,
                    "arguments": func_args,
                    "result": func_result
                })
        
        return result
    
    def run_multi_step(self, user_request: str, max_iterations: int = 10) -> dict:
        """Chạy nhiều bước cho đến khi hoàn thành"""
        
        iteration = 0
        final_results = []
        
        while iteration < max_iterations:
            iteration += 1
            print(f"\n{'='*50}")
            print(f"📍 Iteration {iteration}/{max_iterations}")
            print(f"{'='*50}")
            
            result = self.run(user_request, stream=False)
            
            if not result["tool_calls"]:
                # Không có function call -> kết thúc
                final_results.append(result)
                break
            
            final_results.extend(result["tool_calls"])
        
        return {
            "iterations": iteration,
            "results": final_results,
            "execution_log": self.execution_log
        }


Demo sử dụng
if __name__ == "__main__":
    agent = TaskDecompositionAgent(model="gpt-4.1")
    
    # Test request
    test_request = """
    Phân tích xu hướng giá vàng tháng 3/2026 và đưa ra dự đoán.
    Bao gồm:
    1. Tìm kiếm thông tin giá vàng
    2. Tính toán chỉ số trung bình
    3. So sánh với tháng trước
    4. Đưa ra dự đoán
    """
    
    result = agent.run_multi_step(test_request)
    
    print(f"\n✅ Hoàn thành trong {result['iterations']} iterations")
    print(f"📊 Execution log: {len(result['execution_log'])} functions called")

Chi phí thực tế khi sử dụng HolySheep AI

Giả sử một Task Decomposition Agent xử lý 10M tokens/tháng với cấu hình:

• Input: 6M tokens/tháng
• Output (bao gồm function calls): 4M tokens/tháng

Provider	Input ($/MTok)	Output ($/MTok)	Tổng/tháng	HolySheep savings
OpenAI (GPT-4.1)	$2	$8	$44	-
Anthropic (Claude Sonnet 4.5)	$3	$15	$66	-
Google (Gemini 2.5 Flash)	$1.25	$2.50	$14.50	-
DeepSeek V3.2	$0.28	$0.42	$3.64	-
HolySheep AI	¥0.28	¥0.42	¥3.64 (~$3.64)	85%+ vs OpenAI

Best Practices từ kinh nghiệm thực chiến

Qua nhiều dự án production, tôi rút ra một số nguyên tắc quan trọng:

# 1. Function naming - sử dụng snake_case rõ ràng
✅ Tốt: calculate_average_price, fetch_user_data
❌ Xấu: calc, getData

2. Error handling cho function execution
def safe_execute_function(func_name: str, args: dict, max_retries: int = 3):
    for attempt in range(max_retries):
        try:
            result = execute_function(func_name, args)
            return {"success": True, "data": result}
        except Exception as e:
            if attempt == max_retries - 1:
                return {"success": False, "error": str(e)}
            time.sleep(1 * (attempt + 1))  # Exponential backoff
    
3. Timeout cho long-running functions
import signal

def timeout_handler(signum, frame):
    raise TimeoutError("Function execution timed out")

def execute_with_timeout(func, args, timeout_seconds=30):
    signal.signal(signal.SIGALRM, timeout_handler)
    signal.alarm(timeout_seconds)
    try:
        result = func(args)
        signal.alarm(0)
        return result
    except TimeoutError:
        return {"error": "Execution timeout", "status": "timeout"}

Lỗi thường gặp và cách khắc phục

1. Lỗi "Invalid API Key" hoặc Authentication Error

# ❌ Sai - sử dụng endpoint không đúng
client = openai.OpenAI(
    api_key="YOUR_KEY",
    base_url="https://api.openai.com/v1"  # ❌ SAI
)

✅ Đúng - sử dụng HolySheep endpoint
client = openai.OpenAI(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1"  # ✅ ĐÚNG
)

Khắc phục: Kiểm tra lại API key từ dashboard HolySheep AI và đảm bảo base_url chính xác. Nếu dùng biến môi trường, hãy đặt HOLYSHEEP_API_KEY thay vì OPENAI_API_KEY.

2. Lỗi "Function calling is not supported for this model"

# ❌ Sai - model không hỗ trợ function calling đầy đủ
response = client.chat.completions.create(
    model="gpt-3.5-turbo",  # ❌ Hạn chế
    messages=messages,
    tools=functions,
    tool_choice="auto"
)

✅ Đúng - sử dụng model được kiểm chứng
response = client.chat.completions.create(
    model="gpt-4.1",  # ✅ Hỗ trợ đầy đủ
    messages=messages,
    tools=functions,
    tool_choice="auto"
)

Hoặc sử dụng Claude trên HolySheep
response = client.messages.create(
    model="claude-sonnet-4-20250514",  # ✅ Hỗ trợ function calling
    max_tokens=1024,
    messages=messages,
    tools=functions
)

Khắc phục: Kiểm tra documentation của HolySheep AI để biết model nào hỗ trợ function calling. GPT-4.1 và Claude Sonnet 4.5 trên HolySheep đều hỗ trợ đầy đủ.

3. Lỗi "Tool call parsing failed" hoặc JSONDecodeError

# ❌ Sai - không validate JSON trước khi parse
func_args = json.loads(tool_call.function.arguments)

✅ Đúng - validate và handle exception
def safe_parse_json(arguments: str) -> dict:
    try:
        return json.loads(arguments)
    except json.JSONDecodeError as e:
        # Log lỗi chi tiết
        print(f"JSON Parse Error: {e}")
        print(f"Raw arguments: {arguments}")
        # Fallback: thử sửa lỗi common
        # Thay thế single quotes bằng double quotes
        fixed = arguments.replace("'", '"')
        try:
            return json.loads(fixed)
        except:
            return {"error": "Invalid JSON", "raw": arguments}

func_args = safe_parse_json(tool_call.function.arguments)

✅ Hoặc sử dụng pydantic để validate schema
from pydantic import BaseModel, ValidationError

class FunctionArgs(BaseModel):
    task: str
    context: Optional[str] = None

def validate_and_execute(func_def: dict, raw_args: str):
    try:
        args_dict = json.loads(raw_args)
        validated = FunctionArgs(**args_dict)
        return execute_function(func_def["name"], validated.dict())
    except ValidationError as e:
        return {"error": "Validation failed", "details": e.errors()}

Khắc phục: Luôn wrap JSON parsing trong try-except, implement fallback strategy, và log raw arguments để debug. Sử dụng Pydantic để validate schema tự động.

4. Lỗi "Maximum iterations exceeded" - Infinite Loop

# ❌ Sai - không giới hạn iterations
while True:
    response = agent.run(request)
    if not response.tool_calls:
        break

✅ Đúng - có max_iterations và timeout
import time

MAX_ITERATIONS = 10
MAX_TOTAL_TIME = 60  # seconds

def run_with_limits(agent, request):
    start_time = time.time()
    
    for i in range(MAX_ITERATIONS):
        elapsed = time.time() - start_time
        
        # Check timeout
        if elapsed > MAX_TOTAL_TIME:
            return {
                "status": "timeout",
                "iterations": i,
                "elapsed": elapsed,
                "error": "Total execution time exceeded"
            }
        
        # Check iterations
        if i >= MAX_ITERATIONS - 1:
            return {
                "status": "max_iterations",
                "iterations": i + 1,
                "elapsed": elapsed,
                "warning": "Maximum iterations reached"
            }
        
        response = agent.run(request, stream=False)
        
        if not response.get("tool_calls"):
            return {
                "status": "success",
                "iterations": i + 1,
                "elapsed": elapsed,
                "result": response
            }
    
    return {"status": "unknown", "iterations": MAX_ITERATIONS}

Khắc phục: Luôn đặt giới hạn iterations và timeout. Implement circuit breaker pattern để dừng khi phát hiện loop không tiến triển. Log mỗi iteration để debug.

Kết luận

Task Decomposition Agent với Function Calling là một công cụ cực kỳ mạnh mẽ để xây dựng hệ thống AI có thể thực thi tác vụ phức tạp một cách đáng tin cậy. Qua bài viết này, tôi đã chia sẻ:

• Kiến trúc và implementation đầy đủ
• So sánh chi phí giữa các providers với dữ liệu giá thực tế 2026
• 4 trường hợp lỗi thường gặp cùng mã khắc phục
• Best practices từ kinh nghiệm production

Với HolySheep AI, bạn không chỉ tiết kiệm 85%+ chi phí mà còn được hưởng độ trễ <50ms, thanh toán qua WeChat/Alipay, và tín dụng miễn phí khi đăng ký.

👉 Đăng ký HolySheep AI — nhận tín dụng miễn phí khi đăng ký