2026年7月2日 周四晚上19:30,报名腾讯会议了解“如何构建自进化的动态知识库(Brain)”(限30人)
免费POC, 零成本试错
FDE知识库

FDE知识库

学习大模型的前沿技术与行业落地应用


收藏

丝分缕解!带你了解DSPy核心模块源码实现原理之01篇-Signature类

发布日期:2024-07-15 14:57:45 浏览次数: 3291
作者:AI应用研究Lab

微信搜一搜,关注“AI应用研究Lab”


Prompt或许的新未来, DSPy使用从0到1快速上手以示例驱动的方式,由浅入深的介绍DSPy框架自身的使用流程,并结合知名大模型应用框架LangChain,进一步介绍了两个框架结合的案例。本文将通过示例和源码解析的方式,解读DSPy运行原理。
DSPy是斯坦福 NLP 组推出的一个用于优化和生成Prompt的框架,DSPy将传统手工编写提示词的方式抽象为高代码编程方式,其核心思路为:

(1)将整体流程与单步骤分开。每个步骤聚焦具体的工作,协同完成Prompt的优化

(2)引入多样的优化器,这些优化器是由L驱动的算法,可以根据定义的阈值来调整调用LM的提示及访问参数。

上述步骤主要由 Signature、Module、Optimizer三个核心模块实现。

DSPy 实现原理之Signature类  

DSPy的Signature类,是DSPy核心模块之一,Signature是声明性的规范,定义了 DSPy 模块的输入/输出行为,用于告诉语言模型应执行哪些任务,而不是我们应如何设置 prompt 语言模型。
一个签名包括三个基本元素:
  • 语言模型旨在解决的子任务的简洁描述。

  • 我们提供给语言模型的一个或多个输入字段的描述(例如,输入问题)。

  • 我们期望从语言模型得到的一个或多个输出字段的描述(例如,问题的答案)。
Signature的基本用法示例 
首先来看一个示例来解释Signature的用法,需要准备签名所需的三要素,下面代码进行三步操作:

1、将子任务描述填入函数下面的注释中。

2、定义输入字段(必选),并对输入字段设置描述(可选)。

3、定义输出字段(必选),并对输出字段设置描述(可选)。
class QA(dspy.Signature):    """answer the question of user"""
    user_question = dspy.InputField(desc="用户的问题")    answer = dspy.OutputField()
然后我们使用dspy.ChainOfThought 进行一次推理,并查看提示词最终的内容
question = "what is the color of the sea?"summarize = dspy.ChainOfThought(QA)response = summarize(question=question)#  查看提示词lm.inspect_history(n=1)
提示词的内容如下,通过提示词内容,可以看出Signature可以将类定义中的注释内容,转换为对这个子任务的描述填写到提示词开头部分,然后将输入输出字段,分别以统一的格式(首字母大写,单词用空格分开)排布在 Reasoning的前后,其中Reasoning的内容为 CoT模式的固定提示词。   
answer the question of user
---
Follow the following format.
User Question: 用户的问题ReasoningLet's think step by step in order to ${produce the answer}. We ...Answer: ${answer}
---
Question:what is the color of the sea?ReasoningLet's think step by step in order to Question: what is the color of the sky?ReasoningLet's think step by step in order to determine the color of the sky. The sky appears blue due to the scattering of light waves in the atmosphere. The blue light is scattered more efficiently than other colors of light, which is why the sky appears blue.AnswerBlue
通过上面的例子,我们了解了如何通过写代码的方式,产生提示词,并使得提示词包括我们对任务描述和输入输出参数描述。
此外,Signature还可以通过字符串定义输入输出方式,代替通过继承方式定义,代码如下:
summarize = dspy.ChainOfThought('question -> answer')
但是这种方式只能定义输入输出的字段名,无法定义任务描述 和 字段描述,此时产生的提示词的任务描述是默认任务描述,提示词如下:   
Given the fields `question`, produce the fields `answer`.
---
Follow the following format.
Question: ${question}Reasoning: Let's think step by step in order to ${produce the answer}. We ...Answer: ${answer}
---
至此,我们介绍了Signature的两种用法。

Signature源码解析 

那么,Signature类是如何实现上述功能的呢?
我们将通过源码解释以下问题:
  • Signature执行流程

  • 字符串如何转换为Signature类

  • Signature的变量如何转换为提示词中的内容
