커스텀 LM 엔진

커스텀 LM 엔진 (Custom LM Engines)

자신만의 실행 엔진을 dspy.LM에 제공할 수 있어요. 그 전에 두 상황 중 어느 쪽에 해당하는지 확인해 보세요:

  • lm15이 이미 통신할 수 있는 HTTP 제공자 — OpenAI 호환 서비스, 회사 게이트웨이, DSPy에 번들로 포함된 lm15이 아직 나열하지 않는 호스트. 엔진을 작성하지 마세요: 제공자 선언을 하고 dspy.LM("<provider>/<model>")이 api_key, api_base, timeout, 저장 및 로드를 포함해 내장 제공자처럼 동작하도록 기본 라우팅되게 하세요.
  • 전혀 HTTP 제공자가 아님 — CLI, 프로세스 내 모델, 에이전트 하니스. 엔진을 작성하세요. 엔진이 전체 연결을 소유하므로 아래 커스텀 엔진이 소유하는 것 규칙이 적용됩니다.

최소 동기 엔진 인터페이스는 다음과 같아요:

class MyEngine:
    def complete(self, request: dspy.lm15.Request) -> dspy.lm15.Response:
        ...

DSPy는 프로그램의 입력을 포맷하고 lm15 요청으로 변환한 뒤, 엔진을 호출하고 답변을 Prediction으로 파싱해요. BaseLM을 서브클래싱하거나 OpenAI SDK 객체를 반환할 필요가 없습니다.

!!! warning "DSPy 3.5 컷오프" 커스텀 BaseLM.forward()/aforward() 통합, LegacyEngine/AsyncLegacyEngine, complete_legacy() 단축키는 3.4에서 deprecated되고 3.5에서 제거될 예정이에요. 여기 보이는 request/response 엔진 계약을 구현하세요. 레거시 래퍼는 마이그레이션 기한을 연장하지 않습니다. OpenAI 스타일 lm(messages=[...]) 호출도 제거되고 있어요. lm("hello")는 리스트 반환 편의 기능으로 남지만, 어댑터는 lm(Request(...))를 사용하고 Response를 직접 소비합니다. 마이그레이션 가이드를 참고하세요.

예상되는 백엔드 실패는 AuthError나 RateLimitError 같은 dspy.lm15의 특정 오류를 발생시켜야 해요. DSPy는 이를 공개 LMError 계열로 변환하고 재시도(retry)를 소유합니다. 예상치 못한 예외는 원래 원인을 유지하며, 메시지 텍스트만으로 재시도 가능하다고 추측하지 않아요. 오류 및 재시도 소유권을 참고하세요.

출처: 문서

본문

엔진 작성 대신 제공자 선언하기 (Declaring a provider instead of writing an engine)

lm15은 와이어(wire)에 대해 검증한 제공자들의 레지스트리를 통해 모델 문자열을 라우팅해요. 나열되지 않은 제공자는 레지스트리 항목을 구성하는 것과 동일한 세 가지 사실 — 접근 정책(이름, 키 변수, 주소), 와이어 방언(dialect), 서버의 철자(spelling)를 설명하는 호환 정책 — 을 사용해 프로세스에 대해 선언할 수 있습니다:

import dspy
from dspy.lm15 import AccessPolicy, EndpointSupport, ModelSupport, OpenAIChatCompat, ProviderDefinition

dspy.lm15.register_provider(
    ProviderDefinition.chat(
        AccessPolicy(
            provider="fireworks",
            supports=EndpointSupport(complete=True, stream=True, models=True),
            auth_modes=("bearer",),
            env_keys=("FIREWORKS_API_KEY",),
            base_url="https://api.fireworks.ai/inference/v1",
        ),
        compat=OpenAIChatCompat(max_tokens_field="max_tokens", thinking_format="reasoning_effort"),
        aliases=("fireworks-ai",),          # accept LiteLLM's spelling, fireworks_ai/, as a model prefix
    ),
    metadata_namespaces=("fireworks_ai",),  # this endpoint IS Fireworks: its snapshot entries may describe and price the models
)

lm = dspy.LM("fireworks/accounts/fireworks/models/deepseek-v4p1-flash")   # or fireworks_ai/...

이 작업은 사용하는 LM이 생성되기 전에 애플리케이션의 import 시점에 수행하세요. 각 dspy.LM은 생성 시 존재하는 등록을 바인딩하고 평생 유지합니다 — 선택, 기능, 가격 책정, 그리고 동기·비동기 엔진 모두가 그 하나의 바인딩을 읽어요. 따라서 register_provider(..., replace=True)는 그 뒤에 생성된 LM만 변경하며, 기존 LM을 이동시키지 않고, 한 LM의 동기 호출을 한 정의에, 비동기 호출을 다른 정의에 두는 일도 결코 없습니다.

