Trợ lý kỹ thuật dạng tác tử duy trì trạng thái với Letta: Bộ nhớ cốt lõi, trạng thái dùng chung, truy xuất ngữ nghĩa, công cụ tùy chỉnh, truyền dữ liệu theo luồng và quy trình tác tử linh hoạt

Trong tài liệu hướng dẫn này, mô hình trợ lý kỹ thuật duy trì trạng thái (stateful engineering copilot) được xây dựng trên nền tảng Letta, minh chứng cho khả năng mở rộng của bộ nhớ tác tử (agentic memory) vượt ra ngoài các cuộc hội thoại dựa trên câu lệnh (prompt) truyền thống. Quy trình bắt đầu với việc thiết lập các khối bộ nhớ cốt lõi có thể kiểm tra được, cho phép tác tử (agent) tự động cập nhật, sau đó mở các khối này để đồng bộ hóa trạng thái theo chương trình và chia sẻ quyền truy cập giữa nhiều tác tử. Kiến trúc tiếp tục được mở rộng với bộ nhớ vector lưu trữ phục vụ truy xuất ngữ nghĩa, một công cụ tùy chỉnh phía máy chủ để ước tính rủi ro có tính xác định, tính năng truyền dữ liệu luồng (streaming) nhằm giám sát quá trình thực thi của tác tử, lịch sử tin nhắn phía máy chủ phục vụ các cuộc hội thoại liên tục, cùng khả năng xuất tệp .af nhằm bảo đảm tính linh hoạt. Tổng thể các thành phần này minh chứng cho một kiến trúc bộ nhớ thực tế, trong đó trạng thái bền vững, khả năng truy xuất, các công cụ, cơ chế cộng tác đa tác tử và dấu vết thực thi cùng vận hành như một hệ thống tác tử tích hợp. Sao chép mãĐã sao chépSử dụng trình duyệt khác import subprocess, sys subprocess.run([sys.executable, "-m", "pip", "install", "-q", "-U", "letta-client"], check=True) import os, json, time, textwrap from getpass import getpass from letta_client import Letta def flex(*attempts, label=""): err = None for fn in attempts: try: return fn() except Exception as e: err = e raise RuntimeError(f"Tất cả các phương thức gọi đều thất bại đối với {label or 'lệnh gọi'}: {err}") from err API_KEY = os.environ.get("LETTA_API_KEY") or getpass("Khóa API Letta (platform.letta.com/api-keys): ") BASE_URL = os.environ.get("LETTA_BASE_URL", "").strip() client = flex( lambda: Letta(base_url=BASE_URL, api_key=API_KEY) if BASE_URL else Letta(api_key=API_KEY), lambda: Letta(base_url=BASE_URL, token=API_KEY) if BASE_URL else Letta(token=API_KEY), label="khởi tạo client", ) def hr(title): print("\n" + "=" * 92 + f"\n {title}\n" + "=" * 92) def as_text(content): if isinstance(content, str): return content if isinstance(content, list): return "".join(getattr(p, "text", "") or "" for p in content) return str(content or "") def show(response, hide_reasoning=False): for m in getattr(response, "messages", response) or []: kind = getattr(m, "message_type", None) or getattr(m, "type", "?") if kind == "reasoning_message" and not hide_reasoning: print(" lập luận :", textwrap.shorten(getattr(m, "reasoning", "") or "", 220)) elif kind == "tool_call_message": tc = getattr(m, "tool_call", None) print(" gọi công cụ :", getattr(tc, "name", "?"), textwrap.shorten(str(getattr(tc, "arguments", "")), 200)) elif kind == "tool_return_message": print(" phản hồi công cụ :", textwrap.shorten(str(getattr(m, "tool_return", "")), 260)) elif kind == "assistant_message": print(" trợ lý :", as_text(getattr(m, "content", ""))) elif kind == "approval_request_message": print(" phê duyệt : agent đang chờ người dùng phê duyệt lệnh gọi công cụ") u = getattr(response, "usage", None) if u: print(f" ↳ số bước={getattr(u, 'step_count', '?')} token={getattr(u, 'total_tokens', '?')}") def chat(agent_id, text, max_steps=12, quiet=False): r = flex( lambda: client.agents.messages.create(agent_id=agent_id, input=text, max_steps=max_steps), lambda: client.agents.messages.create(agent_id=agent_id, input=text), lambda: client.agents.messages.create(agent_id=agent_id, messages=[{"role": "user", "content": text}]), label="messages.create", ) if not quiet: print(f"\n người dùng : {text}") show(r) return r PREFERRED = ["openai/gpt-4.1", "openai/gpt-4o-mini", "anthropic/claude-sonnet-4-20250514"] try: handles = [getattr(m, "handle", None) for m in client.models.list()] MODEL = next((h for h in PREFERRED if h in handles), None) or handles[0] except Exception: MODEL = PREFERRED[0] print(f" đã kết nối · định danh mô hình = {MODEL}") hr("1. Tạo agent có trạng thái với các khối bộ nhớ tùy chỉnh") def make_agent(**kw): return flex( lambda: client.agents.create(model=MODEL, embedding="openai/text-embedding-3-small", **kw), lambda: client.agents.create(model=MODEL, **kw), label="agents.create", ) copilot = make_agent( name=f"copilot-{int(time.time())}", description="Trợ lý kỹ thuật với bộ nhớ dài hạn có thể kiểm tra", tags=["tutorial", "copilot"], memory_blocks=[ {"label": "persona", "value": "Mnemo là trợ lý kỹ thuật. Hệ thống phản hồi cô đọng, cụ thể và không bao giờ " "suy diễn thông tin — khi không nắm rõ dữ liệu sẽ thông báo rõ ràng và tiến hành tra cứu.", "limit": 3000}, {"label": "human", "description": "Dữ liệu cố định về kỹ sư phối hợp làm việc: họ tên, vai trò, công nghệ sử dụng, " "sở thích s, phong cách làm việc. Cập nhật mục này bất cứ khi nào ghi nhận được thông tin ổn định.", "value": "Hiện chưa xác định.", "limit": 3000}, {"label": "project_state", "description": "Trạng thái thực tế của dự án đang triển khai: các mốc tiến độ, điểm nghẽn, " "các quyết định và cơ sở lập luận. Duy trì thông tin cô đọng và cập nhật; xóa các dòng lỗi thời.", "value": "(trống)", "limit": 5000}, {"label": "lessons", "description": "Các bài học đắt giá và sai lầm tuyệt đối không lặp lại. Chỉ bổ sung nối tiếp, mỗi mục một dòng.", "value": "(trống)", "limit": 2000}, ], ) print("agent id:", copilot.id) def get_block(agent_id, label): return flex( lambda: client.agents.blocks.retrieve(agent_id=agent_id, block_label=label), lambda: client.agents.blocks.retrieve(agent_id, label), label="agents.blocks.retrieve", ) def set_block(agent_id, label, value): return flex( lambda: client.agents.blocks.update(agent_id=agent_id, block_label=label, value=value), lambda: client.agents.blocks.modify(agent_id=agent_id, block_label=label, value=value), lambda: client.agents.blocks.update(agent_id, label, value=value), label="agents.blocks.update", ) def dump_memory(agent_id, title="bộ nhớ cốt lõi"): blocks = flex(lambda: client.agents.blocks.list(agent_id=agent_id), lambda: client.agents.blocks.list(agent_id), label="agents.blocks.list") print(f"\n──── {title} ────") for b in blocks: ro = " [chỉ đọc]" if getattr(b, "read_only", False) else "" print(f"▸ {b.label}{ro} ({len(b.value or '')}/{getattr(b, 'limit', '?')} ký tự)") print(textwrap.indent(textwrap.fill(b.value or "", 100), " ")) return blocks dump_memory(copilot.id, "bộ nhớ cốt lõi lúc khởi tạo") Lett đã được thiết lập