(1)将整体流程与单步骤分开。每个步骤聚焦具体的工作,协同完成Prompt的优化
上述步骤主要由 Signature、Module、Optimizer三个核心模块实现。
DSPy 实现原理之Signature类
-
语言模型旨在解决的子任务的简洁描述。
-
我们提供给语言模型的一个或多个输入字段的描述(例如,输入问题)。
-
我们期望从语言模型得到的一个或多个输出字段的描述(例如,问题的答案)。
Signature的基本用法示例 1、将子任务描述填入函数下面的注释中。
2、定义输入字段(必选),并对输入字段设置描述(可选)。
class QA(dspy.Signature):"""answer the question of user"""user_question = dspy.InputField(desc="用户的问题")answer = dspy.OutputField()
question = "what is the color of the sea?"summarize = dspy.ChainOfThought(QA)response = summarize(question=question)# 查看提示词lm.inspect_history(n=1)
answer the question of user---Follow the following format.User Question: 用户的问题Reasoning: Let's think step by step in order to ${produce the answer}. We ...Answer: ${answer}---Question:what is the color of the sea?Reasoning: Let's think step by step in order to Question: what is the color of the sky?Reasoning: Let'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.Answer: Blue
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.py 部分删减import astimport reimport typesimport typingfrom copy import deepcopyfrom typing import Any, Dict, Tuple, Type, Union # noqa: UP035from pydantic import BaseModel, Field, create_modelfrom pydantic.fields import FieldInfoimport dspfrom dspy.signatures.field import InputField, OutputField, new_to_old_fieldclass SignatureMeta(type(BaseModel)): # Signature元类def __call__(cls, *args, **kwargs): # noqa: ANN002if 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 fieldsraw_annotations = namespace.get("__annotations__", {})for name, field in namespace.items():if not isinstance(field, FieldInfo):continue # Don't add types to non-field attributesif not name.startswith("__") and name not in raw_annotations:raw_annotations[name] = strnamespace["__annotations__"] = raw_annotations# Let Pydantic do its thingcls = 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 OutputFieldcls._validate_fields()# Ensure all fields have a prefixfor 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.passdef 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 callsyntax 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.annotationfield = type_fieldelse: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 orderassert list(fixed_fields.keys()) == list(fields.keys()) # noqa: S101# Default prompt when no instructions are providedif instructions is None:sig = Signature(signature, "") # Simple way to parse input/output fieldsinstructions = _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 fieldsdef infer_prefix(attribute_name: str) -> str: # 同意在提示词中的格式,如首字母大写等"""Infer a prefix from an attribute name."""# Convert camelCase to snake_case, but handle sequences of capital letters properlys1 = 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 outputwith_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 uppercasedwords = 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)
在初始化时,调用了SignatureMeta类的__call__ 函数和 __new__函数,两个函数创建了pydantic.BaseModal类,并将输入输出字段进行格式化,将任务描述存入 cls.__doc__,至此,所有代码描述的注释和变量,都转变为了提示词中将要用到的字符串内容,在 dspy.Module执行时,则会对提示词进行填充。


