Agent trên API LLM không kiểm duyệt: Gọi hàm, Framework và Guardrails

Giải phẫu vòng lặp gọi hàm

Bỏ qua các framework và mọi agent đều thực hiện bốn việc giống nhau. Bạn gửi đoạn hội thoại cùng danh sách tools. Mô hình phản hồi bằng văn bản, nghĩa là nó đã xong, hoặc bằng tool_calls, nghĩa là nó muốn thực thi một thứ gì đó. Bạn chạy từng lệnh gọi và thêm một tin nhắn tool cho mỗi lệnh gọi, trích dẫn id của lệnh gọi đó. Sau đó, bạn hỏi lại mô hình với đoạn hội thoại dài hơn.

Tín hiệu cho nhánh là finish_reason. Khi nó bằng tool_calls, tin nhắn trợ lý có content được đặt thành null và một mảng tool_calls, mỗi mục chứa id, tên hàm và một chuỗi đối số dưới dạng JSON. trang tham chiếu cho thấy cấu trúc chính xác. Hai chi tiết dễ gây nhầm lẫn: các đối số là một chuỗi cần phải được phân tích cú pháp, và tin nhắn trợ lý chứa các lệnh gọi phải được giữ lại trong lịch sử trước các kết quả công cụ của bạn, theo đúng thứ tự.

Một vòng lặp hoàn chỉnh từ đầu

Script bên dưới định nghĩa một hàm, mô tả nó bằng lược đồ JSON và lặp cho đến khi mô hình tạo ra văn bản thuần túy. Nó sử dụng SDK Python chính thức trỏ đến API. Mỗi lần lặp đều đính kèm tin nhắn trợ lý, sau đó một tin nhắn tool cho mỗi lệnh gọi, và một bộ đếm bước sẽ ngăn chặn hành vi chạy không kiểm soát.

Copyimport json, os
from openai import OpenAI

client = OpenAI(base_url="https://api.uncensoredapis.com/v1", api_key=os.environ["API_KEY"])

TICKETS = {"T-101": "open", "T-102": "closed"}

def ticket_status(ticket_id: str) -> str:
    return TICKETS.get(ticket_id, "unknown ticket")

TOOLS = [{
    "type": "function",
    "function": {
        "name": "ticket_status",
        "description": "Look up the status of a support ticket by id.",
        "parameters": {
            "type": "object",
            "properties": {"ticket_id": {"type": "string"}},
            "required": ["ticket_id"],
        },
    },
}]
REGISTRY = {"ticket_status": ticket_status}

messages = [{"role": "user", "content": "Are tickets T-101 and T-102 still open?"}]

for step in range(6):                      # hard cap on loop turns
    r = client.chat.completions.create(
        model="uncensored", messages=messages, tools=TOOLS, max_tokens=400)
    msg = r.choices[0].message
    messages.append(msg.model_dump(exclude_none=True))
    if r.choices[0].finish_reason != "tool_calls":
        print(msg.content)
        break
    for call in msg.tool_calls:
        fn = REGISTRY.get(call.function.name)
        try:
            args = json.loads(call.function.arguments)
            result = fn(**args) if fn else f"error: unknown tool {call.function.name}"
        except Exception as exc:
            result = f"error: {exc}"
        messages.append({"role": "tool", "tool_call_id": call.id, "content": str(result)})
else:
    print("stopped: step limit reached")

Hãy chú ý những gì vòng lặp làm khi gặp lỗi: các lỗi trở thành kết quả công cụ dưới dạng chuỗi. Điều này cho phép mô hình đọc vấn đề và thử lại, ví dụ như sửa một id vé không hợp lệ, thay vì chương trình của bạn bị treo. Mệnh đề for ... else chỉ kích hoạt nếu đạt ngưỡng mà không có câu trả lời cuối cùng, đây là cách bạn phát hiện mô hình bị lặp lại/vòng lặp.

