Móc sự kiện

Hermes có hai hệ thống hook chạy mã tùy chỉnh tại các điểm quan trọng trong vòng đời:

Hệ thống	Đã đăng ký qua	Chạy vào	Trường hợp sử dụng
Móc cổng	HOOK.yaml + handler.py trong ~/.hermes/hooks/	Chỉ cổng	Ghi nhật ký, cảnh báo, webhook
Móc cắm	ctx.register_hook() trong plugin	CLI + Cổng	Công cụ chặn, số liệu, lan can

Cả hai hệ thống đều không bị chặn - lỗi trong bất kỳ hook nào đều được phát hiện và ghi lại, không bao giờ làm hỏng tác nhân.

Móc sự kiện cổng

Cổng kết nối tự động kích hoạt trong quá trình vận hành cổng (Telegram, Discord, Slack, WhatsApp) mà không chặn đường ống tác nhân chính.

Tạo một Hook

Mỗi hook là một thư mục trong ~/.hermes/hooks/ chứa hai tệp:

~/.hermes/hooks/
└── my-hook/
    ├── HOOK.yaml      

# Declares which events to listen for
    └── handler.py     

# Python handler function

HOOK.yaml

name: my-hook
description: Log all agent activity to a file
events:
  - agent:start
  - agent:end
  - agent:step

Danh sách events xác định sự kiện nào kích hoạt trình xử lý của bạn. Bạn có thể đăng ký bất kỳ sự kết hợp sự kiện nào, bao gồm các ký tự đại diện như command:* .

handler.py

import json
from datetime import datetime
from pathlib import Path

LOG_FILE = Path.home() / ".hermes" / "hooks" / "my-hook" / "activity.log"

async def handle(event_type: str, context: dict):
    """Called for each subscribed event. Must be named 'handle'."""
    entry = {
        "timestamp": datetime.now().isoformat(),
        "event": event_type,
        **context,
    }
    with open(LOG_FILE, "a") as f:
        f.write(json.dumps(entry) + "\n")

Quy tắc xử lý:

• Phải có tên handle
• Nhận event_type (chuỗi) và context (dict)
• Có thể là async def hoặc def thông thường — cả hai đều hoạt động
• Lỗi được bắt và ghi lại, không bao giờ làm hỏng tác nhân

Sự kiện có sẵn

| Sự kiện | Khi nó cháy | Phím ngữ cảnh | |-------|--------------||--------------| | gateway:startup | Quá trình cổng bắt đầu | platforms (danh sách tên nền tảng đang hoạt động) | | session:start | Đã tạo phiên nhắn tin mới | platform , user_id , session_id , session_key | | session:end | Phiên kết thúc (trước khi đặt lại) | platform , user_id , session_key | | session:reset | Người dùng đã chạy /new hoặc /reset | platform , user_id , session_key | | agent:start | Đại lý bắt đầu xử lý tin nhắn | platform , user_id , session_id , message | | agent:step | Mỗi lần lặp của vòng lặp gọi công cụ | platform , user_id , session_id , iteration , tool_names | | agent:end | Đại lý hoàn tất xử lý | platform , user_id , session_id , message , response | | command:* | Bất kỳ lệnh gạch chéo nào được thực thi | platform , user_id , command , args |

So khớp ký tự đại diện

Trình xử lý đã đăng ký command:* kích hoạt cho bất kỳ sự kiện command: nào ( command:model , command:reset , v.v.). Giám sát tất cả các lệnh gạch chéo bằng một lần đăng ký.

Ví dụ

Danh sách kiểm tra khởi động (BOOT.md) - Tích hợp sẵn

Cổng đi kèm với móc boot-md tích hợp sẵn để tìm kiếm ~/.hermes/BOOT.md mỗi lần khởi động. Nếu tệp tồn tại, tác nhân sẽ chạy hướng dẫn của nó trong phiên nền. Không cần cài đặt - chỉ cần tạo tệp.

Tạo ~/.hermes/BOOT.md :

# Startup Checklist

1. Check if any cron jobs failed overnight — run `hermes cron list`
2. Send a message to Discord #general saying "Gateway restarted, all systems go"
3. Check if /opt/app/deploy.log has any errors from the last 24 hours