그 후에는 제공자가 내장 제공자처럼 동작해요:

  • dspy.LM(..., api_key=..., api_base=..., timeout=...)이 존중됩니다. 키를 주지 않으면 FIREWORKS_API_KEY를 읽고, 키가 없으면 그 변수 이름을 밝혀요. 환경의 FIREWORKS_API_BASE 변수는 선언된 제공자를 리다이렉트하지 않습니다(레지스트리 제공자에게는 LiteLLM 시대 게이트웨이를 보존하기 위해 그렇게 합니다): 선언 자체가 주소를 명시하기 때문이에요.
  • engine="auto"는 각 요청을 선언된 호환성에 맞게 계획해요. LiteLLM 전용 설정(extra_headers, organization 등)만 설정한 클라이언트는 I/O 전에 LiteLLM을 선택합니다 — 선언된 주소에서, 선언된 자격 증명으로, LiteLLM의 일반 OpenAI 호환 문(openai/ 또는 해당 방언의 경우 anthropic/)을 통해. 자격 증명은 선언이 선택한 스킴 아래에서만 이동해요 — openai/ 문은 bearer 헤더를 보내고, anthropic/ 문은 x-api-key를 보냅니다 — 그리고 문이 보낼 수 없는 스킴의 선언은 빈 폴백을 받지 선언되지 않은 헤더에 비밀을 넣지 않아요. LM의 별칭은 제공자 프리픽스로 LiteLLM에 절대 전달되지 않는데, LiteLLM이 그 이름을 다른 서비스로 알고 있을 수 있기 때문이에요. 그 호출은 여전히 그 제공자의 것입니다: 오류는 그 제공자를 밝히고, 가격은 LiteLLM이 문의 모델 문자열을 읽는 방식이 아니라 metadata_namespaces에서 나옵니다. 선언된 호환성이 만드는 거부(예: reasoning 필드가 없는 와이어에 reasoning 다이얼)는 최종적입니다: 선언이 그 서버가 할 수 없는 일에 대한 권위이며, 어차피 LiteLLM을 통해 요청을 보내면 설정이 조용히 누락될 거예요. engine="lm15"는 두 경우 모두 거부합니다.
  • dump_state()/load_state()와 프로그램 save()/load()는 추가 작업이 필요 없어요 — 상태가 모델 문자열이기 때문입니다. 등록은 키 변수처럼 환경 사실이며, 프로그램이 로드될 때 자리 잡고 있어야 합니다.

별칭은 철자일 뿐, 그 이상이 아닙니다. 선언된 제공자의 엔드포인트가 지원하는 모델과 그 가격은 별개의 진술입니다:

  • metadata_namespaces=("fireworks_ai",)은 해당 LiteLLM 네임스페이스 아래의 DSPy 모델 메타데이터 스냅샷 항목이 이 제공자의 모델을 (기능을) 설명하고 가격을 매긴다고 말해요. 엔드포인트가 정말로 그 서비스일 때만 제공하세요. 그것 없이는 아무것도 상속되지 않고 비용은 알 수 없음으로 보고돼요.
  • supports=ModelSupport(function_calling=True, ...)은 제공자 전체에 대한 모델 지원을 명시하고, models={"private-v1": ModelSupport(...)}는 모델 ID별로 명시해요. 스냅샷 항목이 발견되면 더 구체적인 사실이므로 진술보다 우선합니다. 둘 다 명시하지 않은 것은 지원되지 않는 것으로 취급됩니다 — 내장 제공자들이 이미 겪는 규칙이죠 — 그리고 진술은 결코 가격을 매기지 않아요.
  • 호환성 객체가 여전히 모든 것을 경계 지어요: thinking_format="none"인 와이어는 진술이 무엇을 말하든 reasoning이 없습니다.

선언이 아닌 것: 영수증(receipt). lm15 자체의 레지스트리 항목은 라이브 캡처에서 고정되는 반면, 선언은 당신의 말이며 라우트가 그렇게 밝혀줘요(Resolution.declared). 동일한 등록을 두 번 등록하는 것은 no-op이고, 같은 ID 아래 다른 등록은 replace=True가 필요하며, lm15이나 LiteLLM이 이미 사용하는 철자는 거부됩니다.

커스텀 엔진이 소유하는 것 (What a custom engine owns)

커스텀 엔진은 dspy.LM이 빌려서 연결을 소유해요. 이로부터 세 가지 규칙이 나오고, DSPy는 추측하는 대신 각각을 강제합니다:

연결 설정은 거부됩니다. dspy.LM(engine=MyEngine(), api_key=...)은 ValueError를 발생시키고, api_base, base_url, timeout, headers, extra_headers 및 기타 클라이언트 설정도 마찬가지예요 — 생성 시, copy() 시, 매 호출 시(lm("hi", api_key=...)) 캐시 조회 전에. LM에서 엔진으로 그들을 위한 채널이 없고, 조용히 버리면 LM이 말한 키와 다르게 엔진의 키로 호출이 실행될 수 있기 때문입니다. 그들은 엔진의 생성자에 주세요.

엔진 쌍은 하나의 단위입니다. async_engine=는 커스텀 engine=와 함께만 허용되고, 각 측은 자기 종류여야 합니다: 동기 엔진의 complete는 Response를 반환하는 일반 함수이고, 비동기 엔진의 것은 코루틴 함수(async def complete)입니다. 비동기 측의 동기 메서드나 그 반대는 첫 호출이 아니라 생성 시 거부됩니다. lm.copy(engine=...)는 둘 다 교체합니다: 같은 호출에서 새 async_engine=를 전달하지 않으면 복사본에는 비동기 엔진이 없어요. engine을 언급하지 않는 복사본은 쌍을 유지합니다.

저장에는 엔진 자신의 상태가 필요합니다. 엔진은 dump_state() -> dict(JSON 직렬화 가능, 비밀 없음)와 클래스메서드 load_state(state) -> engine을 구현할 때 저장됩니다:

class MyEngine:
    def __init__(self, model="gpt-6-astra"):
        self.model = model

    def complete(self, request): ...

    def dump_state(self):
        return {"model": self.model}

    @classmethod
    def load_state(cls, state):
        return cls(**state)

그러면 lm.dump_state()는 {"engine": {"class": "your.module:MyEngine", "state": {...}}}를 기록합니다(async_engine도 마찬가지). 로드는 파일에서 해당 클래스를 import하는데, 이는 커스텀 LM 클래스와 동일한 신뢰 결정이므로 같은 방식으로 게이트됩니다: program.load(path, allow_unsafe_lm_state=True) 또는 dspy.LM.load_state(state, allow_custom_lm_class=True). 이 클래스는 상태를 로드하는 프로세스에서 그 경로로 import 가능해야 합니다. 함수 내부가 아니라 모듈 수준에서 정의하세요 — dump_state()는 다시 import할 수 없는 클래스를 거부해요 — 그리고 세션을 넘어 유지되어야 하는 상태는 import 가능한 모듈에 두세요: 스크립트나 노트북에서 정의된 클래스는 __main__:MyEngine으로 기록되며, __main__이 다시 정의하는 프로세스에서만 로드됩니다. dump_state()는 JSON이 담을 수 없는 엔진 상태도 거부해요. 두 메서드가 없는 엔진은 저장할 수 없으며, dump_state()가 그렇게 밝히고, 해결책은 그 LM 없이 프로그램을 저장하고 로드 후 다시 설정하는 것입니다. DSPy가 LM 상태에 api_key를 넣지 않는 것처럼 엔진 상태에서 비밀을 빼두세요.

이 튜토리얼은 Pi CLI를 커스텀 엔진으로 감쌉니다. Pi는 정상적인 시스템 프롬프트와 도구를 유지하므로, DSPy 프로그램이 저장소를 조사하도록 요청할 수 있어요. 도구 호출을 포함한 전체 에이전트 실행이 하나의 DSPy LM 응답이 됩니다.

사전 요구 사항 (Prerequisites)

  • 커스텀 engine= 인터페이스가 있는 DSPy 빌드.
  • Pi가 설치되어 PATH에서 pi로 사용 가능하고 인증되어 있어야 해요.
  • Pi 계정에서 사용 가능한 모델. 예제는 openai-codex와 gpt-6-astra를 사용하며, 필요하면 이 두 CLI 인자를 바꾸세요.
  • 조사할 로컬 Git 저장소.

!!! warning "Pi는 파일시스템과 셸 접근 권한이 있습니다" 이 예제는 Pi의 정상적인 도구, 설정, 발견된 리소스를 유지해요. 이들은 명령을 실행하고 파일을 수정할 수 있어요. 아무것도 수정하지 말라고 요청하는 것은 지침일 뿐 샌드박스가 아닙니다. 신뢰할 수 있는 저장소에 대해서만 실행하고, 격리가 중요하다면 OS 수준 샌드박스를 사용하세요.

엔진 정의 (Define the engine)

Pi의 print 모드는 도구 루프를 실행하고 최종 어시스턴트 텍스트를 stdout에 써요. 해당 텍스트를 lm15 응답으로 감쌉니다. 이 데모에는 이벤트 스트림 파싱이 필요 없어요.

DSPy의 시스템 지침을 --append-system-prompt로 추가해요. 이는 Pi의 정상 시스템 프롬프트를 보존하면서 답변을 DSPy용으로 포맷하는 방법을 알려줘요.

