zalo-icon
facebook-icon
phone-icon
Dự án AI Agent đầu tiên bằng Python: Từ Tool Calling đến vòng lặp có kiểm soát

AI Agent bằng Python là chủ đề trọng tâm của bài viết này. Cách nhanh nhất để hiểu Agent là tự xây một phiên bản nhỏ mà không che giấu logic sau Framework. Dự án trong bài nhận câu hỏi doanh thu, cho Model chọn một Tool giả lập, kiểm tra Arguments, thực thi hàm và trả kết quả. Mục tiêu là hiểu vòng đời, không phải tạo sản phẩm hoàn chỉnh.

Bạn cần biết Python cơ bản, hàm, Dictionary, JSON và cách dùng biến môi trường. Ví dụ dùng cấu trúc gần với các API Tool Calling phổ biến. Tên Model và SDK có thể thay đổi, nhưng kiến trúc không phụ thuộc một nhà cung cấp.

Mục lục

  1. Mục tiêu dự án
  2. Cấu trúc thư mục
  3. Quản lý API Key
  4. Tạo dữ liệu giả lập
  5. Viết Tool
  6. Khai báo Tool Schema
  7. Gọi Model
  8. Thực thi Tool an toàn
  9. Hoàn thiện Agent Loop
  10. Logging và Task ID
  11. Giới hạn số bước
  12. Xử lý lỗi
  13. Viết Test
  14. Nâng cấp dự án
  15. Checklist
  16. Kết luận
  17. Tài liệu tham khảo

Mục tiêu dự án

Agent nhận câu hỏi như "Doanh thu ngày 2026-09-14 của Q7 là bao nhiêu?". Model không được tự bịa số. Nó phải gọi get_sales. Runtime kiểm tra ngày và chi nhánh, hàm trả dữ liệu giả lập, sau đó Model giải thích kết quả.

Phiên bản đầu chỉ có một Read Tool. Đây là lựa chọn có chủ đích để tập trung vào Tool Loop. Chưa cần RAG, Memory hoặc Multi-Agent.

Cấu trúc thư mục

sales-agent/
  app.py
  tools.py
  schemas.py
  tests/
    test_tools.py
  .env.example
  requirements.txt

tools.py chứa hàm nghiệp vụ. schemas.py chứa schema cung cấp cho Model. app.py điều phối. Tách Tool khỏi Prompt giúp kiểm thử độc lập.

Quản lý API Key

Không ghi API Key trực tiếp trong code hoặc commit lên Git. Dùng biến môi trường và cung cấp file .env.example chỉ chứa tên biến. Production nên dùng Secret Manager, phân quyền tối thiểu và Rotation.

import os

api_key = os.environ["LLM_API_KEY"]

Tạo dữ liệu giả lập

SALES = {
    ("2026-09-14", "Q7"): {
        "revenue": 520_000_000,
        "currency": "VND",
        "orders": 318
    },
    ("2026-09-14", "Thu Duc"): {
        "revenue": 410_000_000,
        "currency": "VND",
        "orders": 251
    }
}

Dữ liệu giả lập giúp học vòng lặp mà không cần Database. Khi thay bằng BigQuery hoặc PostgreSQL, hợp đồng Input và Output của Tool nên được giữ ổn định.

Viết Tool

from datetime import date

ALLOWED_BRANCHES = {"Q7", "Thu Duc"}

def get_sales(sales_date: str, branch: str) -> dict:
    date.fromisoformat(sales_date)

    if branch not in ALLOWED_BRANCHES:
        return {
            "status": "error",
            "error_code": "BRANCH_NOT_ALLOWED"
        }

    row = SALES.get((sales_date, branch))
    if row is None:
        return {
            "status": "success",
            "data": None,
            "message": "No data"
        }

    return {
        "status": "success",
        "data": row,
        "metadata": {
            "sales_date": sales_date,
            "branch": branch,
            "timezone": "Asia/Ho_Chi_Minh"
        }
    }

Validation nằm trong Tool, không phụ thuộc Model. Dữ liệu rỗng khác lỗi hệ thống nên có trạng thái riêng. Tool không trả Stack Trace hoặc bí mật nội bộ cho Model.

Tool Schema trong Python cho AI Agent

Khai báo Tool Schema

GET_SALES_SCHEMA = {
    "name": "get_sales",
    "description": "Lấy doanh thu đã ghi nhận của một chi nhánh trong một ngày",
    "parameters": {
        "type": "object",
        "properties": {
            "sales_date": {
                "type": "string",
                "format": "date"
            },
            "branch": {
                "type": "string",
                "enum": ["Q7", "Thu Duc"]
            }
        },
        "required": ["sales_date", "branch"],
        "additionalProperties": False
    }
}

Enum giúp Model chỉ chọn Branch được hỗ trợ trong Demo. Production không nên đưa danh sách hàng nghìn Branch vào schema. Tool Catalog hoặc Lookup Tool sẽ phù hợp hơn.

Gọi Model