Tác nhân chạy các hướng dẫn này trong một luồng nền để nó không chặn quá trình khởi động cổng. Nếu không có gì cần chú ý, nhân viên sẽ trả lời [SILENT] và không có tin nhắn nào được gửi.

mẹo

Không có BOOT.md? Cái móc âm thầm bỏ qua - không có chi phí. Tạo tệp bất cứ khi nào bạn cần tự động hóa khởi động, xóa nó khi không cần.

Cảnh báo Telegram về các nhiệm vụ dài

Gửi tin nhắn cho chính bạn khi nhân viên thực hiện hơn 10 bước:

# ~/.hermes/hooks/long-task-alert/HOOK.yaml
name: long-task-alert
description: Alert when agent is taking many steps
events:
  - agent:step

# ~/.hermes/hooks/long-task-alert/handler.py
import os
import httpx

THRESHOLD = 10
BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")
CHAT_ID = os.getenv("TELEGRAM_HOME_CHANNEL")

async def handle(event_type: str, context: dict):
    iteration = context.get("iteration", 0)
    if iteration == THRESHOLD and BOT_TOKEN and CHAT_ID:
        tools = ", ".join(context.get("tool_names", []))
        text = f"⚠️ Agent has been running for {iteration} steps. Last tools: {tools}"
        async with httpx.AsyncClient() as client:
            await client.post(
                f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage",
                json={"chat_id": CHAT_ID, "text": text},
            )

Trình ghi nhật ký sử dụng lệnh

Theo dõi lệnh gạch chéo nào được sử dụng:

# ~/.hermes/hooks/command-logger/HOOK.yaml
name: command-logger
description: Log slash command usage
events:
  - command:*

# ~/.hermes/hooks/command-logger/handler.py
import json
from datetime import datetime
from pathlib import Path

LOG = Path.home() / ".hermes" / "logs" / "command_usage.jsonl"

def handle(event_type: str, context: dict):
    LOG.parent.mkdir(parents=True, exist_ok=True)
    entry = {
        "ts": datetime.now().isoformat(),
        "command": context.get("command"),
        "args": context.get("args"),
        "platform": context.get("platform"),
        "user": context.get("user_id"),
    }
    with open(LOG, "a") as f:
        f.write(json.dumps(entry) + "\n")

Webhook bắt đầu phiên

POST lên dịch vụ bên ngoài trong các phiên mới:

# ~/.hermes/hooks/session-webhook/HOOK.yaml
name: session-webhook
description: Notify external service on new sessions
events:
  - session:start
  - session:reset

# ~/.hermes/hooks/session-webhook/handler.py
import httpx

WEBHOOK_URL = "https://your-service.example.com/hermes-events"

async def handle(event_type: str, context: dict):
    async with httpx.AsyncClient() as client:
        await client.post(WEBHOOK_URL, json={
            "event": event_type,
            **context,
        }, timeout=5)

Cách thức hoạt động

1. Khi khởi động cổng, HookRegistry.discover_and_load() quét ~/.hermes/hooks/
2. Mỗi thư mục con có HOOK.yaml + handler.py được tải động
3. Trình xử lý được đăng ký cho các sự kiện đã khai báo của họ
4. Tại mỗi điểm trong vòng đời, hooks.emit() sẽ kích hoạt tất cả các trình xử lý phù hợp
5. Lỗi trong bất kỳ trình xử lý nào đều được phát hiện và ghi lại - một hook bị hỏng không bao giờ làm hỏng tác nhân

thông tin

Móc cổng chỉ kích hoạt trong gateway (Telegram, Discord, Slack, WhatsApp). CLI không tải các móc nối cổng. Để có các hook hoạt động ở mọi nơi, hãy sử dụng plugin hooks.

Móc plugin

Plugins có thể đăng ký các hook kích hoạt trong phiên cả CLI và cổng. Chúng được đăng ký theo chương trình thông qua ctx.register_hook() trong hàm register() của plugin của bạn.

