AI开发必备的3个DSPy技巧

LIBRARY Dec 30, 2024

推荐： ComfyUI-PandasAI | ComfyUI-Autodistill | ComfyUI节点速查手册

DSPy 是我的首选框架，因为它简单且设计周到。我使用它做了很多事情，从构建 MVP 到扩展系统以处理数百万个请求。虽然 DSPy 和 AI 编程仍在不断发展，但找到清晰、可用于生产的指导可能很困难。

在这篇博客中，我分享了我所学到的东西——从实践经验到文档和 GitHub 讨论中的见解。这是一本实用指南，里面充满了使用 DSPy 的可行技巧。

1、可靠的输出

对于使用 LLM 的开发人员来说，最大的挑战是实现结构化和一致的输出。让你的代理工作是一回事，但确保 95% 以上的可靠性是另一回事。最终用户每次都希望获得良好的结果 — 他们不在乎 LLM 是否容易出现幻觉、不一致或偶尔出现不稳定行为。

虽然有些问题（如 API 提供商中断）超出了你的控制范围，但你可以在自己这边解决许多问题。使用 DSPy，你可以使用断言和建议来指导你的程序获得更好的输出并提高可靠性。这些工具还有助于捕获不良结果并实施自动重试机制以确保高质量的响应。

如何使用 dspy.Assert 和 dspy.Suggest ？

要在 DSPy 中使用 dspy.Assert 和 dspy.Suggest，你需要使用布尔验证检查在程序中定义约束以获得所需的结果。这些约束被实现为验证模型输出的简单 Python 函数。

主要区别在于它们的严格性：

Assert 强制执行程序必须满足的严格条件；否则，它将失败。
Suggest 提供非强制性建议以提高性能或输出质量。

以下是如何验证简单 DSPy 模块的示例：

from dspy.primitives.assertions import assert_transform_module, backtrack_handler
# Define a validation function
# You can define you our validation function that is specific to your program

# Overly simplistic validation function
def your_validation_fn(model_outputs):
    return model_outputs.one_word_answer == "Islamabad"

# A simple DSPy program that gives a one-word answer to question
your_module = dspy.ChainOfThought("question->one_word_answer")

model_outputs = my_module(question="What is the capital of Pakistan?")

# Add assertions to your program

dspy.Assert(your_validation_fn(model_outputs), "Validation failed", target_module=your_module)
dspy.Suggest(your_validation_fn(model_outputs), "Consider revising output", target_module=your_module)

你甚至可以将断言添加到自定义程序的 forward 函数中。定义这些约束后，你可以使用 assert_transform_module 函数和 backtrack_handler 将 DSPy 模块与断言包装在一起，将它们集成到程序中。此函数会转换你的程序以包含内部断言回溯和重试逻辑。

class one_word_answer(dspy.Module):
    def __init__(self):
        self.one_word_program = dspy.ChainOfThought("question->one_word_answer")

    def forward(self, question):
        # send the question to the one_word_program
        response = self.one_word_program(question=question)
        # Assertion checks if the output is a single word
        dspy.Assert(response.one_word_answer.count(" ") == 0, "Validation failed, not one word", target_module=self.one_word_program)
        return response

my_module = one_word_answer()

# 1. way to activate assertions in the module, so whenever assertion fails, it will throw error
program_with_assertions = assert_transform_module(my_module(), backtrack_handler)

# 2. way to activate assertions

program_with_assertions = my_module().activate_assertions()

# Whenever the output is not a one word, or has space, then program will throw
# an error.

如果输出验证失败，你将收到错误，你可以捕获此错误并实施解决方案

2、使用多个大型语言模型

大型程序通常需要多个 LLM。原因如下：

