오프라인 엔진 API

오프라인 엔진 API (Offline Engine API)

HTTP 서버를 띄우는 게 오히려 불필요한 복잡함이나 오버헤드가 되는 상황이 있어요. 그럴 때 SGLang은 서버 없이 바로 쓰는 직접 추론 엔진을 제공합니다. 두 가지 대표적인 쓰임이 있어요.

  • 오프라인 배치 추론 (Offline Batch Inference)
  • 엔진 위에 올려 만드는 커스텀 서버 (Custom Server on Top of the Engine)

이 문서는 오프라인 배치 추론을 중심으로 네 가지 추론 모드를 다룹니다.

  • 논스트리밍 동기 생성
  • 스트리밍 동기 생성
  • 논스트리밍 비동기 생성
  • 스트리밍 비동기 생성

추가로, 커스텀 서버를 SGLang 오프라인 엔진 위에 쉽게 올릴 수 있어요. 파이썬 스크립트로 동작하는 자세한 예시는 custom_server에서 볼 수 있습니다.

출처: 공식 문서 - Offline Engine API

Nest Asyncio

오프라인 엔진을 ipython이나 다른 중첩 루프 코드에서 쓰려면 아래 코드를 추가해야 합니다.

import nest_asyncio

nest_asyncio.apply()

고급 사용 (Advanced Usage)

엔진은 vlm inference와 히든 스테이트 추출을 지원해요.

더 많은 사용 사례는 the examples를 참고하면 됩니다.

오프라인 배치 추론 (Offline Batch Inference)

SGLang 오프라인 엔진은 효율적인 스케줄링으로 배치 추론을 지원합니다.

# launch the offline engine
import asyncio

import sglang as sgl
import sglang.test.doc_patch
from sglang.utils import async_stream_and_merge, stream_and_merge

llm = sgl.Engine(model_path="qwen/qwen2.5-0.5b-instruct")

논스트리밍 동기 생성 (Non-streaming Synchronous Generation)

prompts = [
    "Hello, my name is",
    "The president of the United States is",
    "The capital of France is",
    "The future of AI is",
]

sampling_params = {"temperature": 0.8, "top_p": 0.95}

outputs = llm.generate(prompts, sampling_params)
for prompt, output in zip(prompts, outputs):
    print("===============================")
    print(f"Prompt: {prompt}\nGenerated text: {output['text']}")

스트리밍 동기 생성 (Streaming Synchronous Generation)

동기 스트리밍에서는 stream_and_merge로 토큰 조각의 오버랩을 제거해 이어 붙인 결과를 받습니다.

prompts = [
    "Write a short, neutral self-introduction for a fictional character. Hello, my name is",
    "Provide a concise factual statement about France's capital city. The capital of France is",
    "Explain possible future trends in artificial intelligence. The future of AI is",
]

sampling_params = {
    "temperature": 0.2,
    "top_p": 0.9,
}

print("\n=== Testing synchronous streaming generation with overlap removal ===\n")

for prompt in prompts:
    print(f"Prompt: {prompt}")
    merged_output = stream_and_merge(llm, prompt, sampling_params)
    print("Generated text:", merged_output)
    print()

논스트리밍 비동기 생성 (Non-streaming Asynchronous Generation)

배치를 비동기로 돌리려면 async_generate를 기다리면 됩니다.

prompts = [
    "Write a short, neutral self-introduction for a fictional character. Hello, my name is",
    "Provide a concise factual statement about France's capital city. The capital of France is",
    "Explain possible future trends in artificial intelligence. The future of AI is",
]

sampling_params = {"temperature": 0.8, "top_p": 0.95}

print("\n=== Testing asynchronous batch generation ===")

async def main():
    outputs = await llm.async_generate(prompts, sampling_params)

    for prompt, output in zip(prompts, outputs):
        print(f"\nPrompt: {prompt}")
        print(f"Generated text: {output['text']}")

asyncio.run(main())

스트리밍 비동기 생성 (Streaming Asynchronous Generation)

비동기 스트리밍에서는 async_stream_and_merge를 async for로 순회하며 오버랩을 제거한 청크를 받습니다.

prompts = [
    "Write a short, neutral self-introduction for a fictional character. Hello, my name is",
    "Provide a concise factual statement about France's capital city. The capital of France is",
    "Explain possible future trends in artificial intelligence. The future of AI is",
]

sampling_params = {"temperature": 0.8, "top_p": 0.95}

print("\n=== Testing asynchronous streaming generation (no repeats) ===")

async def main():
    for prompt in prompts:
        print(f"\nPrompt: {prompt}")
        print("Generated text: ", end="", flush=True)

        # Replace direct calls to async_generate with our custom overlap-aware version
        async for cleaned_chunk in async_stream_and_merge(llm, prompt, sampling_params):
            print(cleaned_chunk, end="", flush=True)

        print()  # New line after each prompt

asyncio.run(main())

llm.shutdown()

더 알아보기 (Learn more)