Skip to content

教程：DSPy中的调试与可观测性

本指南演示如何在DSPy中调试问题并提升可观测性。现代AI程序通常涉及多个组件，例如语言模型、检索器和工具。DSPy允许您以清晰模块化的方式构建和优化此类复杂AI系统。

然而，随着系统变得越来越复杂，理解系统正在做什么的能力变得至关重要。缺乏透明度会使预测过程很容易变成一个黑盒，导致故障或质量问题难以诊断，生产维护也变得具有挑战性。

在本教程结束时，您将了解如何使用MLflow Tracing调试问题并提升可观测性。您还将探索如何利用回调构建自定义日志记录解决方案。

定义一个程序

我们将从创建一个简单的ReAct智能体开始，它使用ColBERTv2的维基百科数据集作为检索源。你可以用一个更复杂的程序来替换它。

import dspy
import os

os.environ["OPENAI_API_KEY"] = "{your_openai_api_key}"

lm = dspy.LM("openai/gpt-4o-mini")
colbert = dspy.ColBERTv2(url="http://20.102.90.50:2017/wiki17_abstracts")
dspy.configure(lm=lm)


def retrieve(query: str):
    """Retrieve top 3 relevant information from ColBert"""
    results = colbert(query, k=3)
    return [x["text"] for x in results]


agent = dspy.ReAct("question -> answer", tools=[retrieve], max_iters=3)

现在，让我们向智能体提出一个简单的问题：

prediction = agent(question="Which baseball team does Shohei Ohtani play for in June 2025?")
print(prediction.answer)

Shohei Ohtani is expected to play for the Hokkaido Nippon-Ham Fighters in June 2025, based on the available information.

哦，这是不正确的。他不再为北海道日本火腿斗士队效力；他转会到了道奇队，并在2024年赢得了世界大赛！让我们调试程序并探索可能的修复方法。

使用 inspect_history

DSPy 提供了 inspect_history() 实用工具，用于打印出迄今为止所有进行的 LLM 调用：

# Print out 5 LLM calls
dspy.inspect_history(n=5)


[2024-12-01T10:23:29.144257]

系统消息:

您的输入字段为:
1. `question` (str)

...

响应:

响应:

[[ ## 推理过程 ## ]]
关于大谷翔平在2025年6月所属球队的信息搜索未获得具体结果。检索到的数据一致提到他效力于北海道日本火腿斗士队，但未提及指定日期其球队的任何变动或更新。鉴于信息缺乏，可以合理推断他可能仍留在北海道日本火腿斗士队，除非出现当前数据未捕捉到的未来进展。

[[ ## 答案 ## ]]
根据现有信息，预计大谷翔平将在2025年6月为北海道日本火腿斗士队效力。

[[ ## 完成 ## ]]
日志显示智能体未能从搜索工具获取有用信息。然而，检索器具体返回了什么内容？虽然有用，但inspect_history存在一些局限性：

  • 在现实世界的系统中，其他组件如检索器、工具和自定义模块扮演着重要角色，但inspect_history仅记录LLM调用。
  • DSPy程序通常在一个预测中会进行多次LLM调用。单一日志历史记录使得日志组织变得困难，尤其是在处理多个问题时。
  • 元数据（如参数、延迟以及模块之间的关系）未被捕获。

追踪 解决了这些限制并提供了一个更全面的解决方案。

追踪

MLflow 是一个端到端的机器学习平台，与 DSPy 无缝集成，以支持 LLMOps 的最佳实践。使用 MLflow 的自动追踪功能与 DSPy 结合非常简单；无需注册服务或提供 API 密钥。您只需安装 MLflow 并在您的笔记本或脚本中调用 mlflow.dspy.autolog()。

pip install -U mlflow>=2.18.0

安装完成后，通过以下命令启动您的服务器。

# It is highly recommended to use SQL store when using MLflow tracing
mlflow server --backend-store-uri sqlite:///mydb.sqlite

如果不通过--port标志指定不同的端口，你的MLflow服务器将在端口5000上托管。

现在让我们更改代码片段以启用MLflow追踪。我们需要：

  • 告诉MLflow服务器托管在哪里。
  • 应用 mlflow.autolog() 以便自动捕获DSPy追踪。

完整代码如下，现在让我们再次运行它！

import dspy
import os
import mlflow

os.environ["OPENAI_API_KEY"] = "{your_openai_api_key}"

# Tell MLflow about the server URI.
mlflow.set_tracking_uri("http://127.0.0.1:5000")
# Create a unique name for your experiment.
mlflow.set_experiment("DSPy")

lm = dspy.LM("openai/gpt-4o-mini")
colbert = dspy.ColBERTv2(url="http://20.102.90.50:2017/wiki17_abstracts")
dspy.configure(lm=lm)


def retrieve(query: str):
    """Retrieve top 3 relevant information from ColBert"""
    results = colbert(query, k=3)
    return [x["text"] for x in results]


agent = dspy.ReAct("question -> answer", tools=[retrieve], max_iters=3)
print(agent(question="Which baseball team does Shohei Ohtani play for?"))

MLflow 自动为每个预测生成一个trace，并将其记录在你的实验中。要可视化探索这些trace，请在浏览器中打开 http://127.0.0.1:5000/，然后选择你刚刚创建的实验并导航到Traces标签页：

MLflow Trace UI

点击最近的跟踪记录以查看其详细分解：

MLflow Trace View

在这里，你可以检查工作流中每一步的输入和输出。例如，上面的截图展示了retrieve函数的输入和输出。通过检查检索器的输出，你可以看到它返回了过时的信息，这不足以确定大谷翔平在2025年6月效力于哪支球队。你还可以检查其他步骤，例如语言模型的输入、输出和配置。

为了解决信息过时的问题，你可以用由Tavily search驱动的网络搜索工具替换retrieve函数。

from tavily import TavilyClient
import dspy
import mlflow

# Tell MLflow about the server URI.
mlflow.set_tracking_uri("http://127.0.0.1:5000")
# Create a unique name for your experiment.
mlflow.set_experiment("DSPy")

search_client = TavilyClient(api_key="<YOUR_TAVILY_API_KEY>")

def web_search(query: str) -> list[str]:
    """Run a web search and return the content from the top 5 search results"""
    response = search_client.search(query)
    return [r["content"] for r in response["results"]]

agent = dspy.ReAct("question -> answer", tools=[web_search])

prediction = agent(question="Which baseball team does Shohei Ohtani play for?")
print(agent.answer)

Los Angeles Dodgers

以下是一个GIF，展示如何在MLflow用户界面中进行导航：

MLflow Trace UI Navigation

关于如何使用MLflow追踪的完整指南，请参阅 MLflow Tracing Guide。

信息

MLflow 是一个端到端的 LLMOps 平台，提供实验跟踪、评估和部署等广泛功能。要了解更多关于 DSPy 和 MLflow 集成的信息，请访问 本教程。

构建自定义日志解决方案

有时，您可能希望实现自定义日志记录解决方案。例如，您可能需要记录由特定模块触发的特定事件。DSPy的回调机制支持此类用例。BaseCallback类提供了多个处理器用于自定义日志记录行为：

处理程序 描述
on_module_start / on_module_end Triggered when a dspy.Module subclass is invoked.
on_lm_start / on_lm_end Triggered when a dspy.LM subclass is invoked.
on_adapter_format_start / on_adapter_format_end Triggered when a dspy.Adapter subclass formats the input prompt.
on_adapter_parse_start / on_adapter_parse_end Triggered when a dspy.Adapter subclass postprocess the output text from an LM.
on_tool_start / on_tool_end Triggered when a dspy.Tool subclass is invoked.
on_evaluate_start / on_evaluate_end Triggered when a dspy.Evaluate instance is invoked.

以下是一个自定义回调的示例，用于记录ReAct智能体的中间步骤：

import dspy
from dspy.utils.callback import BaseCallback

# 1. Define a custom callback class that extends BaseCallback class
class AgentLoggingCallback(BaseCallback):

    # 2. Implement on_module_end handler to run a custom logging code.
    def on_module_end(self, call_id, outputs, exception):
        step = "Reasoning" if self._is_reasoning_output(outputs) else "Acting"
        print(f"== {step} Step ===")
        for k, v in outputs.items():
            print(f"  {k}: {v}")
        print("\n")

    def _is_reasoning_output(self, outputs):
        return any(k.startswith("Thought") for k in outputs.keys())

# 3. Set the callback to DSPy setting so it will be applied to program execution
dspy.configure(callbacks=[AgentLoggingCallback()])

== Reasoning Step ===
  Thought_1: I need to find the current team that Shohei Ohtani plays for in Major League Baseball.
  Action_1: Search[Shohei Ohtani current team 2023]

== Acting Step ===
  passages: ["Shohei Ohtani ..."]

...

信息

在回调函数中处理输入或输出数据时要谨慎。原地修改它们可能会改变传递给程序的原始数据，从而可能导致意外行为。为避免这种情况，强烈建议在执行任何可能改变数据的操作之前，先创建数据的副本。

优云智算