Giữ prompt tập trung. Một system message ngắn nêu rõ khi nào thì gọi công cụ và khi nào thì trả lời trực tiếp sẽ giảm thiểu các lệnh gọi không cần thiết, và mỗi lệnh gọi bị bỏ qua sẽ tiết kiệm một lượt đi, token và một slot trong giới hạn tốc độ của bạn.

Bảo vệ việc thực thi công cụ

Đầu ra của mô hình là đầu vào không đáng tin cậy. Tên công cụ hoặc đối số do mô hình chọn có thể sai, kỳ lạ hoặc bị ảnh hưởng bởi văn bản mà mô hình đã đọc từ một trang web hoặc người dùng. Hãy coi mọi lệnh gọi như một yêu cầu đến từ internet công cộng.

• Danh sách cho phép theo tên. Tra cứu hàm trong cơ sở dữ liệu bạn đã xây dựng. Không bao giờ dùng chuỗi để import hoặc đánh giá mã.
• Tách quyền đọc và ghi. Công cụ chỉ đọc chạy tự do. Bất kỳ thứ gì gửi email, tiêu tiền, xóa dữ liệu hoặc chạm vào tệp cần sự phê duyệt rõ ràng của con người.
• Xác thực đối số dựa trên lược đồ bạn đã công bố: khóa bắt buộc, không có khóa thừa, kiểu dữ liệu và giới hạn độ dài.
• Giới hạn thời gian và đầu ra. Áp dụng thời gian chờ và cắt ngắn kết quả trước khi đưa lại để đầu ra lớn của công cụ không chiếm hết cửa sổ ngữ cảnh 100.000 token của bạn.
• Giới hạn số lần lặp và chi phí. Giới hạn số bước mỗi tác vụ và token mỗi phiên, sử dụng khối usage.
• Ghi log mọi lệnh gọi với tên, đối số và kết quả để bạn có thể kiểm toán hành vi.

Copyimport json
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutTimeout

ALLOWED = {"ticket_status", "lookup_invoice"}      # read-only tools only
NEEDS_APPROVAL = {"refund_order"}                  # side effects wait for a human
POOL = ThreadPoolExecutor(max_workers=4)

def validate(args: dict, schema: dict) -> None:
    props = schema["properties"]
    for key in schema.get("required", []):
        if key not in args:
            raise ValueError(f"missing argument: {key}")
    for key, val in args.items():
        if key not in props:
            raise ValueError(f"unexpected argument: {key}")
        if props[key]["type"] == "string" and not isinstance(val, str):
            raise ValueError(f"{key} must be a string")
        if isinstance(val, str) and len(val) > 200:
            raise ValueError(f"{key} too long")

def run_tool(name, raw_args, registry, schemas, approve=lambda n, a: False):
    if name in NEEDS_APPROVAL:
        args = json.loads(raw_args)
        if not approve(name, args):
            return "denied: this action needs human approval"
    elif name not in ALLOWED:
        return f"error: tool {name} is not permitted"
    args = json.loads(raw_args)
    validate(args, schemas[name]["parameters"])
    future = POOL.submit(registry[name], **args)
    try:
        return str(future.result(timeout=5))[:2000]    # time and size limits
    except FutTimeout:
        return "error: tool timed out"

Hãy coi kết quả công cụ cũng không đáng tin cậy. Nếu một công cụ trả về văn bản được thu thập có chứa hướng dẫn, mô hình có thể làm theo chúng. Bao bọc nội dung như vậy bằng các dấu phân cách rõ ràng trong tin nhắn công cụ và nêu trong system prompt của bạn rằng đầu ra công cụ là dữ liệu, không phải lệnh.

Viết lược đồ công cụ mà mô hình có thể sử dụng

Schema là tài liệu duy nhất mô hình thấy, vì vậy nó xứng đáng được chăm chút giống như một API công khai. Đặt tên các hàm bằng động từ và danh từ phù hợp với nhiệm vụ, chẳng hạn như ticket_status hoặc lookup_invoice, thay vì các nhãn chung chung như run hoặc helper. Viết description dưới dạng hướng dẫn: hàm làm gì, trả về gì và khi nào không nên sử dụng nó.