def register(ctx):
    ctx.register_hook("pre_tool_call", my_tool_observer)
    ctx.register_hook("post_tool_call", my_tool_logger)
    ctx.register_hook("pre_llm_call", my_memory_callback)
    ctx.register_hook("post_llm_call", my_sync_callback)
    ctx.register_hook("on_session_start", my_init_callback)
    ctx.register_hook("on_session_end", my_cleanup_callback)

Quy tắc chung cho tất cả các hook:

• Lệnh gọi lại nhận đối số từ khóa. Luôn chấp nhận **kwargs để có khả năng tương thích về sau — các tham số mới có thể được thêm vào trong các phiên bản sau này mà không làm hỏng plugin của bạn.
• Nếu lệnh gọi lại bị lỗi, lệnh gọi lại đó sẽ được ghi lại và bị bỏ qua. Các hook khác và đại lý vẫn tiếp tục bình thường. Một plugin hoạt động sai không bao giờ có thể phá vỡ tác nhân.
• Tất cả các hook đều quan sát viên bắn và quên có giá trị trả về bị bỏ qua — ngoại trừ pre_llm_call , có thể chèn bối cảnh.

Tham khảo nhanh

Móc	Kích hoạt khi	Trả về
pre_tool_call	Trước khi bất kỳ công cụ nào thực thi	bỏ qua
post_tool_call	Sau khi bất kỳ công cụ nào trả về	bỏ qua
pre_llm_call	Mỗi lượt một lần, trước vòng lặp gọi công cụ	tiêm bối cảnh
post_llm_call	Mỗi lượt một lần, sau vòng gọi công cụ	bỏ qua
on_session_start	Đã tạo phiên mới (chỉ lượt đầu tiên)	bỏ qua
on_session_end	Phiên kết thúc	bỏ qua

---

pre_tool_call

Kích hoạt ngay lập tức trước mỗi lần thực thi công cụ — các công cụ tích hợp sẵn cũng như công cụ plugin tương tự.

Chữ ký gọi lại:

def my_callback(tool_name: str, args: dict, task_id: str, **kwargs):

Tham số	Loại	Mô tả
tool_name	str	Tên của công cụ sắp thực thi (ví dụ: "terminal" , "web_search" , "read_file" )
args	dict	Các đối số mà mô hình truyền cho công cụ
task_id	str	Mã định danh phiên/nhiệm vụ. Chuỗi trống nếu không được đặt.

Kích hoạt: Trong model_tools.py , bên trong handle_function_call() , trước khi trình xử lý của công cụ chạy. Kích hoạt một lần cho mỗi lệnh gọi công cụ — nếu mô hình gọi song song 3 công cụ thì thao tác này sẽ kích hoạt 3 lần.Giá trị trả về: Bỏ qua.

Các trường hợp sử dụng: Ghi nhật ký, kiểm tra đường đi, bộ đếm lệnh gọi công cụ, chặn các hoạt động nguy hiểm (in cảnh báo), giới hạn tốc độ.

Ví dụ — nhật ký kiểm tra lệnh gọi công cụ:

import json, logging
from datetime import datetime

logger = logging.getLogger(__name__)

def audit_tool_call(tool_name, args, task_id, **kwargs):
    logger.info("TOOL_CALL session=%s tool=%s args=%s",
                task_id, tool_name, json.dumps(args)[:200])

def register(ctx):
    ctx.register_hook("pre_tool_call", audit_tool_call)

Ví dụ — cảnh báo về các công cụ nguy hiểm:

DANGEROUS = {"terminal", "write_file", "patch"}

def warn_dangerous(tool_name, **kwargs):
    if tool_name in DANGEROUS:
        print(f"⚠ Executing potentially dangerous tool: {tool_name}")

def register(ctx):
    ctx.register_hook("pre_tool_call", warn_dangerous)

---

post_tool_call

Kích hoạt ngay sau mỗi lần thực thi công cụ trở lại.

Chữ ký gọi lại:

def my_callback(tool_name: str, args: dict, result: str, task_id: str, **kwargs):

Tham số	Loại	Mô tả
tool_name	str	Tên công cụ vừa thực thi
args	dict	Các đối số mà mô hình truyền cho công cụ
result	str	Giá trị trả về của công cụ (luôn là chuỗi JSON)
task_id	str	Mã định danh phiên/nhiệm vụ. Chuỗi trống nếu không được đặt.

