教程：DSPy 中的调试和可观测性
本指南演示了如何在 DSPy 中调试问题和提高可观测性。现代 AI 程序通常涉及多个组件，例如语言模型、检索器和工具。DSPy 允许您以清晰模块化的方式构建和优化这种复杂的 AI 系统。
然而，随着系统变得越来越复杂，理解您的系统正在做什么的能力变得至关重要。如果没有透明度，预测过程很容易变成一个黑箱，这使得诊断故障或质量问题变得困难，并且生产维护具有挑战性。
在本教程结束时，您将了解如何使用MLflow 跟踪调试问题和提高可观测性。您还将探索如何使用回调构建自定义日志解决方案。
定义程序
我们将首先创建一个简单的 ReAct 智能体，它使用 ColBERTv2 的 Wikipedia 数据集作为检索源。您可以将其替换为更复杂的程序。
import dspy
from dspy.datasets import HotPotQA
lm = dspy.LM('openai/gpt-4o-mini')
colbert = dspy.ColBERTv2(url='http://20.102.90.50:2017/wiki17_abstracts')
dspy.configure(lm=lm, rm=colbert)
agent = dspy.ReAct("question -> answer", tools=[dspy.Retrieve(k=1)])
现在，让我们问智能体一个简单的问题
prediction = agent(question="Which baseball team does Shohei Ohtani play for?")
print(prediction.answer)
哦，这是不正确的。他不再为天使队效力了；他转到了道奇队并在 2024 年赢得了世界大赛冠军！让我们调试程序并探索潜在的修复方案。
使用 inspect_history
DSPy 提供了 inspect_history() 工具，它会打印出目前为止所有已调用的 LLM
[2024-12-01T10:23:29.144257]
System message:
Your input fields are:
1. `question` (str)
...
Response:
[[ ## Thought_5 ## ]]
The search results continue to be unhelpful and do not provide the current team for Shohei Ohtani in Major League Baseball. I need to conclude that he plays for the Los Angeles Angels based on prior knowledge, as the searches have not yielded updated information.
[[ ## Action_5 ## ]]
Finish[Los Angeles Angels]
[[ ## completed ## ]]
inspect_history 有用，但它有一些限制
- 在实际系统中，检索器、工具和自定义模块等其他组件发挥着重要作用，但
inspect_history只记录 LLM 调用。 - DSPy 程序通常在一次预测中进行多次 LLM 调用。单一的日志历史很难组织日志，尤其是在处理多个问题时。
- 参数、延迟以及模块之间的关系等元数据未被捕获。
跟踪解决了这些限制并提供了更全面的解决方案。
跟踪
MLflow 是一个端到端的机器学习平台，与 DSPy 无缝集成以支持 LLMOps 的最佳实践。在 DSPy 中使用 MLflow 的自动跟踪功能非常简单；无需注册服务或 API 密钥。您只需安装 MLflow 并在您的 Notebook 或脚本中调用 mlflow.dspy.autolog()。
import mlflow
mlflow.dspy.autolog()
# This is optional. Create an MLflow Experiment to store and organize your traces.
mlflow.set_experiment("DSPy")
现在一切就绪！让我们再次运行您的智能体
MLflow 会自动为预测生成一个跟踪并将其记录到实验中。要可视化地探索跟踪，请使用以下命令启动 MLflow UI 并在浏览器中访问它
从检索器步骤的输出中，您可以观察到它返回了过时信息；表明大谷翔平仍在日本联赛打球，最终答案是基于 LLM 的先验知识！我们应该更新数据集或添加其他工具以确保访问最新信息。
信息
MLflow 是一个端到端的 LLMOps 平台，提供了实验跟踪、评估和部署等广泛功能。要了解更多关于 DSPy 和 MLflow 集成的知识，请访问本教程。
例如，我们可以使用 Tavily 网络搜索 API 为智能体添加网络搜索功能。
from dspy.predict.react import Tool
from tavily import TavilyClient
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=[Tool(web_search)])
prediction = agent(question="Which baseball team does Shohei Ohtani play for?")
print(agent.answer)
构建自定义日志解决方案
有时，您可能需要实现一个自定义日志解决方案。例如，您可能需要记录由特定模块触发的特定事件。DSPy 的回调机制支持此类用例。BaseCallback 类提供了几种处理程序用于自定义日志行为
| 处理程序 | 描述 |
|---|---|
on_module_start / on_module_end |
当调用 dspy.Module 的子类时触发。 |
on_lm_start / on_lm_end |
当调用 dspy.LM 的子类时触发。 |
on_adapter_format_start / on_adapter_format_end |
当 dspy.Adapter 的子类格式化输入提示时触发。 |
on_adapter_parse_start / on_adapter_parse_end |
当 dspy.Adapter 的子类后处理来自 LLM 的输出文本时触发。 |
这是一个自定义回调示例，它记录 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 ..."]
...
信息
使用回调处理输入或输出数据时要谨慎。原地修改它们可能会改变传递给程序的原始数据，从而可能导致意外行为。为避免这种情况，强烈建议在执行任何可能修改数据的操作之前创建一个数据副本。