Tác nhân AI bền vững với Gemini và Temporal

Hướng dẫn này hướng dẫn bạn cách xây dựng một vòng lặp có tác nhân theo kiểu ReAct sử dụng Gemini API để suy luận và Temporal để duy trì. Bạn có thể xem toàn bộ mã nguồn của hướng dẫn này trên GitHub.

Trợ lý có thể gọi các công cụ, chẳng hạn như tra cứu cảnh báo thời tiết hoặc xác định vị trí địa lý của địa chỉ IP và sẽ lặp lại cho đến khi có đủ thông tin để phản hồi.

Điểm khác biệt giữa bản minh hoạ này và bản minh hoạ tác nhân thông thường là độ bền. Mọi lệnh gọi LLM, mọi lệnh gọi công cụ và mọi bước của vòng lặp dựa trên tác nhân đều được Temporal duy trì. Nếu quy trình gặp sự cố, mạng bị ngắt hoặc API hết thời gian chờ, Temporal sẽ tự động thử lại và tiếp tục từ bước đã hoàn tất gần đây nhất. Không có nhật ký cuộc trò chuyện nào bị mất và không có lệnh gọi công cụ nào bị lặp lại không chính xác.

Kiến trúc

Cấu trúc này bao gồm 3 phần:

  • Quy trình công việc: Vòng lặp có tác nhân điều phối logic thực thi.
  • Hoạt động: Các đơn vị công việc riêng lẻ (lệnh gọi LLM, lệnh gọi công cụ) mà Temporal duy trì.
  • Worker: Quy trình thực thi quy trình công việc và hoạt động.

Trong ví dụ này, bạn sẽ đặt cả 3 phần này vào một tệp duy nhất (durable_agent_worker.py). Trong quá trình triển khai thực tế, bạn sẽ tách chúng ra để có nhiều lợi thế về việc triển khai và khả năng mở rộng. Bạn sẽ đặt mã cung cấp lời nhắc cho tác nhân trong tệp thứ hai (start_workflow.py).

Điều kiện tiên quyết

Để hoàn tất hướng dẫn này, bạn cần:

  • Khoá Gemini API. Bạn có thể tạo một khoá API miễn phí trong Google AI Studio.
  • Python phiên bản 3.10 trở lên.
  • Temporal CLI để chạy máy chủ phát triển cục bộ.

Thiết lập

Trước khi bắt đầu, hãy đảm bảo bạn có một máy chủ phát triển Temporal đang chạy cục bộ:

temporal server start-dev

Tiếp theo, hãy cài đặt các phần phụ thuộc bắt buộc:

pip install temporalio google-genai httpx pydantic python-dotenv

Tạo một tệp .env trong thư mục dự án bằng khoá Gemini API của bạn. Bạn có thể lấy khoá API từ Google AI Studio.

echo "GOOGLE_API_KEY=your-api-key-here" > .env

Triển khai

Phần còn lại của hướng dẫn này sẽ trình bày về durable_agent_worker.py từ trên xuống dưới, từng bước xây dựng tác nhân. Tạo tệp và làm theo.

Nhập và thiết lập hộp cát

Bắt đầu bằng những nội dung nhập phải được xác định trước. Khối workflow.unsafe.imports_passed_through() cho biết hộp cát quy trình công việc của Temporal cho phép một số mô-đun nhất định đi qua mà không bị hạn chế. Điều này là cần thiết vì một số thư viện (đáng chú ý là httpx, phân lớp con urllib.request.Request) sử dụng các mẫu mà hộp cát sẽ chặn.

from temporalio import workflow

with workflow.unsafe.imports_passed_through():
    import pydantic_core  # noqa: F401
    import annotated_types  # noqa: F401

    import httpx
    from pydantic import BaseModel, Field
    from google import genai
    from google.genai import types

Hướng dẫn về hệ thống

Tiếp theo, hãy xác định tính cách của trợ lý. Các chỉ dẫn hệ thống cho mô hình biết cách hoạt động. Nhân viên hỗ trợ này được hướng dẫn phản hồi bằng thơ hai câu khi không cần dùng công cụ.

SYSTEM_INSTRUCTIONS = """
You are a helpful agent that can use tools to help the user.
You will be given an input from the user and a list of tools to use.
You may or may not need to use the tools to satisfy the user ask.
If no tools are needed, respond in haikus.
"""

Định nghĩa về công cụ

Bây giờ, hãy xác định những công cụ mà tác nhân có thể sử dụng. Mỗi công cụ là một hàm không đồng bộ có chuỗi tài liệu mô tả. Các công cụ nhận tham số sẽ sử dụng một mô hình Pydantic làm đối số duy nhất. Đây là phương pháp hay nhất của Temporal giúp chữ ký hoạt động ổn định khi bạn thêm các trường không bắt buộc theo thời gian.

import json

NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-app/1.0"

class GetWeatherAlertsRequest(BaseModel):
    """Request model for getting weather alerts."""

    state: str = Field(description="Two-letter US state code (e.g. CA, NY)")

async def