我们深入Signature类的代码中进行简单的解释(文件位置:dspy\signatures):
# signature.py 部分删减import astimport reimport typesimport typingfrom copy import deepcopyfrom typing import Any, Dict, Tuple, Type, Union  # noqa: UP035
from pydantic import BaseModel, Field, create_modelfrom pydantic.fields import FieldInfo
import dspfrom dspy.signatures.field import InputField, OutputField, new_to_old_field

class SignatureMeta(type(BaseModel)):   # Signature元类    def __call__(cls, *args, **kwargs):  # noqa: ANN002        if cls is Signature:            return make_signature(*args, **kwargs)        return super().__call__(*args, **kwargs)
    def __new__(mcs, signature_name, bases, namespace, **kwargs):  # noqa: N804     # 初始化Signature时调用该函数        # Set `str` as the default type for all fields        raw_annotations = namespace.get("__annotations__", {})        for name, field in namespace.items():            if not isinstance(field, FieldInfo):                continue  # Don't add types to non-field attributes            if not name.startswith("__") and name not in raw_annotations:                raw_annotations[name] = str        namespace["__annotations__"] = raw_annotations
        # Let Pydantic do its thing        cls = super().__new__(mcs, signature_name, bases, namespace, **kwargs)
        # If we don't have instructions, it might be because we are a derived generic type.        # In that case, we should inherit the instructions from the base class.        if cls.__doc__ is None:            for base in bases:                if isinstance(base, SignatureMeta):                    doc = getattr(base, "__doc__", "")                    if doc != "":                        cls.__doc__ = doc
        # The more likely case is that the user has just not given us a type.        # In that case, we should default to the input/output format.        if cls.__doc__ is None:            cls.__doc__ = _default_instructions(cls)
        # Ensure all fields are declared with InputField or OutputField        cls._validate_fields()
        # Ensure all fields have a prefix        for name, field in cls.model_fields.items():            if "prefix" not in field.json_schema_extra:                field.json_schema_extra["prefix"] = infer_prefix(name) + ":"            if "desc" not in field.json_schema_extra:                field.json_schema_extra["desc"] = f"${{{name}}}"
        return cls
    ...

class Signature(BaseModel, metaclass=SignatureMeta):    ""  # noqa: D419
    # Note: Don't put a docstring here, as it will become the default instructions    # for any signature that doesn't define it's own instructions.    pass
