튜토리얼: DSPy 프로그램 배포하기

튜토리얼: DSPy 프로그램 배포하기 (Deploying your DSPy program)

이 가이드에서는 DSPy 프로그램을 프로덕션에 배포하는 두 가지 방법을 알아볼게요. 가벼운 배포에는 FastAPI를, 프로그램 버전 관리와 운영 관리가 필요한 프로덕션급 배포에는 MLflow를 사용해요.

아래에서는 배포하려는 다음과 같은 간단한 DSPy 프로그램이 있다고 가정할게요. 더 정교한 프로그램으로 바꿔도 됩니다.

import dspy

dspy.configure(lm=dspy.LM("openai/gpt-4o-mini"))
dspy_program = dspy.ChainOfThought("question -> answer")

출처: 문서

본문

FastAPI로 배포하기 (Deploying with FastAPI)

FastAPI는 DSPy 프로그램을 REST API로 서빙하는 간단한 방법을 제공해요. 프로그램 코드에 직접 접근할 수 있고 가벼운 배포 솔루션이 필요할 때 이상적입니다.

> pip install fastapi uvicorn
> export OPENAI_API_KEY="your-openai-api-key"

위에서 정의한 dspy_program을 서빙하는 FastAPI 애플리케이션을 만들어볼게요.

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

import dspy

app = FastAPI(
    title="DSPy Program API",
    description="A simple API serving a DSPy Chain of Thought program",
    version="1.0.0"
)

# Define request model for better documentation and validation
class Question(BaseModel):
    text: str

# Configure your language model and 'asyncify' your DSPy program.
lm = dspy.LM("openai/gpt-4o-mini")
dspy.configure(lm=lm, async_max_workers=4) # default is 8
dspy_program = dspy.ChainOfThought("question -> answer")
dspy_program = dspy.asyncify(dspy_program)

