튜토리얼: DSPy에서의 디버깅과 관찰성

튜토리얼: DSPy에서의 디버깅과 관찰성 (Debugging and Observability in DSPy)

이 가이드에서는 DSPy에서 문제를 디버깅하고 관찰성(observability)을 개선하는 방법을 알아볼게요. 현대 AI 프로그램은 언어 모델, 리트리버, 도구 같은 여러 구성 요소를 포함하는 경우가 많아요. DSPy는 이러한 복잡한 AI 시스템을 깔끔하고 모듈식으로 구축하고 최적화할 수 있게 해줍니다.

하지만 시스템이 더 정교해질수록 시스템이 무엇을 하고 있는지 이해하는 능력이 중요해져요. 투명성이 없으면 예측 과정은 쉽게 블랙박스가 되어, 실패나 품질 문제를 진단하기 어렵고 프로덕션 유지보수도 까다로워집니다.

이 튜토리얼이 끝나면 MLflow Tracing을 사용해 문제를 디버깅하고 관찰성을 개선하는 방법을 이해하게 될 거예요. 또한 콜백(callback)을 사용해 커스텀 로깅 솔루션을 구축하는 방법도 살펴봅니다.

출처: 문서

본문

프로그램 정의 (Define a Program)

먼저 ColBERTv2의 Wikipedia 데이터셋을 검색 소스로 사용하는 간단한 ReAct 에이전트를 만들어볼게요. 더 정교한 프로그램으로 바꿔도 됩니다.

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.

이건 틀린 답이에요. 그는 더 이상 Hokkaido Nippon-Ham Fighters에서 뛰지 않고, Dodgers로 이적해 2024년 월드시리즈에서 우승했죠! 프로그램을 디버깅하고 잠재적 수정 방법을 살펴볼게요.

inspect_history 사용하기 (Using inspect_history)

DSPy는 지금까지 이루어진 모든 LLM 호출을 출력하는 inspect_history() 유틸리티를 제공해요:

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

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

System message:

Your input fields are:
1. `question` (str)

...

Response:

Response:

[[ ## reasoning ## ]]
The search for information regarding Shohei Ohtani's team in June 2025 did not yield any specific results. The retrieved data consistently mentioned that he plays for the Hokkaido Nippon-Ham Fighters, but there was no indication of any changes or updates regarding his team for the specified date. Given the lack of information, it is reasonable to conclude that he may still be with the Hokkaido Nippon-Ham Fighters unless there are future developments that are not captured in the current data.

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

[[ ## completed ## ]]

로그를 보면 에이전트가 검색 도구에서 유용한 정보를 가져오지 못했다는 걸 알 수 있어요. 하지만 리트리버가 정확히 무엇을 반환했을까요? 유용하지만, inspect_history에는 몇 가지 한계가 있어요:

• 실제 시스템에서는 리트리버, 도구, 커스텀 모듈 같은 다른 구성 요소가 중요한 역할을 하는데, inspect_history는 LLM 호출만 로깅해요.
• DSPy 프로그램은 단일 예측 내에서 여러 번 LLM 호출을 하는 경우가 많아요. 모놀리식 로그 히스토리는 여러 질문을 처리할 때 로그를 정리하기 어렵게 만듭니다.
• 매개변수, 지연 시간, 모듈 간의 관계 같은 메타데이터는 캡처되지 않아요.

Tracing은 이러한 한계를 해결하고 더 종합적인 솔루션을 제공해요.

트레이싱 (Tracing)

MLflow는 DSPy와 매끄럽게 통합되어 LLMOps 모범 사례를 지원하는 엔드투엔드 머신러닝 플랫폼이에요. DSPy와 함께 MLflow의 자동 트레이싱 기능을 사용하는 건 간단합니다. 서비스 가입이나 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) 를 생성하고 실험 내에 기록해요. 이 트레이스들을 시각적으로 탐색하려면 브라우저에서 http://127.0.0.1:5000/을 열고 방금 만든 실험을 선택한 뒤 Traces 탭으로 이동하세요:

가장 최근 트레이스를 클릭하면 상세 분석을 볼 수 있어요:

여기서 워크플로우의 각 단계 입력과 출력을 검토할 수 있어요. 예를 들어, 위 스크린샷은 retrieve 함수의 입력과 출력을 보여줘요. 리트리버의 출력을 검사하면 오래된 정보를 반환했고, 이는 2025년 6월에 쇼헤이 오타니가 어느 팀에서 뛰는지 결정하기에 부족하다는 걸 알 수 있어요. 언어 모델의 입력, 출력, 구성 같은 다른 단계도 검사할 수 있어요.

오래된 정보 문제를 해결하려면 retrieve 함수를 Tavily search로 구동되는 웹 검색 도구로 바꿀 수 있어요.

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

아래는 MLflow UI를 탐색하는 방법을 보여주는 GIF예요:

MLflow 트레이싱 사용에 대한 완전한 가이드는 MLflow Tracing Guide를 참고하세요.

!!! info "MLflow에 대해 더 알아보기"

MLflow는 실험 추적, 평가, 배포 같은 광범위한 기능을 제공하는 엔드투엔드 LLMOps 플랫폼이에요. DSPy와 MLflow 통합에 대해 더 알아보려면 [이 튜토리얼](../deployment/index.md#deploying-with-mlflow)을 방문하세요.

커스텀 로깅 솔루션 구축하기 (Building a Custom Logging Solution)

때로는 커스텀 로깅 솔루션을 구현하고 싶을 수 있어요. 예를 들어, 특정 모듈이 트리거한 특정 이벤트를 로깅해야 할 수도 있죠. DSPy의 콜백(callback) 메커니즘이 이런 사용 사례를 지원해요. BaseCallback 클래스는 로깅 동작을 커스터마이징하기 위한 여러 핸들러를 제공합니다:

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 ..."]

...

!!! info "콜백에서 입력과 출력 다루기"

콜백에서 입력 또는 출력 데이터를 다룰 때는 주의해야 해요. 이를 제자리(in-place)에서 변경하면 프로그램에 전달된 원본 데이터가 수정되어 예기치 않은 동작이 발생할 수 있어요. 이를 피하려면 데이터를 변경할 수 있는 작업을 수행하기 전에 데이터의 복사본을 만드는 것을 강력히 권장합니다.

더 알아보기 (Learn more)

• Optimizer Tracking 튜토리얼
• Deployment 튜토리얼
• 빌드 AI 프로그램 튜토리얼