def make_signature(         # 根据给定的参数创建Signature 实例    signature: Union[str, Dict[str, Tuple[type, FieldInfo]]],    instructions: str = None,    signature_name: str = "StringSignature",) -> Type[Signature]:    """Create a new Signature type with the given fields and instructions.
    Note:        Even though we're calling a type, we're not making an instance of the type.        In general, instances of Signature types are not allowed to be made. The call        syntax is provided for convenience.
    Args:        signature: The signature format, specified as "input1, input2 -> output1, output2".        instructions: An optional prompt for the signature.        signature_name: An optional name for the new signature type.    """    fields = _parse_signature(signature) if isinstance(signature, str) else signature
    # Validate the fields, this is important because we sometimes forget the    # slightly unintuitive syntax with tuples of (type, Field)    fixed_fields = {}    for name, type_field in fields.items():        if not isinstance(name, str):            raise ValueError(f"Field names must be strings, not {type(name)}")        if isinstance(type_field, FieldInfo):            type_ = type_field.annotation            field = type_field        else:            if not isinstance(type_field, tuple):                raise ValueError(f"Field values must be tuples, not {type(type_field)}")            type_, field = type_field        # It might be better to be explicit about the type, but it currently would break        # program of thought and teleprompters, so we just silently default to string.        if type_ is None:            type_ = str        # if not isinstance(type_, type) and not isinstance(typing.get_origin(type_), type):        if not isinstance(type_, (type, typing._GenericAlias, types.GenericAlias)):            raise ValueError(f"Field types must be types, not {type(type_)}")        if not isinstance(field, FieldInfo):            raise ValueError(f"Field values must be Field instances, not {type(field)}")        fixed_fields[name] = (type_, field)
    # Fixing the fields shouldn't change the order    assert list(fixed_fields.keys()) == list(fields.keys())  # noqa: S101
    # Default prompt when no instructions are provided    if instructions is None:        sig = Signature(signature, "")  # Simple way to parse input/output fields        instructions = _default_instructions(sig)
    return create_model(        signature_name,        __base__=Signature,        __doc__=instructions,        **fixed_fields,    )
def _parse_signature(signature: str) -> Tuple[Type, Field]:     # 将字符串形式的输入输出转为对象    if signature.count("->") != 1:        raise ValueError(f"Invalid signature format: '{signature}', must contain exactly one '->'.")
    fields = {}    inputs_str, outputs_str = map(str.strip, signature.split("->"))    inputs = [v.strip() for v in inputs_str.split(",") if v.strip()]    outputs = [v.strip() for v in outputs_str.split(",") if v.strip()]    for name_type in inputs:        name, type_ = _parse_named_type_node(name_type)        fields[name] = (type_, InputField())    for name_type in outputs:        name, type_ = _parse_named_type_node(name_type)        fields[name] = (type_, OutputField())
    return fields
def infer_prefix(attribute_name: str) -> str:       # 同意在提示词中的格式,如首字母大写等    """Infer a prefix from an attribute name."""    # Convert camelCase to snake_case, but handle sequences of capital letters properly    s1 = re.sub("(.)([A-Z][a-z]+)", r"\1_\2", attribute_name)    intermediate_name = re.sub("([a-z0-9])([A-Z])", r"\1_\2", s1)
    # Insert underscores around numbers to ensure spaces in the final output    with_underscores_around_numbers = re.sub(        r"([a-zA-Z])(\d)",        r"\1_\2",        intermediate_name,    )    with_underscores_around_numbers = re.sub(        r"(\d)([a-zA-Z])",        r"\1_\2",        with_underscores_around_numbers,    )
    # Convert snake_case to 'Proper Title Case', but ensure acronyms are uppercased    words = with_underscores_around_numbers.split("_")    title_cased_words = []    for word in words:        if word.isupper():            title_cased_words.append(word)        else:            title_cased_words.append(word.capitalize())
    return " ".join(title_cased_words)
1、执行流程
从代码中代码中可以看出,Signature类集成了pydantic.BaseModel,并设置了元类 SignatureMeta;pydantic.BaseModel是格式校验的类;SignatureMeta是自定义类,Signature的大部分提示词逻辑都来自SignatureMeta类。   
在Signature类及子类初始化时,首先会调用 SignatureMeta的__call__ 函数,__call__中调用了 make_signature函数,该函数主要是解析输出的字段,最终调用pydantic.create_model函数创建pydantic的格式类(__call__ -> make_signature -> pydantic.create_model),格式类中已经包含了输入输出字段以及对应的字段描述。
然后调用SignatureMeta的 __new__函数,该函数将子任务描述,传入到 cls.__doc__变量中,并且利用infer_prefix函数,修改了输入输出字段的格式,使他们统一为首字母大写的形式,存入到 field.json_schema_extra["prefix"] 中,将字段的描述,存入 field.json_schema_extra["desc"] 字段中,并返回类。
2、字符串如何转为Signature类
如果输入为字符串类型,则 make_signature中的_parse_signature 函数会 格式化这个字符串并转换为 dspy.InputFiled 或 dspy.OutputField类型,至此,通过字符串和Signature类定义的方式都统一成了相同类型。
3、Signature的变量如何转换为提示词中的内容

在初始化时,调用了SignatureMeta类的__call__ 函数和 __new__函数,两个函数创建了pydantic.BaseModal类,并将输入输出字段进行格式化,将任务描述存入 cls.__doc__,至此,所有代码描述的注释和变量,都转变为了提示词中将要用到的字符串内容,在 dspy.Module执行时,则会对提示词进行填充。


53AI,企业落地大模型首选服务商

产品:场景落地咨询+大模型应用平台+行业解决方案

承诺:免费POC验证,效果达标后再合作。零风险落地应用大模型,已交付160+中大型企业

联系我们

售前咨询
186 6662 7370
预约演示
185 8882 0121

微信扫码

添加专属顾问

回到顶部

加载中...

扫码咨询

扫码登录
登录即表示您同意《53AI网站服务协议》
服务协议

欢迎您使用【53AI 官方网站】(以下简称“本网站”或“我们”)。本《会员服务协议》(以下简称“本协议”)是您(以下简称“会员”或“用户”)与【深圳市博思协创网络科技有限公司】之间关于注册、登录及使用本网站会员服务所订立的法律协议。

在您注册或登录前,请务必审慎阅读、充分理解各条款内容,特别是免除或限制责任的条款、知识产权条款、争议解决条款等。此类条款将以加粗形式提示您注意。 当您通过微信公众号授权、手机验证码验证或其他方式成功登录本网站时,即视为您已完全理解并同意接受本协议的全部内容。

一、 定义

本网站:指由【深圳市博思协创网络科技有限公司】运营的,域名为【53ai.com】的网站及相关移动端页面。

会员服务:指本网站向注册会员提供的知识库文章查阅、内容检索及其他相关增值服务。

知识库内容:指本网站发布的包括但不限于文字、图表、数据、研究报告、行业分析等数字化内容资源。

二、 账号注册与登录

登录方式:本网站支持以下登录方式,您可根据实际情况选择:

微信公众号授权登录:您同意将您的微信OpenID信息授权给本网站,用于创建或关联会员账号。

手机验证码登录:您需提供真实有效的手机号码,并通过短信验证码完成身份验证与登录/注册。

账号安全:您的账号仅限您本人使用,禁止赠与、借用、租用、转让或售卖。因您保管不善导致的账号被盗、密码泄露等损失,由您自行承担。

实名认证:根据相关法律法规要求,我们可能要求您在特定功能下完成实名认证。如您拒绝提供,可能无法使用部分或全部服务。

未成年人保护:若您未满18周岁,请在法定监护人的陪同下阅读本协议,并在征得监护人同意后使用本服务。

三、 服务内容与规范

知识库查阅权限:会员登录后,有权按照其会员等级对应的权限范围,在线浏览、检索本网站知识库中的相关文章及内容。

服务变更:我们有权根据业务发展需要,调整、变更或终止部分服务内容,并将以网站公告、公众号消息等方式提前通知。

禁止行为:您在使用服务时不得实施以下行为:

利用技术手段批量爬取、下载、转存知识库内容;

将知识库内容用于商业目的或未经授权地向第三方传播;

干扰本网站正常运行或侵犯其他用户合法权益;

发布违法违规信息或从事违反公序良俗的活动。

四、 知识产权声明

权利归属:本网站知识库中的排版设计、软件代码等内容的知识产权均归【公司全称】或原权利人所有,受《中华人民共和国著作权法》等法律保护。

有限许可:本网站授予会员一项非独占、不可转让、不可转授权的普通许可,仅限于个人学习、研究之目的在线查阅知识库内容。

侵权追责:未经书面许可,任何单位或个人不得以任何形式复制、转载、摘编、镜像、汇编或以其他方式使用上述内容。一经发现,我们保留追究其法律责任的权利。

五、 个人信息保护

我们重视对您个人信息的保护。关于我们如何收集、使用、存储和保护您的个人信息,请单独阅读 《隐私政策》。

您通过微信公众号授权或手机号验证所提供的信息,我们将严格按照《个人信息保护法》的规定处理,仅用于身份识别、服务提供及安全验证等必要用途。

您可以随时通过网站设置或联系客服行使查阅、更正、删除个人信息及撤回授权同意的权利。

六、 免责声明

内容准确性:知识库内容仅供参考,不构成专业建议。我们不对其完整性、准确性、时效性作任何明示或暗示的保证,您应自行判断并承担使用风险。

不可抗力:因自然灾害、政策法规变化、网络故障、第三方平台接口异常(如微信接口维护、运营商短信通道故障)等不可抗力导致的服务中断或延迟,我们不承担违约责任。

第三方链接:本网站可能包含指向第三方网站的链接,该等网站的内容和服务不受我们控制,请您自行甄别风险。

七、 违约责任

如您违反本协议约定,我们有权视情节采取警告、限制功能、暂停服务、注销账号等措施,并保留要求赔偿损失的权利。

如因您的违约行为导致我们遭受行政处罚、第三方索赔或商誉损失,您应承担全部赔偿责任(包括但不限于罚款、赔偿金、律师费、公证费等)。

八、 法律适用与争议解决

本协议的订立、执行和解释均适用中华人民共和国大陆地区法律。

因本协议产生的或与本协议有关的任何争议,双方应友好协商解决;协商不成的,任何一方均可向【公司所在地】有管辖权的人民法院提起诉讼。

九、 其他

本协议构成双方就本服务达成的完整协议,取代此前任何口头或书面约定。

本协议任一条款被认定为无效或不可执行的,不影响其他条款的效力。

我们对本协议享有最终解释权,并在法律允许的范围内保留随时修改的权利。修改后的协议一经公布即生效,继续使用服务即视为同意修订内容。


已查阅