Kích hoạt: Trong model_tools.py , bên trong handle_function_call() , sau khi trình xử lý của công cụ quay trở lại. Kích hoạt một lần cho mỗi lệnh gọi công cụ. không kích hoạt nếu công cụ đưa ra một ngoại lệ chưa được xử lý (thay vào đó, lỗi được phát hiện và trả về dưới dạng một chuỗi JSON lỗi và post_tool_call kích hoạt với chuỗi lỗi đó dưới dạng result ).

Giá trị trả về: Bỏ qua.

Các trường hợp sử dụng: Ghi lại kết quả của công cụ, thu thập số liệu, theo dõi tỷ lệ thành công/thất bại của công cụ, gửi thông báo khi các công cụ cụ thể hoàn tất.

Ví dụ — theo dõi số liệu sử dụng công cụ:

from collections import Counter
import json

_tool_counts = Counter()
_error_counts = Counter()

def track_metrics(tool_name, result, **kwargs):
    _tool_counts[tool_name] += 1
    try:
        parsed = json.loads(result)
        if "error" in parsed:
            _error_counts[tool_name] += 1
    except (json.JSONDecodeError, TypeError):
        pass

def register(ctx):
    ctx.register_hook("post_tool_call", track_metrics)

---

pre_llm_call

Bắn một lần mỗi lượt, trước khi vòng lặp gọi công cụ bắt đầu. Đây là hook duy nhất có giá trị trả về được sử dụng — nó có thể đưa ngữ cảnh vào thông báo người dùng của lượt hiện tại.

Chữ ký gọi lại:

def my_callback(session_id: str, user_message: str, conversation_history: list,
                is_first_turn: bool, model: str, platform: str, **kwargs):

Tham số	Loại	Mô tả
session_id	str	Mã định danh duy nhất cho phiên hiện tại
user_message	str	Tin nhắn ban đầu của người dùng cho lượt này (trước khi tiêm bất kỳ kỹ năng nào)
conversation_history	list	Bản sao danh sách tin nhắn đầy đủ (định dạng OpenAI: [{"role": "user", "content": "..."}] )
is_first_turn	bool	True nếu đây là lượt đầu tiên của phiên mới, False ở các lượt tiếp theo
model	str	Mã định danh mô hình (ví dụ: "anthropic/claude-sonnet-4.6" )
platform	str	Nơi phiên đang chạy: "cli" , "telegram" , "discord" , v.v.

Kích hoạt: Trong run_agent.py , bên trong run_conversation() , sau khi nén ngữ cảnh nhưng trước vòng lặp while chính. Kích hoạt một lần cho mỗi lệnh gọi run_conversation() (tức là một lần cho mỗi lượt người dùng), không phải một lần cho mỗi lệnh gọi API trong vòng lặp công cụ.

Giá trị trả về: Nếu lệnh gọi lại trả về một lệnh có khóa "context" hoặc một chuỗi thuần túy không trống, thì văn bản sẽ được thêm vào thông báo người dùng của lượt hiện tại. Trả lại None nếu không tiêm.

# Inject context
return {"context": "Recalled memories:\n- User likes Python\n- Working on hermes-agent"}

# Plain string (equivalent)
return "Recalled memories:\n- User likes Python"

# No injection
return None

```**Nơi ngữ cảnh được chèn:** Luôn là **thông báo của người dùng**, không bao giờ là lời nhắc của hệ thống. Điều này sẽ duy trì bộ nhớ đệm lời nhắc — lời nhắc hệ thống vẫn giống hệt nhau qua các lượt, do đó các mã thông báo đã lưu trong bộ nhớ đệm sẽ được sử dụng lại. Lời nhắc của hệ thống là lãnh thổ của Hermes (hướng dẫn mô hình, thực thi công cụ, tính cách, kỹ năng). Các plugin đóng góp ngữ cảnh cùng với thông tin đầu vào của người dùng.

Tất cả ngữ cảnh được chèn đều **phù du** — chỉ được thêm vào thời điểm gọi API. Tin nhắn ban đầu của người dùng trong lịch sử hội thoại không bao giờ bị thay đổi và không có gì được lưu lại trong cơ sở dữ liệu phiên.

Khi **nhiều plugin** trả về ngữ cảnh, kết quả đầu ra của chúng được nối với hai dòng mới theo thứ tự khám phá plugin (theo bảng chữ cái theo tên thư mục).

**Các trường hợp sử dụng:** Thu hồi bộ nhớ, chèn ngữ cảnh RAG, lan can, phân tích mỗi lượt.

**Ví dụ — thu hồi bộ nhớ:**

```python
import httpx

