钩子¶
钩子提供了一种强大的机制，用于在 Instructor 库的完成和解析过程中拦截和处理事件。它们允许你在 API 交互的不同阶段添加自定义行为、日志记录或错误处理。
概述¶
Instructor 中的钩子系统基于 Hooks 类，该类管理事件注册和发射。它支持几种预定义事件，这些事件对应于完成和解析过程的不同阶段。
支持的钩子事件¶
completion:kwargs¶
当提供完成参数时发射此钩子。它接收传递给完成函数的所有参数。这些参数将包含 model、messages、tools，在任何 response_model 或 validation_context 参数被转换为各自的值之后。
completion:response¶
当收到完成响应时发射此钩子。它接收来自完成 API 的原始响应对象。
completion:error¶
在完成过程中发生错误时发射此钩子，这发生在任何重试尝试之前，并且响应被解析为 pydantic 模型之前。
parse:error¶
当将响应解析为 pydantic 模型时发生错误时发射此钩子。这可能发生在响应无效或 pydantic 模型与响应不兼容时。
completion:last_attempt¶
当进行最后一次重试尝试时发射此钩子。
实现细节¶
钩子系统在 instructor/hooks.py 文件中实现。Hooks 类处理钩子事件的注册和发射。你可以参考此文件来查看钩子在幕后是如何工作的。使用钩子的重试逻辑在 instructor/retry.py 文件中实现。这展示了在完成过程中发生错误后如何使用钩子进行重试。
钩子类型¶
钩子系统使用类型化的 Protocol 类来为处理函数提供更好的类型安全。
from typing import Any, Protocol
# Handler protocol types for type safety
class CompletionKwargsHandler(Protocol):
"""Protocol for completion kwargs handlers."""
def __call__(self, *args: Any, **kwargs: Any) -> None: ...
class CompletionResponseHandler(Protocol):
"""Protocol for completion response handlers."""
def __call__(self, response: Any) -> None: ...
class CompletionErrorHandler(Protocol):
"""Protocol for completion error and last attempt handlers."""
def __call__(self, error: Exception) -> None: ...
class ParseErrorHandler(Protocol):
"""Protocol for parse error handlers."""
def __call__(self, error: Exception) -> None: ...
这些 Protocol 类型有助于确保你的处理函数对每种类型的钩子都有正确的签名。
钩子名称¶
钩子名称可以指定为枚举值 (HookName.COMPLETION_KWARGS) 或字符串 ("completion:kwargs")。
from instructor.hooks import HookName
# Using enum
client.on(HookName.COMPLETION_KWARGS, handler)
# Using string
client.on("completion:kwargs", handler)
注册钩子¶
你可以使用 Instructor 客户端或 Hooks 实例的 on 方法注册钩子。这里有一个示例：
import instructor
import openai
import pprint
client = instructor.from_openai(openai.OpenAI())
def log_completion_kwargs(*args, **kwargs):
pprint.pprint({"args": args, "kwargs": kwargs})
client.on("completion:kwargs", log_completion_kwargs)
resp = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": "Hello, world!"}],
response_model=str,
)
print(resp)
#> Hello, user! How can I assist you today?
发射事件¶
事件由 Instructor 库在适当的时间自动发射。在大多数情况下，你不需要手动发射事件。在内部，所有发射方法都使用一个通用的 emit 方法，该方法处理错误捕获并提供一致的行为。
移除钩子¶
你可以使用 off 方法移除特定的钩子：
import instructor
import openai
import pprint
client = instructor.from_openai(openai.OpenAI())
def log_completion_kwargs(*args, **kwargs):
pprint.pprint({"args": args, "kwargs": kwargs})
# Register the hook
client.on("completion:kwargs", log_completion_kwargs)
# Then later, remove it when no longer needed
client.off("completion:kwargs", log_completion_kwargs)
清除钩子¶
要移除某个特定事件的所有钩子或所有事件的所有钩子：
import instructor
import openai
client = instructor.from_openai(openai.OpenAI())
# Define a simple handler
def log_completion_kwargs(*args, **kwargs):
print("Logging completion kwargs...")
# Register the hook
client.on("completion:kwargs", log_completion_kwargs)
# Make a request that triggers the hook
resp = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": "Hello, world!"}],
response_model=str,
)
# Clear hooks for a specific event
client.clear("completion:kwargs")
# Register another handler for a different event
def log_response(response):
print("Logging response...")
client.on("completion:response", log_response)
# Clear all hooks
client.clear()
示例：日志记录和调试¶
这是一个综合示例，演示了如何使用钩子进行日志记录和调试：
import instructor
import openai
import pydantic
def log_completion_kwargs(*args, **kwargs) -> None:
print("## Completion kwargs:")
print(kwargs)
# Example output:
# {
# "messages": [
# {
# "role": "user",
# "content": "Extract the user name and age from the following text: 'John is 20 years old'",
# }
# ],
# "model": "gpt-4o-mini",
# "tools": [
# {
# "type": "function",
# "function": {
# "name": "User",
# "description": "Correctly extracted `User` with all the required parameters with correct types",
# "parameters": {
# "properties": {
# "name": {"title": "Name", "type": "string"},
# "age": {"title": "Age", "type": "integer"},
# },
# "required": ["age", "name"],
# "type": "object",
# },
# },
# }
# ],
# "tool_choice": {"type": "function", "function": {"name": "User"}},
# }
def log_completion_response(response) -> None:
print("## Completion response:")
# Example output:
# {
# 'id': 'chatcmpl-AWl4Mj5Jrv7m7JkOTIiHXSldQIOFm',
# 'choices': [
# {
# 'finish_reason': 'stop',
# 'index': 0,
# 'logprobs': None,
# 'message': {
# 'content': None,
# 'refusal': None,
# 'role': 'assistant',
# 'audio': None,
# 'function_call': None,
# 'tool_calls': [
# {
# 'id': 'call_6oQ9WXxeSiVEV71B9IYtsbIE',
# 'function': {
# 'arguments': '{"name":"John","age":-1}',
# 'name': 'User',
# },
# 'type': 'function',
# }
# ],
# },
# }
# ],
# 'created': 1732370794,
# 'model': 'gpt-4o-mini-2024-07-18',
# 'object': 'chat.completion',
# 'service_tier': None,
# 'system_fingerprint': 'fp_0705bf87c0',
# 'usage': {
# 'completion_tokens': 10,
# 'prompt_tokens': 87,
# 'total_tokens': 97,
# 'completion_tokens_details': {
# 'audio_tokens': 0,
# 'reasoning_tokens': 0,
# 'accepted_prediction_tokens': 0,
# 'rejected_prediction_tokens': 0,
# },
# 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0},
# },
# }
print(response.model_dump())
def handle_completion_error(error: Exception) -> None:
print(f"## Completion error: {error}")
print(f"Type: {type(error).__name__}")
print(f"Message: {str(error)}")
def log_parse_error(error: Exception) -> None:
print(f"## Parse error: {error}")
print(f"Type: {type(error).__name__}")
print(f"Message: {str(error)}")
# Handler for a custom logger that records how many errors have occurred
class ErrorCounter:
def __init__(self) -> None:
self.error_count = 0
def count_error(self, error: Exception) -> None:
self.error_count += 1
print(f"Error count: {self.error_count}")
client = instructor.from_openai(openai.OpenAI())
# Register the hooks
client.on("completion:kwargs", log_completion_kwargs)
client.on("completion:response", log_completion_response)
client.on("completion:error", handle_completion_error)
client.on("parse:error", log_parse_error)
# Example with error counter
error_counter = ErrorCounter()
client.on("completion:error", error_counter.count_error)
client.on("parse:error", error_counter.count_error)
# Define a model for extraction
class User(pydantic.BaseModel):
name: str
age: int
# Try extraction with a potentially problematic input
try:
resp = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "user",
"content": "Extract the user name and age: 'John is twenty years old'",
}
],
response_model=User,
)
print(f"Extracted: {resp}")
except Exception as e:
print(f"Main exception caught: {e}")
# Check the error count
print(f"Total errors recorded: {error_counter.error_count}")
高级：创建自定义钩子¶
虽然 Instructor 库提供了一些内置钩子，但你可能需要为特定用例创建自定义钩子。你可以通过扩展 HookName 枚举并为你的自定义事件添加处理程序来实现：
from typing import Protocol, Any
from enum import Enum
from instructor.hooks import Hooks, HookName
# Extend the HookName enum
class CustomHookName(str, Enum):
CUSTOM_EVENT = "custom:event"
# Make it compatible with the base HookName enum
COMPLETION_KWARGS = HookName.COMPLETION_KWARGS.value
COMPLETION_RESPONSE = HookName.COMPLETION_RESPONSE.value
COMPLETION_ERROR = HookName.COMPLETION_ERROR.value
PARSE_ERROR = HookName.PARSE_ERROR.value
COMPLETION_LAST_ATTEMPT = HookName.COMPLETION_LAST_ATTEMPT.value
# Create a hooks instance
hooks = Hooks()
# Define a handler
def custom_handler(data):
print(f"Custom event: {data}")
# Register the handler
hooks.on(CustomHookName.CUSTOM_EVENT, custom_handler)
# Emit the event
hooks.emit(CustomHookName.CUSTOM_EVENT, {"data": "value"})
使用 Protocol 类型确保类型安全¶
钩子系统使用 Python 的 Protocol 类型来为处理函数提供更好的类型安全。这有助于在开发阶段捕获错误，并提供更好的 IDE 支持和自动补全。
如果你正在编写自己的处理程序，可以指定适当的类型：
from instructor.hooks import CompletionErrorHandler
def my_error_handler(error: Exception) -> None:
print(f"Error occurred: {error}")
# Type checking will verify this is a valid error handler
handler: CompletionErrorHandler = my_error_handler
client.on("completion:error", handler)
测试中的钩子¶
钩子对于测试特别有用，因为它们允许你在不修改应用程序代码的情况下检查参数和响应：
import unittest
from unittest.mock import Mock
import instructor
import openai
class TestMyApp(unittest.TestCase):
def test_completion(self):
client = instructor.from_openai(openai.OpenAI())
mock_handler = Mock()
client.on("completion:response", mock_handler)
# Call your code that uses the client
result = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": "Hello"}],
response_model=str,
)
# Verify the mock was called
mock_handler.assert_called_once()
# You can also inspect the arguments
response_arg = mock_handler.call_args[0][0]
self.assertEqual(response_arg.model, "gpt-3.5-turbo")
这种方法允许你在不模拟整个客户端的情况下测试你的代码。
使用钩子¶
from openai import OpenAI
import instructor
# Initialize client
client = instructor.patch(OpenAI())
# Example with all hooks enabled (default)
response = client.chat.completions.create(
model="gpt-3.5-turbo",
response_model=str,
messages=[{"role": "user", "content": "Hello!"}],
)
from enum import Enum, auto
import instructor
from openai import OpenAI
# Define standard hook names
class HookName(Enum):
COMPLETION_KWARGS = auto()
COMPLETION_RESPONSE = auto()
COMPLETION_ERROR = auto()
COMPLETION_LAST_ATTEMPT = auto()
PARSE_ERROR = auto()
# Create a new enum for custom hooks
class CustomHookName(Enum):
MY_CUSTOM_HOOK = "my_custom_hook"
ANOTHER_HOOK = "another_hook"
# Initialize client with custom hooks
client = instructor.patch(OpenAI())