成本：当你使用顶级模型处理 1000 个请求时，会导致高昂的成本，因为其中许多请求可以用低端模型来处理。
速度：所有 LLM API 提供商都有基于层级的请求限制，但你的用户请求需要及时得到处理。你可以将请求发送到不同的 API 或模型。
输出质量：你可以查看 LLM 基准，看看哪个模型在处理某些请求方面做得更好。有些模型在代码生成方面表现更好（尤其是在调整成本后）。
系统过载：许多企业用例更喜欢本地托管的 LLM，将所有请求发送到一个 LLM容易导致系统过载。

在 DSPy 中，你可以定义一个全局 LLM，这是系统在你发出请求时使用的默认 LLM。你可以像这样定义它：

lm = dspy.LM('openai/gpt-4o', model_type='chat', max_tokens =1000, api_key = '<>')
dspy.configure(lm=lm)

你可以像这样切换 LM，但每次更改全局设置都会导致问题，尤其是在你发出并发请求时（大型系统很常见）。

幸运的是，你可以使用上下文管理器像这样切换 LLM。它也是线程安全的，可以异步工作。

# Define LLMs 
lm1 = dspy.LM()
lm2 - dspy.LM()
lm3 = dspy.LM()

# you can define these conditions inside your DSPy Module
# The system switches based on any defined condition
if condition1:
  with dspy.settings.context(lm=lm1):
    response = dspy.Predict(..) 
if condition2:
  with dspy.settings.context(lm=lm2):
    response = dspy.Predict(..)
if condition3:
  with dspy.settings.context(lm=lm3):
    response = dspy.Predict(..)

多个 LLM 很好，如果要服务用户的多个请求怎么办？下一章将教你如何使用 DSPy 进行异步或并行请求。

3、并发或异步请求

构建最小可行原型很简单，因为它们通常一次只服务一个用户。但是当你需要扩展以处理数万甚至数百万个请求时会发生什么？在这种情况下，你可以利用线程或异步请求来有效地管理负载。方法如下：

3.1 使用 DSPy 进行“异步化”

此代码片段将帮助你在程序中发出异步请求：

import asyncio

# set the maximum number of async workers
dspy.settings.configure(async_max_workers=4) # default is 8

# define async version of your module
# The system was defined previously but can be any dspy.Module
async_system = dspy.asyncify(one_word_answer())

# a simple async program
async def get_all_responses(questions):
# Creates couroutines to be awaited later
  tasks = [async_system(question=q) for q in questions]
# You can also asyncio.wait, as_completed, wait for etc
  responses = await asyncio.gather(*tasks)
  return responses



questions = [
    "What is your name?",
    "How old are you?",
    "Where are you from?",
    "What is your favorite hobby?",
    "What do you do for a living?",
    "What is your favorite programming language?",
    "What are your goals for the next year?",
    "What is your favorite book or movie?",
    "Have you ever traveled abroad?",
    "What motivates you to learn new skills?"
]

# run the async program
all_responses = asyncio.run(get_all_responses(questions))

上面的代码片段允许你异步等待响应。你可以更改 async_max_worker 参数以同时扩展到 1000 个工作线程。

3.2 使用 DSPy 进行线程处理

在最新版本的 dspy 中，每个模块都有一个内置方法 batch，允许你并行处理请求。

以下是该方法的输入：

代码片段：

import dspy

# Create questions for analysis
questions = [
    "What is the relationship between customer lifetime value and transaction frequency?",
    "How does the average transaction value impact customer retention?",
    "Which customer segments show the highest predicted CLV and why?",
    "What are the key factors influencing the accuracy of our CLV predictions?",
    "How can we improve our CLV prediction model based on the current results?"
]


# Create a small DSPy Module, which gives one word answers
system = dspy.ChainOfThought("question->one_word_answer")

# Creating examples, which can be sent to module.batch
questions = [dspy.Example(question=q).with_inputs('question') for q in questions]

# Sends all the 5 questions in parallel, with certain settings
responses = system.batch(questions, num_threads = 3,return_failed_examples=True,max_errors=3)