MEMORY_API = "https://your-memory-api.example.com"

def recall(session_id, user_message, is_first_turn, **kwargs):
    try:
        resp = httpx.post(f"{MEMORY_API}/recall", json={
            "session_id": session_id,
            "query": user_message,
        }, timeout=3)
        memories = resp.json().get("results", [])
        if not memories:
            return None
        text = "Recalled context:\n" + "\n".join(f"- {m['text']}" for m in memories)
        return {"context": text}
    except Exception:
        return None

def register(ctx):
    ctx.register_hook("pre_llm_call", recall)

Ví dụ — lan can:

POLICY = "Never execute commands that delete files without explicit user confirmation."

def guardrails(**kwargs):
    return {"context": POLICY}

def register(ctx):
    ctx.register_hook("pre_llm_call", guardrails)

---

post_llm_call

Kích hoạt một lần mỗi lượt, sau khi vòng lặp gọi công cụ hoàn thành và tác nhân đã đưa ra phản hồi cuối cùng. Chỉ kích hoạt ở lượt thành công - không kích hoạt nếu lượt bị gián đoạn.

Chữ ký gọi lại:

def my_callback(session_id: str, user_message: str, assistant_response: str,
                conversation_history: list, model: str, platform: str, **kwargs):

Tham số	Loại	Mô tả
session_id	str	Mã định danh duy nhất cho phiên hiện tại
user_message	str	Tin nhắn ban đầu của người dùng cho lượt này
assistant_response	str	Phản hồi văn bản cuối cùng của đại lý cho lượt này
conversation_history	list	Bản sao danh sách tin nhắn đầy đủ sau khi hoàn thành lượt
model	str	Mã định danh mô hình
platform	str	Phiên đang chạy ở đâu

Kích hoạt: Trong run_agent.py , bên trong run_conversation() , sau khi vòng lặp công cụ thoát ra với phản hồi cuối cùng. Được bảo vệ bởi if final_response and not interrupted - vì vậy, nó không kích hoạt khi người dùng ngắt giữa lượt hoặc tác nhân đạt đến giới hạn lặp lại mà không tạo ra phản hồi.

Giá trị trả về: Bỏ qua.

Trường hợp sử dụng: Đồng bộ hóa dữ liệu cuộc hội thoại với hệ thống bộ nhớ ngoài, tính toán số liệu chất lượng phản hồi, ghi nhật ký tóm tắt lượt, kích hoạt các hành động tiếp theo.

Ví dụ — đồng bộ với bộ nhớ ngoài:

import httpx

MEMORY_API = "https://your-memory-api.example.com"

def sync_memory(session_id, user_message, assistant_response, **kwargs):
    try:
        httpx.post(f"{MEMORY_API}/store", json={
            "session_id": session_id,
            "user": user_message,
            "assistant": assistant_response,
        }, timeout=5)
    except Exception:
        pass  

# best-effort

def register(ctx):
    ctx.register_hook("post_llm_call", sync_memory)

Ví dụ — theo dõi độ dài phản hồi:

import logging
logger = logging.getLogger(__name__)

def log_response_length(session_id, assistant_response, model, **kwargs):
    logger.info("RESPONSE session=%s model=%s chars=%d",
                session_id, model, len(assistant_response or ""))

def register(ctx):
    ctx.register_hook("post_llm_call", log_response_length)

---

on_session_start

Kích hoạt một lần khi một phiên hoàn toàn mới được tạo. không kích hoạt khi tiếp tục phiên (khi người dùng gửi tin nhắn thứ hai trong phiên hiện có).

Chữ ký gọi lại:

def my_callback(session_id: str, model: str, platform: str, **kwargs):

