llamafactory/data/formatter.py [源码解析]

llamafactory/data/formatter.py [源码解析]
import json
import re
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import List, Literal, Optional, Tuple, Union

from .data_utils import SLOTS
from .tool_utils import DefaultToolUtils, GLM4ToolUtils


@dataclass
class Formatter(ABC):
    slots: SLOTS = field(default_factory=list)
    tool_format: Optional[Literal["default", "glm4"]] = None

    @abstractmethod
    def apply(self, **kwargs) -> SLOTS: ...

    def extract(self, content: str) -> Union[str, List[Tuple[str, str]]]:
        raise NotImplementedError


@dataclass
class EmptyFormatter(Formatter):
    def __post_init__(self):
        has_placeholder = False
        for slot in filter(lambda s: isinstance(s, str), self.slots):
            if re.search(r"\{\{[a-zA-Z_][a-zA-Z0-9_]*\}\}", slot):
                has_placeholder = True

        if has_placeholder:
            raise ValueError("Empty formatter should not contain any placeholder.")

    def apply(self, **kwargs) -> SLOTS:
        return self.slots

我会为您逐行解释这段代码:

python

复制

from .data_utils import SLOTS
from .tool_utils import DefaultToolUtils, GLM4ToolUtils

这些是导入语句,从当前包的其他模块导入所需的类型和工具。

python

复制

@dataclass
class Formatter(ABC):

定义了一个名为Formatter的抽象基类,使用了@dataclass装饰器。

python

复制

    slots: SLOTS = field(default_factory=list)

定义了一个名为slots的属性,类型为SLOTS,默认值是一个空列表。

python

复制

    tool_format: Optional[Literal["default", "glm4"]] = None

定义了一个名为tool_format的可选属性,可以是"default"或"glm4",默认为None。

python

复制

    @abstractmethod
    def apply(self, **kwargs) -> SLOTS: ...

定义了一个抽象方法apply,它接受任意关键字参数并返回SLOTS类型。子类必须实现这个方法。

python

复制

    def extract(self, content: str) -> Union[str, List[Tuple[str, str]]]:
        raise NotImplementedError

定义了一个extract方法,但默认实现抛出NotImplementedError。子类可以选择性地重写这个方法。

python

复制

@dataclass
class EmptyFormatter(Formatter):

定义了一个名为EmptyFormatter的类,它继承自Formatter

python

复制

    def __post_init__(self):

这是一个在对象初始化后自动调用的方法。

python

复制

        has_placeholder = False
        for slot in filter(lambda s: isinstance(s, str), self.slots):
            if re.search(r"\{\{[a-zA-Z_][a-zA-Z0-9_]*\}\}", slot):
                has_placeholder = True

这段代码检查self.slots中的字符串是否包含占位符(形如{{name}})。

python

复制

        if has_placeholder:
            raise ValueError("Empty formatter should not contain any placeholder.")

如果发现占位符,抛出ValueError异常。这是因为EmptyFormatter不应包含任何占位符。

python

复制

    def apply(self, **kwargs) -> SLOTS:
        return self.slots

EmptyFormatterapply方法简单地返回self.slots,不进行任何处理。

这段代码定义了一个格式化器的基础结构。Formatter是一个抽象基类,定义了格式化器应该具有的基本接口。EmptyFormatter是一个具体的实现,它不对输入进行任何处理,只是简单地返回原始的slots。这种设计允许在需要时创建更复杂的格式化器,同时保持了一个统一的接口。

@dataclass
class StringFormatter(Formatter):
    def __post_init__(self):
        has_placeholder = False
        for slot in filter(lambda s: isinstance(s, str), self.slots):
            if re.search(r"\{\{[a-zA-Z_][a-zA-Z0-9_]*\}\}", slot):
                has_placeholder = True

        if not has_placeholder:
            raise ValueError("A placeholder is required in the string formatter.")

    def apply(self, **kwargs) -> SLOTS:
        elements = []
        for slot in self.slots:
            if isinstance(slot, str):
                for name, value in kwargs.items():
                    if not isinstance(value, str):
                        raise RuntimeError("Expected a string, got {}".format(value))

                    slot = slot.replace("{{" + name + "}}", value, 1)
                elements.append(slot)
            elif isinstance(slot, (dict, set)):
                elements.append(slot)
            else:
                raise RuntimeError("Input must be string, set[str] or dict[str, str], got {}".format(type(slot)))

        return elements

我会为您逐行解释这段代码:

python

复制

@dataclass
class StringFormatter(Formatter):

这定义了一个名为StringFormatter的类,它继承自Formatter类。@dataclass装饰器用于自动生成一些常用的方法,如__init____repr__等。

python

复制

    def __post_init__(self):

这是一个特殊方法,在对象初始化后自动调用。它用于执行一些额外的初始化逻辑。

python

复制

        has_placeholder = False
        for slot in filter(lambda s: isinstance(s, str), self.slots):
            if re.search(r"\{\{[a-zA-Z_][a-zA-Z0-9_]*\}\}", slot):
                has_placeholder = True

这段代码检查self.slots中的字符串是否包含占位符(形如{{name}})。它使用正则表达式来查找这种模式。

python

复制

        if not has_placeholder:
            raise ValueError("A placeholder is required in the string formatter.")

如果没有找到占位符,抛出一个ValueError异常。

python