• Giữ tham số ít và phẳng. Hai hoặc ba trường đơn giản được gọi chính xác thường xuyên hơn nhiều so với một đối tượng lồng ghép có mười trường.
• Sử dụng enums cho các tập hợp đóng, ví dụ "enum": ["open", "closed"], để mô hình không thể tự tạo ra một giá trị.
• Mô tả từng thuộc tính bằng một đơn vị hoặc ví dụ: "ID vé chẳng hạn như T-101" tốt hơn một chuỗi trần.
• Trả về kết quả gọn gàng. Chuỗi hoặc các đối tượng JSON nhỏ hoạt động tốt nhất; các bản dump dài đốt ngữ cảnh cho ít lợi ích.
• Ưu tiên nhiều công cụ chuyên biệt thay vì một công cụ tổng hợp có cờ chế độ.

Khi một lệnh gọi không vượt qua xác thực, hãy trả về thông báo xác thực dưới dạng kết quả công cụ thay vì ném lỗi. Mô hình thường sửa lại lệnh gọi ở lượt tiếp theo vì nó có thể đọc được những gì đã sai. Nếu nó thất bại theo cách đó hai lần, hãy viết lại mô tả, không phải vòng lặp.

Cấu hình LangChain

Lớp ChatOpenAI của LangChain chấp nhận base_url, nên không cần gói nhà cung cấp tùy chỉnh. Đặt model thành uncensored, truyền khóa từ môi trường của bạn và liên kết công cụ với bind_tools. Cài đặt langchain-openai và langchain-core.

Copyimport os
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

@tool
def ticket_status(ticket_id: str) -> str:
    """Return the status of a support ticket."""
    return {"T-101": "open", "T-102": "closed"}.get(ticket_id, "unknown ticket")

llm = ChatOpenAI(
    model="uncensored",
    base_url="https://api.uncensoredapis.com/v1",
    api_key=os.environ["API_KEY"],
    temperature=0.3,
    max_tokens=400,
)
llm_with_tools = llm.bind_tools([ticket_status])

ai = llm_with_tools.invoke("What is the status of ticket T-101?")
for call in ai.tool_calls:
    print(call["name"], call["args"], "->", ticket_status.invoke(call["args"]))

Kết quả của invoke là một AIMessage có danh sách tool_calls chứa các tên và đối số đã được phân tích cú pháp, vì vậy bạn bỏ qua bước phân tích JSON thủ công. Để có một vòng lặp tự trị đầy đủ, hãy truyền cùng một đối tượng llm vào các hàm tạo agent của LangChain; endpoint chỉ cần tôn trọng gọi hàm, điều mà nó làm được. Nếu một agent dường như bỏ qua các công cụ của nó, hãy kiểm tra xem docstring và các gợi ý kiểu có tồn tại không, vì chúng trở thành schema mà mô hình thấy.

Cấu hình LlamaIndex

LlamaIndex cung cấp OpenAILike cho các máy chủ tuân theo định dạng hoàn thành trò chuyện. Ba cài đặt quan trọng. api_base trỏ đến API, is_chat_model=True chọn điểm cuối trò chuyện, và is_function_calling_model=True báo cho framework biết nó có thể gửi các công cụ. Đặt context_window thành 100000 để các bộ phân tách văn bản và bộ đệm bộ nhớ của nó tự điều chỉnh kích thước đúng cách. Cài đặt llama-index-llms-openai-like.

Copyimport os
from llama_index.llms.openai_like import OpenAILike
from llama_index.core.tools import FunctionTool

def ticket_status(ticket_id: str) -> str:
    """Return the status of a support ticket."""
    return {"T-101": "open", "T-102": "closed"}.get(ticket_id, "unknown ticket")

llm = OpenAILike(
    model="uncensored",
    api_base="https://api.uncensoredapis.com/v1",
    api_key=os.environ["API_KEY"],
    is_chat_model=True,
    is_function_calling_model=True,
    context_window=100000,
    max_tokens=400,
)