Tham số	Loại	Mô tả
session_id	str	Mã định danh duy nhất cho phiên mới
model	str	Mã định danh mô hình
platform	str	Phiên đang chạy ở đâu

Kích hoạt: Trong run_agent.py , bên trong run_conversation() , trong lượt đầu tiên của phiên mới — cụ thể là sau khi lời nhắc hệ thống được tạo nhưng trước khi vòng lặp công cụ bắt đầu. Kiểm tra là if not conversation_history (không có tin nhắn trước = phiên mới).

Giá trị trả về: Bỏ qua.Các trường hợp sử dụng: Đang khởi tạo trạng thái trong phạm vi phiên, làm nóng bộ nhớ đệm, đăng ký phiên với dịch vụ bên ngoài, phiên ghi nhật ký bắt đầu.

Ví dụ — khởi tạo bộ đệm phiên:

_session_caches = {}

def init_session(session_id, model, platform, **kwargs):
    _session_caches[session_id] = {
        "model": model,
        "platform": platform,
        "tool_calls": 0,
        "started": __import__("datetime").datetime.now().isoformat(),
    }

def register(ctx):
    ctx.register_hook("on_session_start", init_session)

---

on_session_end

Kích hoạt vào cuối mỗi lệnh gọi run_conversation(), bất kể kết quả như thế nào. Đồng thời kích hoạt từ trình xử lý thoát của CLI nếu tác nhân đang ở giữa lượt khi người dùng thoát.

Chữ ký gọi lại:

def my_callback(session_id: str, completed: bool, interrupted: bool,
                model: str, platform: str, **kwargs):

Tham số	Loại	Mô tả
session_id	str	Mã định danh duy nhất cho phiên
completed	bool	True nếu nhân viên hỗ trợ đưa ra phản hồi cuối cùng, False nếu không
interrupted	bool	True nếu lượt bị gián đoạn (người dùng đã gửi tin nhắn mới /stop hoặc thoát)
model	str	Mã định danh mô hình
platform	str	Phiên đang chạy ở đâu

Cháy: Ở hai nơi:

1. ** run_agent.py ** — vào cuối mỗi cuộc gọi run_conversation(), sau khi dọn dẹp xong. Luôn bắn, ngay cả khi lượt bị lỗi.
2. ** cli.py ** — trong trình xử lý atexit của CLI, nhưng chỉ nếu tác nhân đang ở giữa lượt ( _agent_running=True ) khi xảy ra lần thoát. Điều này bắt Ctrl+C và /exit trong quá trình xử lý. Trong trường hợp này, completed=False và interrupted=True .

Giá trị trả về: Bỏ qua.

Các trường hợp sử dụng: Xóa bộ đệm, đóng kết nối, duy trì trạng thái phiên, thời lượng phiên ghi nhật ký, dọn sạch các tài nguyên được khởi tạo trong on_session_start .

Ví dụ — xả và dọn dẹp:

_session_caches = {}

def cleanup_session(session_id, completed, interrupted, **kwargs):
    cache = _session_caches.pop(session_id, None)
    if cache:
        

# Flush accumulated data to disk or external service
        status = "completed" if completed else ("interrupted" if interrupted else "failed")
        print(f"Session {session_id} ended: {status}, {cache['tool_calls']} tool calls")

def register(ctx):
    ctx.register_hook("on_session_end", cleanup_session)

Ví dụ — theo dõi thời lượng phiên:

import time, logging
logger = logging.getLogger(__name__)

_start_times = {}

def on_start(session_id, **kwargs):
    _start_times[session_id] = time.time()

def on_end(session_id, completed, interrupted, **kwargs):
    start = _start_times.pop(session_id, None)
    if start:
        duration = time.time() - start
        logger.info("SESSION_DURATION session=%s seconds=%.1f completed=%s interrupted=%s",
                     session_id, duration, completed, interrupted)

def register(ctx):
    ctx.register_hook("on_session_start", on_start)
    ctx.register_hook("on_session_end", on_end)

---

Xem Hướng dẫn xây dựng plugin để biết hướng dẫn đầy đủ bao gồm lược đồ công cụ, trình xử lý và mẫu hook nâng cao.