Files
wehub-resource-sync 91e75e620b
CI: cua-driver distro-compat matrix / debian:12 (glibc 2.36) (push) Has been cancelled
CI: SPDX Headers / Check SPDX headers (warn-only) (push) Has been cancelled
CD: Docs MCP Server / build (linux/amd64) (push) Has been cancelled
CD: Docs MCP Server / build (linux/arm64) (push) Has been cancelled
CD: Docs MCP Server / merge (push) Has been cancelled
CI: cua-driver distro-compat matrix / Resolve release version (push) Has been cancelled
CI: cua-driver distro-compat matrix / fedora:41 (glibc 2.40) (push) Has been cancelled
CI: cua-driver distro-compat matrix / rockylinux:9 (glibc 2.34) (push) Has been cancelled
CI: cua-driver distro-compat matrix / ubuntu:22.04 (glibc 2.35) (push) Has been cancelled
CI: cua-driver distro-compat matrix / ubuntu:24.04 (glibc 2.39) (push) Has been cancelled
CI: cua-driver distro-compat matrix / Distro compat summary (push) Has been cancelled
CI: Rust Linux unit / Rust Linux unit and compile (push) Has been cancelled
CI: Rust Windows unit / Rust Windows unit and compile (push) Has been cancelled
CI: Nix Linux Rust source / Nix / compositor build (push) Has been cancelled
CI: Nix Linux Rust source / Nix / driver package (push) Has been cancelled
CI: Nix Linux Rust source / Nix / Rust unit tests (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 13:03:19 +08:00

48 lines
1.5 KiB
Python

"""
Prompt instructions callback.
This callback allows simple prompt engineering by pre-pending a user
instructions message to the start of the conversation before each LLM call.
Usage:
from cua_agent.callbacks import PromptInstructionsCallback
agent = ComputerAgent(
model="openai/computer-use-preview",
callbacks=[PromptInstructionsCallback("Follow these rules...")]
)
"""
from typing import Any, Dict, List, Optional
from .base import AsyncCallbackHandler
class PromptInstructionsCallback(AsyncCallbackHandler):
"""
Prepend a user instructions message to the message list.
This is a minimal, non-invasive way to guide the agent's behavior without
modifying agent loops or tools. It works with any provider/loop since it
only alters the messages array before sending to the model.
"""
def __init__(self, instructions: Optional[str]) -> None:
self.instructions = instructions
async def on_llm_start(self, messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
# Pre-pend instructions message
if not self.instructions:
return messages
# Ensure we don't duplicate if already present at the front
if messages and isinstance(messages[0], dict):
first = messages[0]
if first.get("role") == "user" and first.get("content") == self.instructions:
return messages
return [
{"role": "user", "content": self.instructions},
] + messages