高级模型插件#

模型插件教程涵盖了开发一个插件的基础知识，该插件用于添加对新模型的支持。

本文档涵盖了更高级的主题。

接受API密钥的模型#

调用诸如OpenAI、Anthropic或Google Gemini等API提供商的模型通常需要一个API密钥。

LLM的API密钥管理机制在此描述。

如果你的插件需要一个API密钥，你应该继承llm.KeyModel类而不是llm.Model类。像这样开始你的模型定义：

import llm

class HostedModel(llm.KeyModel):
    needs_key = "hosted" # Required
    key_env_var = "HOSTED_API_KEY" # Optional

这告诉LLM，您的模型需要一个API密钥，该密钥可能保存在密钥注册表中，键名为hosted，或者也可能作为HOSTED_API_KEY环境变量提供。

然后当你定义你的execute()方法时，它应该接受一个额外的key=参数，如下所示：

    def execute(self, prompt, stream, response, conversation, key=None):
        # key= here will be the API key to use

LLM 将从环境变量、密钥注册表或已传递给 LLM 的密钥中传入密钥，作为 --key 命令行选项或 model.prompt(..., key=) 参数。

异步模型#

插件可以选择性地提供其模型的异步版本，适用于与Python asyncio一起使用。这对于通过HTTP API访问的远程模型特别有用。

模型的异步版本子类化llm.AsyncModel而不是llm.Model。它必须实现一个async def execute()异步生成器方法，而不是def execute()。

此示例展示了OpenAI默认插件的一个子集，说明了此方法可能如何工作：

from typing import AsyncGenerator
import llm

class MyAsyncModel(llm.AsyncModel):
    # This can duplicate the model_id of the sync model:
    model_id = "my-model-id"

    async def execute(
        self, prompt, stream, response, conversation=None
    ) -> AsyncGenerator[str, None]:
        if stream:
            completion = await client.chat.completions.create(
                model=self.model_id,
                messages=messages,
                stream=True,
            )
            async for chunk in completion:
                yield chunk.choices[0].delta.content
        else:
            completion = await client.chat.completions.create(
                model=self.model_name or self.model_id,
                messages=messages,
                stream=False,
            )
            yield completion.choices[0].message.content

如果你的模型需要一个API密钥，你应该改为继承llm.AsyncKeyModel，并在你的.execute()方法上有一个key=参数：

class MyAsyncModel(llm.AsyncKeyModel):
    ...
    async def execute(
        self, prompt, stream, response, conversation=None, key=None
    ) -> AsyncGenerator[str, None]:

然后，这个异步模型实例应该传递给register_models()插件钩子中的register()方法：

@hookimpl
def register_models(register):
    register(
        MyModel(), MyAsyncModel(), aliases=("my-model-aliases",)
    )

多模态模型的附件#

诸如GPT-4o、Claude 3.5 Sonnet和Google的Gemini 1.5等模型是多模态的：它们接受图像形式的输入，甚至可能接受音频、视频和其他格式的输入。

LLM 将这些称为附件。模型可以指定它们接受的附件类型，然后在 .execute() 方法中实现特殊代码来处理它们。

请参阅Python附件文档了解如何在Python API中使用附件。

指定附件类型#

一个 Model 子类可以通过定义一个 attachment_types 类属性来列出它接受的附件类型：

class NewModel(llm.Model):
    model_id = "new-model"
    attachment_types = {
        "image/png",
        "image/jpeg",
        "image/webp",
        "image/gif",
    }

当使用llm -a filename将附件传递给LLM时，会检测这些内容类型，或者用户可以使用--attachment-type filename image/png选项来指定。

注意： MP3文件的附件类型将被检测为audio/mpeg，而不是audio/mp3。

LLM 将使用 attachment_types 属性来验证提供的附件是否应被接受，然后再将它们传递给模型。

处理附件#

传递给execute()方法的prompt对象将具有一个attachments属性，该属性包含用户提供的Attachment对象列表。