@app.post("/predict")
async def predict(question: Question):
    try:
        result = await dspy_program(question=question.text)
        return {
            "status": "success",
            "data": result.toDict()
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

위 코드에서 dspy.asyncify를 호출해 DSPy 프로그램을 높은 처리량의 FastAPI 배포를 위한 async 모드로 변환해요. 현재는 DSPy 프로그램을 별도 스레드에서 실행하고 그 결과를 await합니다.

기본적으로 생성되는 스레드 수의 제한은 8이에요. 이를 워커 풀(worker pool)이라고 생각하면 됩니다. 실행 중인 프로그램이 8개 있는 상태에서 한 번 더 호출하면, 9번째 호출은 8개 중 하나가 반환될 때까지 기다려요. 새 async_max_workers 설정으로 async 용량을 구성할 수 있어요.

??? "DSPy 2.6.0+ 에서의 스트리밍 (Streaming, in DSPy 2.6.0+)"

스트리밍도 DSPy 2.6.0+에서 지원되며, `pip install -U dspy`로 설치할 수 있어요.

`dspy.streamify`를 사용해 DSPy 프로그램을 스트리밍 모드로 변환할 수 있어요. 최종 예측이 준비되기 전에 중간 출력(예: O1 스타일 추론)을 클라이언트로 스트리밍하고 싶을 때 유용해요. 내부적으로 asyncify를 사용하며 실행 시맨틱을 상속합니다.

```python
dspy_program = dspy.asyncify(dspy.ChainOfThought("question -> answer"))
streaming_dspy_program = dspy.streamify(dspy_program)

@app.post("/predict/stream")
async def stream(question: Question):
    async def generate():
        async for value in streaming_dspy_program(question=question.text):
            if isinstance(value, dspy.Prediction):
                data = {"prediction": value.labels().toDict()}
            elif isinstance(value, litellm.ModelResponse):
                data = {"chunk": value.json()}
            yield f"data: {orjson.dumps(data).decode()}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

# Since you're often going to want to stream the result of a DSPy program as server-sent events,
# we've included a helper function for that, which is equivalent to the code above.

from dspy.utils.streaming import streaming_response

@app.post("/predict/stream")
async def stream(question: Question):
    stream = streaming_dspy_program(question=question.text)
    return StreamingResponse(streaming_response(stream), media_type="text/event-stream")
```

코드를 파일에 저장하세요(예: fastapi_dspy.py). 그런 다음 앱을 다음과 같이 서빙할 수 있어요:

> uvicorn fastapi_dspy:app --reload

그러면 http://127.0.0.1:8000/에 로컬 서버가 시작돼요. 아래 파이썬 코드로 테스트할 수 있어요:

import requests

response = requests.post(
    "http://127.0.0.1:8000/predict",
    json={"text": "What is the capital of France?"}
)
print(response.json())

다음과 같은 응답을 볼 수 있어요:

{
  "status": "success",
  "data": {
    "reasoning": "The capital of France is a well-known fact, commonly taught in geography classes and referenced in various contexts. Paris is recognized globally as the capital city, serving as the political, cultural, and economic center of the country.",
    "answer": "The capital of France is Paris."
  }
}

MLflow로 배포하기 (Deploying with MLflow)

DSPy 프로그램을 패키징해 격리된 환경에서 배포하려면 MLflow로 배포하는 것을 권장해요. MLflow는 버전 관리, 추적, 배포를 포함한 머신러닝 워크플로우 관리를 위한 인기 있는 플랫폼입니다.

> pip install mlflow>=2.18.0

DSPy 프로그램을 저장할 MLflow 추적 서버를 띄워볼게요. 아래 명령은 http://127.0.0.1:5000/에 로컬 서버를 시작합니다.

> mlflow ui

그런 다음 DSPy 프로그램을 정의하고 MLflow 서버에 로그할 수 있어요. MLflow에서 "log"는 과부하된(overloaded) 용어인데, 기본적으로 프로그램 정보를 환경 요구사항과 함께 MLflow 서버에 저장한다는 뜻이에요. 이는 mlflow.dspy.log_model() 함수로 수행됩니다:

!!! note

MLflow 2.22.0 기준으로, MLflow로 배포할 때는 DSPy 프로그램을 커스텀 DSPy Module 클래스로 감싸야 한다는 주의사항이 있어요. 이는 MLflow가 위치 인자(positional arguments)를 요구하는 반면, DSPy 사전 빌드 모듈(예: `dspy.Predict` 또는 `dspy.ChainOfThought`)은 위치 인자를 허용하지 않기 때문입니다. 이를 해결하려면 `dspy.Module`을 상속하는 래퍼 클래스를 만들고 아래 예제처럼 `forward()` 메서드에 프로그램 로직을 구현하세요.

import dspy
import mlflow

mlflow.set_tracking_uri("http://127.0.0.1:5000/")
mlflow.set_experiment("deploy_dspy_program")

lm = dspy.LM("openai/gpt-4o-mini")
dspy.configure(lm=lm)

class MyProgram(dspy.Module):
    def __init__(self):
        super().__init__()
        self.cot = dspy.ChainOfThought("question -> answer")

    def forward(self, messages):
        return self.cot(question=messages[0]["content"])

dspy_program = MyProgram()

with mlflow.start_run():
    mlflow.dspy.log_model(
        dspy_program,
        "dspy_program",
        input_example={"messages": [{"role": "user", "content": "What is LLM agent?"}]},
        task="llm/v1/chat",
    )

배포된 프로그램이 OpenAI 채팅 API와 같은 형식으로 자동으로 입력을 받고 출력을 생성하도록 task="llm/v1/chat"을 설정하는 것을 권장해요. 이는 LM 애플리케이션의 공통 인터페이스입니다. 위 코드를 파일(예: mlflow_dspy.py)에 저장하고 실행하세요.

프로그램을 로그한 후 MLflow UI에서 저장된 정보를 볼 수 있어요. http://127.0.0.1:5000/을 열고 deploy_dspy_program 실험을 선택한 뒤 방금 만든 run을 선택하면, Artifacts 탭 아래에 로그된 프로그램 정보가 아래 스크린샷과 비슷하게 표시되는 걸 볼 수 있어요:

MLflow UI

UI에서 run id를 가져와서(또는 mlflow_dspy.py 실행 시 콘솔 출력에서) 로그된 프로그램을 다음 명령으로 배포할 수 있어요:

> mlflow models serve -m runs:/{run_id}/model -p 6000

프로그램이 배포된 후 다음 명령으로 테스트할 수 있어요:

> curl http://127.0.0.1:6000/invocations -H "Content-Type:application/json"  --data '{"messages": [{"content": "what is 2 + 2?", "role": "user"}]}'

다음과 같은 응답을 볼 수 있어요:

{
  "choices": [
    {
      "index": 0,
      "message": {
        "role": "assistant",
        "content": "{\"reasoning\": \"The question asks for the sum of 2 and 2. To find the answer, we simply add the two numbers together: 2 + 2 = 4.\", \"answer\": \"4\"}"
      },
      "finish_reason": "stop"
    }
  ]
}

MLflow로 DSPy 프로그램을 배포하는 방법과 배포 커스터마이징 방법의 완전한 가이드는 MLflow 문서를 참고하세요.

MLflow 배포를 위한 모범 사례 (Best Practices for MLflow Deployment)

  1. 환경 관리 (Environment Management): Python 의존성을 항상 conda.yaml 또는 requirements.txt 파일에 명시하세요.
  2. 버전 관리 (Versioning): 모델 버전에 의미 있는 태그와 설명을 사용하세요.
  3. 입력 검증 (Input Validation): 명확한 입력 스키마와 예제를 정의하세요.
  4. 모니터링 (Monitoring): 프로덕션 배포를 위한 적절한 로깅과 모니터링을 설정하세요.

프로덕션 배포를 위해서는 MLflow를 컨테이너화와 함께 사용하는 것을 고려해 보세요:

> mlflow models build-docker -m "runs:/{run_id}/model" -n "dspy-program"
> docker run -p 6000:8080 dspy-program

프로덕션 배포 옵션과 모범 사례의 완전한 가이드는 MLflow 문서를 참고하세요.

더 알아보기 (Learn more)