tool = FunctionTool.from_defaults(fn=ticket_status)
response = llm.predict_and_call([tool], "Check ticket T-102 for me.")
print(response)

Từ đây, bạn có thể truyền cùng một llm cho một worker agent hoặc một query engine. Hãy nhớ rằng các pipeline truy xuất có thể nhồi nhiều chunk vào một prompt; với cửa sổ ngữ cảnh 100.000 token được chia sẻ giữa prompt và kết quả, hãy giới hạn số lượng chunk được truy xuất và giữ max_tokens ở mức vừa phải.

Vận hành agent với ngân sách hạn chế

Một agent biến một tin nhắn người dùng thành nhiều yêu cầu, vì vậy bạn hãy lập kế hoạch dung lượng. Một nhiệm vụ với năm vòng công cụ tốn sáu lệnh gọi mô hình. Mười nhiệm vụ đồng thời do đó tạo ra sáu mươi yêu cầu trong một khoảng thời gian ngắn, và trần 300 yêu cầu mỗi phút đến nhanh hơn so với một ứng dụng trò chuyện. Đặt bộ điều tiết xô token từ sách hướng dẫn lỗi và giới hạn tốc độ ở phía trước client của bạn.

Ngữ cảnh phát triển qua mỗi vòng vì mỗi kết quả công cụ được nối thêm. Tóm tắt hoặc loại bỏ các kết quả công cụ cũ một khi chúng đã được sử dụng, và để mắt đến usage.prompt_tokens; khi nó tiến gần đến 50.000, hãy nén lịch sử. Về chi phí, hãy giả định một nhiệm vụ minh họa gồm 6 lệnh gọi trung bình 3.000 token prompt và 200 token kết quả: đó là 18.000 token đầu vào (khoảng $0,0045 ở mức $0,25 mỗi triệu) và 1.200 token đầu ra (khoảng $0,0012 ở mức $1,00 mỗi triệu). Những số lượng token này là các giả định cho các tổng số, không phải là các phép đo. Nếu bạn mới làm quen với endpoint, hãy bắt đầu với docs và tín dụng dùng thử miễn phí.

Kiểm thử và gỡ lỗi một agent

Agent thất bại theo những cách mà các bài kiểm thử đơn vị trên một hàm đơn lẻ bỏ sót, vì vậy hãy kiểm thử cuộc hội thoại. Ghi lại danh sách tin nhắn đầy đủ cho mọi lần chạy, bao gồm cả đối số và kết quả công cụ, và phát lại các lỗi từ nhật ký. Ba thói quen mang lại lợi ích nhanh chóng.

1. Giả lập mô hình. Tạm dừng client để nó trả về các tin nhắn tool_calls đã được kịch bản hóa. Điều này kiểm tra registry, xác thực và việc kết thúc vòng lặp mà không cần mạng và không tốn chi phí.
2. Ghim việc lấy mẫu. Sử dụng temperature thấp, chẳng hạn như 0,2, cho các nhiệm vụ dựa trên công cụ. Độ ngẫu nhiên thấp hơn mang lại JSON đối số có thể lặp lại hơn, và bạn có thể tăng nó sau đó cho các bước nặng về văn bản.
3. Kiểm tra trên các quỹ đạo. Đối với một số ít prompt cố định, hãy kiểm tra xem công cụ nào đã được gọi và theo thứ tự nào, không chỉ văn bản cuối cùng.

Hãy theo dõi finish_reason. Giá trị length ở giữa một lệnh gọi công cụ có nghĩa là đối số đã bị cắt; hãy tăng max_tokens (lên đến 32.000) và JSON sẽ được hoàn thành. Truyền phát (streaming) cũng hoạt động với công cụ, nhưng bạn cần tích lũy các đoạn đối số theo chỉ mục lệnh gọi và chỉ phân tích cú pháp sau khi luồng đóng. Để ôn lại về các chunk đó, xem phần truyền phát (streaming) trong tài liệu tham khảo.