一个Attachment实例具有以下属性：

url (str): 附件的URL，如果它是作为URL提供的
path (str): 如果附件是以文件形式提供的，则解析后的文件路径
type (str): 附件的内容类型，如果提供了的话
content (bytes): 附件的二进制内容，如果提供了的话

通常只会设置url、path或content中的一个。

通常你应该通过以下方法之一访问类型和内容：

attachment.resolve_type() -> str: 如果可用，返回type，否则尝试通过查看内容的前几个字节来猜测类型
attachment.content_bytes() -> bytes: 返回二进制内容，可能需要从文件读取或从URL获取
attachment.base64_content() -> str: 返回内容为base64编码的字符串

一个id()方法返回此内容的数据库ID，该ID是二进制内容的SHA256哈希值，或者在外部URL托管的附件的情况下，是{"url": url}的哈希值。这是一个实现细节，您不应该直接访问。

请注意，带有附件的提示可能完全不包含文本提示，在这种情况下，prompt.prompt 将是 None。

以下是OpenAI插件处理附件的方式，包括未提供prompt.prompt的情况：

if not prompt.attachments:
    messages.append({"role": "user", "content": prompt.prompt})
else:
    attachment_message = []
    if prompt.prompt:
        attachment_message.append({"type": "text", "text": prompt.prompt})
    for attachment in prompt.attachments:
        attachment_message.append(_attachment(attachment))
    messages.append({"role": "user", "content": attachment_message})


# And the code for creating the attachment message
def _attachment(attachment):
    url = attachment.url
    base64_content = ""
    if not url or attachment.resolve_type().startswith("audio/"):
        base64_content = attachment.base64_content()
        url = f"data:{attachment.resolve_type()};base64,{base64_content}"
    if attachment.resolve_type().startswith("image/"):
        return {"type": "image_url", "image_url": {"url": url}}
    else:
        format_ = "wav" if attachment.resolve_type() == "audio/wav" else "mp3"
        return {
            "type": "input_audio",
            "input_audio": {
                "data": base64_content,
                "format": format_,
            },
        }

如你所见，如果可用，它使用attachment.url，否则回退到使用base64_content()方法将图像直接嵌入到发送给API的JSON中。对于OpenAI API，音频附件始终以base64编码的字符串形式包含。

之前对话中的附件#

实现继续对话能力的模型可以使用response.attachments属性重建先前的消息JSON。

以下是OpenAI插件如何实现这一点：

for prev_response in conversation.responses:
    if prev_response.attachments:
        attachment_message = []
        if prev_response.prompt.prompt:
            attachment_message.append(
                {"type": "text", "text": prev_response.prompt.prompt}
            )
        for attachment in prev_response.attachments:
            attachment_message.append(_attachment(attachment))
        messages.append({"role": "user", "content": attachment_message})
    else:
        messages.append(
            {"role": "user", "content": prev_response.prompt.prompt}
        )
    messages.append({"role": "assistant", "content": prev_response.text_or_raise()})

那里使用的response.text_or_raise()方法将返回响应的文本，如果响应是尚未完全解析的AsyncResponse实例，则会引发ValueError异常。

这是一个稍微有点奇怪的黑客技巧，用于解决在同步和异步模型中共享构建messages列表逻辑的常见需求。

跟踪令牌使用情况#

按令牌收费的模型应跟踪每个提示使用的令牌数量。response.set_usage() 方法可用于记录响应使用的令牌数量 - 这些数据将通过 Python API 提供，并记录到 SQLite 数据库中供命令行用户使用。

response 这里是传递给 .execute() 作为参数的响应对象。

在你的.execute()方法结束时调用response.set_usage()。它接受关键字参数input=、output=和details=——这三个都是可选的。input和output应该是整数，而details应该是一个字典，提供输入和输出令牌计数之外的额外信息。

此示例记录了15个输入令牌，340个输出令牌，并指出有37个令牌被缓存：

response.set_usage(input=15, output=340, details={"cached": 37})