复制

    def apply(self, **kwargs) -> SLOTS:

定义apply方法,接受任意关键字参数,返回类型为SLOTS(这可能是在其他地方定义的类型)。

python

复制

        elements = []

初始化一个空列表来存储处理后的元素。

python

复制

        for slot in self.slots:
            if isinstance(slot, str):

遍历self.slots中的每个元素。如果元素是字符串:

python

复制

                for name, value in kwargs.items():
                    if not isinstance(value, str):
                        raise RuntimeError("Expected a string, got {}".format(value))

                    slot = slot.replace("{{" + name + "}}", value, 1)

用提供的关键字参数替换字符串中的占位符。如果值不是字符串,抛出RuntimeError

python

复制

                elements.append(slot)

将处理后的字符串添加到elements列表。

python

复制

            elif isinstance(slot, (dict, set)):
                elements.append(slot)

如果slot是字典或集合,直接添加到elements列表。

python

复制

            else:
                raise RuntimeError("Input must be string, set[str] or dict[str, str], got {}".format(type(slot)))

如果slot既不是字符串也不是字典或集合,抛出RuntimeError

python

复制

        return elements

返回处理后的elements列表。

总的来说,这个StringFormatter类用于处理包含占位符的字符串模板。它确保模板中至少有一个占位符,并提供了一个apply方法来用实际值替换这些占位符。这种formatter可能用于生成结构化的文本,如聊天模板或其他需要动态内容填充的场景。

Read more

科普文:软件架构Nginx系列之【nginx配置https详细操作说明】

科普文:软件架构Nginx系列之【nginx配置https详细操作说明】

一些企业对公司数据要求加密,用于安全措施。web加密也是其中一个重要的环节,所以需要加入自建证书。 1、安装nginx 1.1 依赖安装 yum -y install wget gcc gcc-c++ pcre-devel openssl-devel 1.2 nginx软件包下载 wget http://nginx.org/download/nginx-1.19.0.tar.gz 1.3 解压、编译、安装 [root@localhost ~]# tar xf nginx-1.19.0.tar.gz [root@localhost ~]# cd nginx-1.19.0

By Ne0inhk
科普文:软件架构Nginx系列之【nginx.conf配置文件结构详细说明】

科普文:软件架构Nginx系列之【nginx.conf配置文件结构详细说明】

nginx.conf配置文件结构 如上图所示:nginx.conf主要由events、http、server、location、upstream等块配置项和一些行配置项组成。 Nginx配置文件(conf/nginx.conf)整体分为三部分 全局块 ---- 和网络连接相关的配置 events块 ---- 和网络连接相关的配置 http块 ---- 代理、缓存、日志记录、虚拟主机配置 . http全局块 . Server块 ---- Server全局块 ---- location块 注意: http块中可以配置多个Server块,每个Server块中可以配置多个location块。 Nginx配置文件是一个文本文件,通常位于/etc/nginx/nginx.conf,用于定义Nginx服务器的行为和功能。该文件使用简洁而灵活的语法,主要分为以下几个部分: * ‌全局块‌:配置影响Nginx全局的指令,如用户、工作进程数、进程权限等。 * ‌事件配置块‌

By Ne0inhk
科普文:软件架构Nginx系列之【Nginx结合Openresty通过Lua+Redis实现动态封禁IP:限流】

科普文:软件架构Nginx系列之【Nginx结合Openresty通过Lua+Redis实现动态封禁IP:限流】

需求 为了封禁某些爬虫或者恶意用户对服务器的请求,我们需要建立一个动态的 IP 黑名单。 对于黑名单中的 IP ,我们将拒绝提供服务。 并且可以设置封禁失效时间 环境准备 * linux version: centos7 / ubuntu 等 * redis version: 5.0.5 * nginx version: nginx-openresty 设计方案 实现 IP 黑名单的功能有很多途径: 1、在操作系统层面,配置 iptables,来拦截指定 IP 的网络请求。 * 优点:简单直接,在服务器物理层面上进行拦截 * 缺点:每次需要手动上服务器修改配置文件,操作繁琐且不灵活 2、在 Web 服务器层面,通过 Nginx 自身的 deny 选项或者 lua

By Ne0inhk
科普文:软件架构Nginx系列之【Nginx的超时timeout配置详解】

科普文:软件架构Nginx系列之【Nginx的超时timeout配置详解】

Nginx 处理的每个请求均有相应的超时设置。如果做好这些超时时间的限定,判定超时后资源被释放,用来处理其他的请求,以此提升 Nginx 的性能。 keepalive_timeout HTTP 是一种无状态协议,客户端向服务器发送一个 TCP 请求,服务端响应完毕后断开连接。 如果客户端向服务器发送多个请求,每个请求都要建立各自独立的连接以传输数据。 HTTP 有一个 KeepAlive 模式,它告诉 webserver 在处理完一个请求后保持这个 TCP 连接的打开状态。若接收到来自客户端的其它请求,服务端会利用这个未被关闭的连接,而不需要再建立一个连接。 KeepAlive 在一段时间内保持打开状态,它们会在这段时间内占用资源。占用过多就会影响性能。 Nginx 使用 keepalive_timeout 来指定 KeepAlive 的超时时间(timeout)。指定每个 TCP 连接最多可以保持多长时间。Nginx 的默认值是 75 秒,有些浏览器最多只保持

By Ne0inhk