import subprocess
import dspy
from dspy.lm15 import Message, Response, Usage


class PiEngine:
    def complete(self, request):
        result = subprocess.run(
            ["pi", "--print", "--no-session",
             "--provider", "openai-codex", "--model", "gpt-6-astra",
             "--append-system-prompt", request.system or ""],
            input=request.messages[-1].text,
            text=True, capture_output=True, check=True, timeout=120,
        )
        return Response(
            id=None, model=request.model,
            message=Message.assistant(result.stdout.strip()),
            finish_reason="stop", usage=Usage(),
        )

Usage()는 사용량이 DSPy에 보고되지 않았다는 뜻이며, 실행이 0 토큰을 소비했다는 뜻이 아니에요. 이 최소 래퍼는 Pi의 도구 루프 전체의 토큰 사용량이나 비용을 집계하지 않습니다.

DSPy 프로그램 실행 (Run a DSPy program)

작업 디렉터리가 저장소 루트인 노트북에서 이 셀들을 실행하세요. 마지막 표현식은 print() 없이 예측을 표시합니다:

dspy.configure(
    lm=dspy.LM("pi", engine=PiEngine(), cache=False, num_retries=0),
    adapter=dspy.ChatAdapter(use_json_adapter_fallback=False),
)

program = dspy.Predict("question -> answer")
program(
    question="Find the largest tracked file by byte size in this repository. "
             "Use a tool to check. Report its path and size; do not modify anything."
)

여기서 "pi"는 히스토리용 DSPy 모델 레이블이에요. 엔진의 CLI 인자가 실제 제공자와 모델을 선택합니다. PiEngine은 상태를 보유하지 않으므로, 저장 가능하게 만들려면 {}를 반환하는 dump_state와 cls()를 반환하는 load_state를 추가하세요.

DSPy 저장소에 대한 답변은 다음과 같을 수 있어요:

Prediction(
    answer='Largest tracked file: docs/docs/tutorials/observability/mlflow_trace_ui_navigation.gif\nSize: 8,704,325 bytes.'
)

답변은 저장소에 따라 달라져요. Pi는 여전히 도구를 사용하지만 print 모드는 도구 이벤트를 노출하지 않으므로 이는 증분 스트리밍이 아닙니다.

데모 경계 (Demo boundaries)

  • 최소 입력 처리: 마지막 사용자 텍스트 메시지와 평문 시스템 프롬프트만 전달해요. 이전 메시지, 생성 옵션, DSPy 선언 도구는 전달되거나 검증되지 않아요. 이 데모를 데모, 대화 히스토리, 미디어 없이 사용하세요. Pi 자신의 도구는 계속 활성화됩니다.
  • 동기 전용: 비동기 대응물이나 스트리밍 구현이 없어요.
  • 저장된 Pi 세션 없음: --no-session은 트랜스크립트 지속을 피하지만, Pi는 자체 설정이나 자격 증명을 업데이트할 수 있어요.
  • 자동 DSPy 재생 없음: 캐싱, LM 재시도, 어댑터 폴백은 데모된 프로그램에 대해 비활성화됩니다. Pi 자신의 재시도와 압축 설정은 여전히 적용됩니다.
  • 상세 원격 측정 없음: 사용량, 비용, 도구 이벤트는 print 모드에서 노출되지 않아요. 이 데모는 성공적인 프로세스 종료를 stop으로 표시하며 토큰 한도 종료를 구분할 수 없어요. 권위 있는 종료 이유와 회계를 원하면 각 어시스턴트 턴에 대해 --mode json을 사용하세요.
  • 정상 Pi 출력: 확장이 stdout에 영향을 줄 수 있어요. 통제된 프로덕션 프로토콜을 위해서는 JSON/RPC 모드를 사용하고 신뢰할 수 있는 확장을 명시적으로 선택하세요.
  • 버퍼링된 하위 프로세스 출력: stdout과 stderr는 메모리에 수집됩니다. 타임아웃은 직접적인 Pi 프로세스에만 적용되며, 이 데모는 전체 하위 프로세스 트리를 관리하지 않아요.
  • 인자 안의 시스템 지침: 추가된 프롬프트는 로컬 프로세스 검사에 보이며 OS 인자 길이 제한의 적용을 받아요. 이 최소 예제에서는 민감한 시스템 지침을 피하세요.

이들은 커스텀 엔진 인터페이스를 시연하기 위한 의도적인 단축이며, 프로덕션 하위 프로세스 백엔드가 아닙니다. 더 충실한 통합을 위해서는 완전한 메시지와 제공자 메타데이터를 보존하고, 취소와 바운드 출력 처리를 구현하며, 각 생성 옵션을 명시적으로 매핑하거나 거부하세요.

더 알아보기 (Learn more)