messages = [
    {
        "role": "system",
        "content": (
            "Bạn là trợ lý doanh thu. "
            "Luôn dùng Tool cho câu hỏi cần số liệu. "
            "Không suy đoán khi không có dữ liệu."
        )
    },
    {
        "role": "user",
        "content": user_question
    }
]

response = client.create_response(
    model=MODEL_NAME,
    messages=messages,
    tools=[GET_SALES_SCHEMA]
)

Tên Method phụ thuộc SDK. Điều cần hiểu là Request chứa Messages và Tool Definitions. Response có thể là Final Text hoặc Tool Call.

Thực thi Tool an toàn

TOOL_REGISTRY = {
    "get_sales": get_sales
}

def execute_tool(call):
    tool = TOOL_REGISTRY.get(call.name)
    if tool is None:
        return {
            "status": "error",
            "error_code": "UNKNOWN_TOOL"
        }

    args = validate_arguments(call.arguments)
    return tool(**args)

Không dùng eval để thực thi tên hàm do Model tạo. Registry là Allowlist. validate_arguments parse JSON, kiểm tra schema và giới hạn kích thước.

Vòng lặp AI Agent bằng Python

Hoàn thiện Agent Loop

def run_agent(question: str) -> str:
    messages = build_initial_messages(question)

    for step in range(MAX_STEPS):
        response = call_model(messages)

        if response.type == "final":
            return response.text

        for call in response.tool_calls:
            result = execute_tool(call)
            messages.append(response.as_message())
            messages.append({
                "role": "tool",
                "tool_call_id": call.id,
                "content": json.dumps(result)
            })

    raise RuntimeError("Agent exceeded MAX_STEPS")

Agent Loop có giới hạn rõ. Response của Assistant và Tool Result phải giữ đúng Call ID để Model biết kết quả thuộc lời gọi nào. Với nhiều Tool Calls, Runtime có thể chạy song song nếu độc lập.

Logging và Unit Test cho AI Agent

Logging và Task ID

Mỗi lần chạy tạo Task ID. Log nên có Step, Model, duration, token usage, Tool Name, trạng thái và error code. Không log API Key hoặc toàn bộ dữ liệu nhạy cảm.

logger.info({
    "task_id": task_id,
    "step": step,
    "event": "tool_completed",
    "tool": call.name,
    "duration_ms": duration_ms,
    "status": result["status"]
})

Giới hạn số bước

MAX_STEPS ngăn vòng lặp vô hạn. Production còn cần Max Tool Calls, Token Budget, Cost Budget và Deadline. Nếu vượt giới hạn, Agent trả trạng thái không hoàn thành kèm lý do thay vì tạo câu trả lời đoán.

Xử lý lỗi

Validation Error có thể đưa về Model để sửa Arguments. Permission Error phải dừng. Timeout của Read Tool có thể retry giới hạn. Timeout của Write Tool cần Idempotency vì hành động có thể đã xảy ra.

Lỗi nên có error_code ổn định cho chương trình và message an toàn cho Model. Không phụ thuộc vào việc Model đọc một câu lỗi tự do.

Viết Test

Unit Test Tool với ngày đúng, ngày sai, Branch bị chặn và dữ liệu rỗng. Integration Test giả lập Model Response để kiểm tra Loop. Evaluation Test dùng câu hỏi tự nhiên để đo Model chọn đúng Tool và Arguments.

def test_get_sales_q7():
    result = get_sales("2026-09-14", "Q7")
    assert result["status"] == "success"
    assert result["data"]["revenue"] == 520_000_000

def test_reject_unknown_branch():
    result = get_sales("2026-09-14", "Unknown")
    assert result["error_code"] == "BRANCH_NOT_ALLOWED"

Nâng cấp dự án

  • Thay dữ liệu giả bằng Data Service Read-only.
  • Thêm Tool lấy tồn kho và so sánh doanh thu.
  • Lưu Trace vào Observability Backend.
  • Tạo 50 câu hỏi Evaluation có Expected Tool và Arguments.
  • Thêm User Identity và Row-level Security.
  • Thêm Structured Output cho Final Answer.
  • Chỉ sau đó mới thử Write Tool có Approval.

Checklist

  • Secret không nằm trong code.
  • Tool Registry là Allowlist.
  • Arguments được validate trước thực thi.
  • Tool Result có schema và metadata.
  • Agent Loop có Max Steps.
  • Log gắn Task ID nhưng không lộ bí mật.
  • Test tách Tool, Loop và Model behavior.

Kết luận

Dự án nhỏ này cho thấy Agent không nằm trong một dòng gọi Model. Nó nằm ở vòng lặp kết nối Model, Tool, Validation, State và giới hạn thực thi. Model đề xuất hành động, còn Runtime giữ quyền kiểm soát.

Hãy làm phiên bản một Tool chạy ổn định và có Test trước khi thêm Framework. Khi hiểu từng bước, bạn có thể dùng SDK hoặc Agent Framework mà vẫn biết dữ liệu đang đi đâu và lỗi cần sửa ở lớp nào.

Tài liệu tham khảo

TechData.AI - Leading The Future.

Hoàng Minh.

Comments are closed!

Scroll to Top