본문으로 건너뛰기

스트리밍

Agent와 기본 LLM은 부분 문자열(토큰)을 하나씩 순차적으로 생성하고, 생성된 각 토큰이 누적되어 최종 출력을 형성합니다. 생성 과정이 길수록 결과를 완성하는 데 더 오래 걸리므로, 이러한 중간 결과를 활용하는 것이 종종 유용합니다. 이것들은 전체 결과의 증분(델타)으로 스트리밍될 수 있습니다.

이는 사용자가 부분 결과를 검토한 후 콘텐츠 생성을 중지하고 싶을 때나, 에이전트 기반 애플리케이션 개발자가 애플리케이션의 반응성을 높이기 위해 부분 결과를 실시간으로 사용자에게 스트리밍하고 싶을 때 유용합니다.

델타 스트리밍 방법​

지금까지 사용해 온 Agent의 run() 메서드는 완전한 메시지를 순차적으로 반환하는 반면, run_delta()(Python, Rust) 또는 runDelta()(Nodejs, Web) 메서드는 각 LLM 생성 단계에 대한 부분 결과의 시퀀스를 반환하며, 이를 메시지의 델타 라고 합니다.

import asyncio

import ailoy as ai


async def main():
    lm = await ai.LangModel.new_local("Qwen/Qwen3-0.6B")
    agent = ai.Agent(lm)

    async for resp in agent.run_delta("Please give me a short poem about AI."):
        if resp.delta.contents and isinstance(resp.delta.contents[0], ai.PartDelta.Text):
            # 줄바꿈 없이 텍스트 델타 출력
            print(resp.delta.contents[0].text, end="")
    print()


if __name__ == "__main__":
    asyncio.run(main())

델타를 완성된 메시지로 변환​

메시지 델타를 순차적으로 누적하여 완전한 메시지를 구성할 수도 있습니다.

메시지가 완전히 생성되면 완료 이유(finish reason)가 제공됩니다. 완료 이유가 나타날 때까지 메시지 델타를 누적한 다음 finish()를 호출하여 완전한 메시지를 생성합니다.

import asyncio

import ailoy as ai


async def main():
    lm = await ai.LangModel.new_local("Qwen/Qwen3-0.6B")
    agent = ai.Agent(lm)

    GREEN = "\x1b[32m"
    RESET = "\x1b[0m"

    acc = ai.MessageDelta()  # 누적의 기반
    async for resp in agent.run_delta("Please give me a short poem about AI."):
        if resp.delta.contents and isinstance(resp.delta.contents[0], ai.PartDelta.Text):
            # 텍스트 델타를 녹색으로 출력
            print(GREEN + resp.delta.contents[0].text + RESET, end="")
        acc += resp.delta  # 새로 생성된 델타를 기반에 누적

        # finish_reason이 존재하면 전체 메시지가 생성된 것입니다.
        if resp.finish_reason is not None:
            message = acc.to_message()
            if isinstance(message.contents[0], ai.Part.Text):
                print("\n\n" + message.contents[0].text)
            acc = ai.MessageDelta()  # 기반을 다시 초기화
    print()


if __name__ == "__main__":
    asyncio.run(main())