에이전트 응답 스트리밍하기

에이전트 응답 스트리밍하기

스트리밍 응답이란 무엇인가

스트리밍 응답(streamed response) 은 메시지 내용을 작고 점진적인 조각(chunk)으로 나눠 전달해요. 전체 응답이 로딩될 때까지 기다리는 대신, 메시지가 펼쳐지는 동안 사용자가 바로 보고 상호작용할 수 있게 하는 방식이라 사용자 경험이 좋아져요. 사용자는 즉시 정보를 처리하기 시작할 수 있어 반응성과 상호작용감이 높아지고, 지연을 줄이며 대화 과정 전반에 걸쳐 관심을 유지하게 도와줍니다.

Semantic Kernel에서의 스트리밍

스트리밍을 지원하는 Semantic Kernel의 AI 서비스는, 완전히 형성된 메시지에 쓰이는 콘텐츠 타입과 다른 콘텐츠 타입을 사용해요. 이 타입들은 스트리밍 데이터의 점진적 특성을 다루도록 설계되어 있고, Agent Framework에서도 같은 타입을 사용해 두 시스템 사이에서 일관성과 효율을 유지합니다. 대표적인 타입으론 StreamingChatMessageContent, StreamingTextContent, StreamingFileReferenceContent, StreamingAnnotationContent 가 있어요.

ChatCompletionAgent 에서 스트리밍 응답

ChatCompletionAgent 에서 스트리밍 응답을 호출할 때, AgentThread 안의 ChatHistory 는 전체 응답을 받은 뒤에 갱신돼요. 응답은 점진적으로 스트리밍되지만, 기록에는 완성된 메시지만 남아요. 그 덕분에 ChatHistory 는 항상 완전히 형성된 응답을 반영해서 일관성을 유지해요.

from semantic_kernel.agents import ChatCompletionAgent, ChatHistoryAgentThread

agent = ChatCompletionAgent(...)

thread: ChatHistoryAgentThread = None

# Generate the streamed agent response(s)
async for response in agent.invoke_stream(messages="user input", thread=thread):
  # Process streamed response(s)...
  thread = response.thread

C#에서는 await foreach (StreamingChatMessageContent response in agent.InvokeStreamingAsync(message, agentThread)) 형태로 스트리밍 응답을 순회하고, 필요하면 agentThread.GetMessagesAsync() 로 thread에 추가된 메시지를 읽을 수 있어요.

OpenAIAssistantAgent 에서 스트리밍 응답

OpenAIAssistantAgent 에서 스트리밍 응답을 호출하면, 어시스턴트가 대화 상태를 원격 스레드(remote thread) 로 유지해요. 필요하면 원격 스레드에서 메시지를 읽을 수 있어요.

from semantic_kernel.agents import AssistantAgentThread, AzureAssistantAgent, OpenAIAssistantAgent

agent = OpenAIAssistantAgent(...)  # or = AzureAssistantAgent(...)

thread: AssistantAgentThread = None

# Generate the streamed agent response(s)
async for response in agent.invoke_stream(messages="user input", thread=thread):
  # Process streamed response(s)...
  thread = response.thread

# Read the messages from the remote thread
async for response in thread.get_messages():
  # Process messages

# Delete the thread
await thread.delete()

기존 thread_id 로 스레드를 만들려면 AssistantAgentThread(client=client, thread_id="your-existing-thread-id") 처럼 생성자에 넘기면 돼요. C#도 같은 방식으로 OpenAIAssistantAgentThread(assistantClient, "your-existing-thread-id") 형태예요.

스트리밍 응답에서 중간 메시지 처리하기

스트리밍 응답의 특성상 LLM 모델이 텍스트를 점진적 조각으로 돌려주므로, 전체 응답을 기다리지 않고도 UI나 콘솔에서 더 빠르게 렌더링할 수 있어요. 여기서 한 발 더 나아가, 함수 호출 결과 같은 중간 콘텐츠를 처리하고 싶을 수 있어요. 스트리밍 응답을 호출할 때 콜백 함수를 제공하면 됩니다. 이 콜백은 ChatMessageContent 로 감싼 완전한 메시지를 받아요.

Python에서 agent.invoke_stream(...) 에 on_intermediate_message 콜백을 넘기면, 에이전트의 최종 응답을 만드는 과정에서 생성되는 중간 메시지를 받을 수 있어요. 콜백을 주지 않으면 에이전트는 중간 도구 호출 단계 없이 최종 응답만 반환해요.

import asyncio
from typing import Annotated

from semantic_kernel.agents import AzureResponsesAgent
from semantic_kernel.contents import ChatMessageContent, FunctionCallContent, FunctionResultContent
from semantic_kernel.functions import kernel_function


class MenuPlugin:
    """A sample Menu Plugin used for the concept sample."""

    @kernel_function(description="Provides a list of specials from the menu.")
    def get_specials(self, menu_item: str) -> Annotated[str, "Returns the specials from the menu."]:
        return """
        Special Soup: Clam Chowder
        Special Salad: Cobb Salad
        Special Drink: Chai Tea
        """

    @kernel_function(description="Provides the price of the requested menu item.")
    def get_item_price(
        self, menu_item: Annotated[str, "The name of the menu item."]
    ) -> Annotated[str, "Returns the price of the menu item."]:
        return "$9.99"


# This callback is called for each intermediate message,
# allowing one to handle FunctionCallContent and FunctionResultContent.
async def handle_streaming_intermediate_steps(message: ChatMessageContent) -> None:
    for item in message.items or []:
        if isinstance(item, FunctionResultContent):
            print(f"Function Result:> {item.result} for function: {item.name}")
        elif isinstance(item, FunctionCallContent):
            print(f"Function Call:> {item.name} with arguments: {item.arguments}")
        else:
            print(f"{item}")


async def main():
    # Create the client and a Semantic Kernel agent for the OpenAI Responses API
    client, model = AzureResponsesAgent.setup_resources()
    agent = AzureResponsesAgent(
        ai_model_id=model,
        client=client,
        instructions="Answer questions about the menu.",
        name="Host",
        plugins=[MenuPlugin()],
    )
    thread = None
    try:
        for user_input in USER_INPUTS:
            print(f"# {AuthorRole.USER}: '{user_input}'")
            first_chunk = True
            async for response in agent.invoke_stream(
                messages=user_input,
                thread=thread,
                on_intermediate_message=handle_streaming_intermediate_steps,
            ):
                thread = response.thread
                if first_chunk:
                    print(f"# {response.name}: ", end="", flush=True)
                    first_chunk = False
                print(response.content, end="", flush=True)
            print()
    finally:
        await thread.delete() if thread else None

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

실행하면 함수 호출과 결과가 스트리밍 중에 중간 메시지로 찍힙니다. "What is the special soup?" 질문에 MenuPlugin-get_specials 호출과 그 결과가 출력된 뒤, 에이전트가 "The special soup today is Clam Chowder…"라고 답하는 식이에요.

다음 단계

에이전트와 함께 템플릿 사용하기 나 Agent orchestration 문서를 확인해 보세요.