ChatOpenAI 와 FastAPI 를 이용해 Streaming 채팅 구현하기

AI 가 그려줬는데 좀..;;

ChatOpenAI 와 FastAPI 를 이용해 Streaming 채팅 구현하기

 

이번 글에서는 FastAPI + SSE 기반 스트리밍 챗 서버를 만들고, OpenAI SDK 대신 LangChain의 ChatOpenAI를 사용해 구현하는 방법을 알아봅니다.

• client / server 완전 분리 구조
• SSE(text/event-stream) 기반 실시간 스트리밍
• ChatOpenAI의 astream()을 사용한 asyncio-native 구현

 

server.py – ChatOpenAI 기반 스트리밍 서버

• 요청 메시지를 LangChain 메시지로 변환
• ChatOpenAI를 사용해 스트리밍 응답 생성
• SSE 형식으로 클라이언트에 전달

import json
import uvicorn

from dotenv                  import load_dotenv
from pydantic                import BaseModel
from typing                  import List
from typing                  import Optional
from pydantic                import Field
from langchain_core.messages import HumanMessage
from langchain_core.messages import AIMessage
from langchain_core.messages import SystemMessage
from langchain_openai        import ChatOpenAI
from fastapi                 import FastAPI
from fastapi                 import HTTPException
from fastapi.responses       import StreamingResponse

load_dotenv()

class Message(BaseModel):
    role    : str
    content : str

class ChatRequest(BaseModel):
    messages    : List[Message]
    model       : str             = "gpt-4o-mini"
    temperature : Optional[float] = Field(default = 0.7, ge = 0, le = 2)
    max_tokens  : Optional[int  ] = None

def getLangchainMessageList(messageList : List[Message]):
    targetMessageList = []
    for message in messageList:
        if message.role == "system":
            targetMessageList.append(SystemMessage(content = message.content))
        elif message.role == "user":
            targetMessageList.append(HumanMessage(content = message.content))
        elif message.role == "assistant":
            targetMessageList.append(AIMessage(content = message.content))
        else:
            raise ValueError(f"Unknown role : {message.role}")
    return targetMessageList

async def generateStream(chatRequest : ChatRequest):
    try:
        chatOpenAI = ChatOpenAI(
            model       = chatRequest.model,
            temperature = chatRequest.temperature,
            max_tokens  = chatRequest.max_tokens,
            streaming   = True
        )
        messageList = getLangchainMessageList(chatRequest.messages)
        async for aiMessageChunk in chatOpenAI.astream(messageList):
            if aiMessageChunk.content:
                yield f"data: {json.dumps({'type' : 'content', 'content' : aiMessageChunk.content})}\n\n"
        yield "data: [DONE]\n\n"
    except Exception as exception:
        print(f"STREAMING ERROR : {str(exception)}")
        yield f"data: {json.dumps({'type' : 'error', 'error' : str(exception)})}\n\n"

fastAPI = FastAPI()

@fastAPI.post("/v1/chat/completion")
async def processChatCompletion(chatRequest : ChatRequest):
    if not chatRequest.messages:
        raise HTTPException(status_code = 400, detail = "Messages cannot be empty")
    return StreamingResponse(generateStream(chatRequest), media_type = "text/event-stream", headers= {"Cache-Control" : "no-cache", "Connection" : "keep-alive"})

@fastAPI.get("/health")
async def processHealth():
    return {"status" : "healthy"}

if __name__ == "__main__":
    uvicorn.run(fastAPI, host = "0.0.0.0", port = 8000)

 

client.py – 스트리밍 챗 클라이언트

• 메시지리스트 관리
• FastAPI 서버로 메시지 전송
• SSE 응답을 실시간 출력

import httpx
import json
import asyncio

from datetime import datetime
from typing   import List
from typing   import Dict
from typing   import Optional

class ChatClient:
    def __init__(self, serverURL : str = "http://localhost:8000"):
        self.serverURL                          = serverURL
        self.messageList : List[Dict[str, str]] = []
        self.timeout                            = httpx.Timeout(60.0, connect = 5.0)

    def addMessage(self, role : str, content : str) -> None:
        self.messageList.append(
            {
                "role"      : role,
                "content"   : content,
                "timestamp" : datetime.now().isoformat()
            }
        )

    def getMessageList(self) -> List[Dict[str, str]]:
        return [{"role" : message["role"], "content" : message["content"]} for message in self.messageList]

    def clearMessageList(self) -> None:
        self.messageList.clear()
        print("MESSAGE LIST CLEARED")

    async def checkServerHealth(self) -> bool:
        try:
            async with httpx.AsyncClient(timeout = self.timeout) as asyncClient:
                response = await asyncClient.get(f"{self.serverURL}/health")
                return response.status_code == 200
        except Exception as exception:
            print(f"HEALTH CHECK FAILED : {exception}")
            return False

    async def astream(self, userInput : str, model : str = "gpt-4o-mini", temperature : float = 0.7) -> Optional[str]:
        self.addMessage("user", userInput)
        requestDictionary = {
            "messages"    : self.getMessageList(),
            "model"       : model,
            "temperature" : temperature
        }
        async with httpx.AsyncClient(timeout = self.timeout) as asyncClient:
            try:
                async with asyncClient.stream("POST", f"{self.serverURL}/v1/chat/completion", json = requestDictionary) as response:
                    if response.status_code != 200:
                        print(f"HTTP ERROR : {response.status_code}")
                        return None
                    fullResponse = ""
                    print("ASSISTANT : ", end = "", flush = True)
                    async for line in response.aiter_lines():
                        if not line.startswith("data: "):
                            continue
                        dataLine = line[6:]
                        if dataLine == "[DONE]":
                            break
                        try:
                            dataDictionary = json.loads(dataLine)
                            if dataDictionary.get("type") == "content":
                                content = dataDictionary["content"]
                                print(content, end = "", flush = True)
                                fullResponse += content
                            elif dataDictionary.get("type") == "error":
                                print()
                                print(f"ERROR : {dataDictionary.get('error')}")
                                return None
                        except json.JSONDecodeError:
                            continue
                    print()
                    if fullResponse:
                        self.addMessage("assistant", fullResponse)
                    return fullResponse
            except httpx.TimeoutException:
                print("REQUEST TIMEOUT")
            except httpx.ConnectError:
                print("CANNOT CONNECT TO SERVER")
            except Exception:
                print("UNEXPECTED ERROR", exc_info = True)
        return None

    def showMessageList(self) -> None:
        print()
        print("-" * 50)
        print("CURRENT MESSAGE LIST")
        print("-" * 50)
        if not self.messageList:
            print("(EMPTY)")
        else:
            for i, message in enumerate(self.messageList, 1):
                content = message["content"]
                if len(content) > 100:
                    content = content[:100] + "..."
                print(f"{i}. [{message['role'].upper()}] {content}")
        print("-" * 50)
        print()

async def main():
    chatClient = ChatClient()
    print("COMMANDS : quit | clear | show | health")
    print()
    if not await chatClient.checkServerHealth():
        print("SERVER HEALTH CHECK FAILED")
        print()
    while True:
        try:
            userInput = input("YOU : ").strip()
            if userInput in ("quit", "exit"):
                break
            elif userInput == "clear":
                chatClient.clearMessageList()
            elif userInput == "show":
                chatClient.showMessageList()
            elif userInput == "health":
                ok = await chatClient.checkServerHealth()
                print("HEALTHY" if ok else "UNHEALTHY")
            elif userInput:
                await chatClient.astream(userInput)
        except KeyboardInterrupt:
            break

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

 

requirements.txt

fastapi>=0.115.0
uvicorn[standard]>=0.32.0
httpx>=0.28.0
python-dotenv>=1.0.1
pydantic>=2.10.0
langchain>=0.2.14
langchain-openai>=0.1.22

 

실행

# 가상환경 활성화
python -m venv venv
.\venv\Scripts\activate

# 가상환경 진입 후 패키지 설치
pip install -r requirements.txt

server 실행

(venv) PS D:\python-vm\an> python server.py

터미널 새로열고 가상환경으로 진입해 client 를 실행합니다.

.\venv\Scripts\activate
(venv) PS D:\python-vm\an> python client.py

 

실행 결과

바나나와 사과에 대해 물어보고 둘중 어느게 더 영양가 있는지 물어봐서 앞의 메세지를 가지고 응답을 하는지 확인합니다.

(venv) PS D:\python-vm\an> python client.py       
FastAPI Chat Client
Commands: quit | clear | show | health

INFO:httpx:HTTP Request: GET http://localhost:8000/health "HTTP/1.1 200 OK"
You: 사과에 대해서 설명해
INFO:httpx:HTTP Request: POST http://localhost:8000/chat/stream "HTTP/1.1 200 OK"
Assistant: 사과는 대표적인 과일 중 하나로, 주로 가을철에 수확됩니다. 사과는 다양한 품종이 있으며, 그에 따라 색깔, 크기, 맛이 다릅니다. 일반적으로 사과는 붉은색, 초록색, 노란색 등 다양한 색상을 가지고 있습니다. 

사과의 주요 영양소로는 비타민 C, 식이섬유, 항산화 물질 등이 있으며, 이는 면역력 증진, 소화 개선, 심혈관 건강에 도움을 줄 수 있습니다. 또한, 사과는 칼로리가 상대적으로 낮아 다이어트 식품으로도 인기가 많습니다.

사과는 생으로 먹거나, 주스, 파이, 잼 등 다양한 형태로 가공되어 소비됩니다. 또한, 사과는 여러 문화에서 상징적인 의미를 가지며, 특히 유혹이나 지혜의 상징으로 자주 언급되곤 합니다. 

사과를 선택할 때는 껍질에 손상이 없고, 탄력이 있는 것을 고르는 것이 좋습니다. 보관할 때는 서늘하고 건조한 곳에 두는 것이 최적입니다.
You: 바나나에 대해서 설명해
INFO:httpx:HTTP Request: POST http://localhost:8000/chat/stream "HTTP/1.1 200 OK"
Assistant: 바나나는 대표적인 열대 과일로, 주로 중앙아메리카, 남아메리카, 아프리카 등에서 재배됩니다. 바나나는 길고 곡선형의 형태를 가지고 있으며, 껍질은 보통 노란색이나 초록색으로, 성숙하면서 색이 변합니다. 익은 바나나는 부드럽고 달콤 한 맛이 특징입니다.

바나나는 영양가가 높고, 주로 다음과 같은 영양소가 포함되어 있습니다:

1. **비타민**: 비타민 C와 비타민 B6가 풍부하여 면역력을 높이고 에너지를 증진하는 데 도움이 됩니다.
2. **미네랄**: 특히 칼륨이 풍부하여, 심장 건강과 혈압 조절에 도움을 줄 수 있습니다.
3. **식이섬유**: 소화를 돕고 장 건강에 긍정적인 영향을 미치는 식이섬유가 포함되어 있습니다.

바나나는 간편하게 먹을 수 있는 과일로, 스무디, 디저트, 샐러드 등 다양한 요리에 활용됩니다. 또한, 운동 후에 에너지를 보충하는 간식으로도 인기가 많습니다.
바나나는 간편하게 먹을 수 있는 과일로, 스무디, 디저트, 샐러드 등 다양한 요리에 활용됩니다. 또한, 운동 후에 에너지를 보충하는 간식으로도 인기가 많습니다.

바나나를 선택할 때는 껍질에 검은 점이 많지 않고, 탄력이 있는 것을 고르는 것이 좋습니다. 보관할 때는 서늘한 곳에서 보관하며, 익은 바나나는 냉장고에 두면 껍질은 갈색으로 변할 수 있지만 과육은 오랫동안 신선함을 유지할 수 있습니다.        
You: 어느게 더 영양가가 많아?
INFO:httpx:HTTP Request: POST http://localhost:8000/chat/stream "HTTP/1.1 200 OK"
Assistant: 사과와 바나나는 각기 다른 영양소를 가지고 있어 직접적으로 어느 것이 더 영양가가 많다고 단정짓기는 어렵습니다. 두 과일 모두 건강에 유익한 성분을 포함하고 있으며, 각각의 장점이 있습니다.

### 사과의 영양적 장점:
You: 어느게 더 영양가가 많아?
INFO:httpx:HTTP Request: POST http://localhost:8000/chat/stream "HTTP/1.1 200 OK"
Assistant: 사과와 바나나는 각기 다른 영양소를 가지고 있어 직접적으로 어느 것이 더 영양가가 많다고 단정짓기는 어렵습니다. 두 과일 모두 건강에 유익한 성분을 포함하고 있으며, 각각의 장점이 있습니다.

### 사과의 영양적 장점:
INFO:httpx:HTTP Request: POST http://localhost:8000/chat/stream "HTTP/1.1 200 OK"
Assistant: 사과와 바나나는 각기 다른 영양소를 가지고 있어 직접적으로 어느 것이 더 영양가가 많다고 단정짓기는 어렵습니다. 두 과일 모두 건강에 유익한 성분을 포함하고 있으며, 각각의 장점이 있습니다.

### 사과의 영양적 장점:
Assistant: 사과와 바나나는 각기 다른 영양소를 가지고 있어 직접적으로 어느 것이 더 영양가가 많다고 단정짓기는 어렵습니다. 두 과일 모두 건강에 유익한 성분을 포함하고 있으며, 각각의 장점이 있습니다.

### 사과의 영양적 장점:

### 사과의 영양적 장점:
### 사과의 영양적 장점:
- **식이섬유**: 사과는 특히 펙틴이라는 수용성 식이섬유가 풍부하여 소화 건강에 도움을 줄 수 있습니다.
- **비타민 C**: 면역력 증진과 항산화 작용에 기여합니다.
- **항산화 물질**: 사과에는 플라보노이드와 폴리페놀 같은 항산화 물질이 포함되어 있어 심혈관 건강에 이롭습니다.
- **비타민 C**: 면역력 증진과 항산화 작용에 기여합니다.
- **항산화 물질**: 사과에는 플라보노이드와 폴리페놀 같은 항산화 물질이 포함되어 있어 심혈관 건강에 이롭습니다.
- **항산화 물질**: 사과에는 플라보노이드와 폴리페놀 같은 항산화 물질이 포함되어 있어 심혈관 건강에 이롭습니다.

### 바나나의 영양적 장점:
- **칼륨**: 바나나는 칼륨이 풍부하여 혈압 조절과 심장 건강에 도움을 줄 수 있습니다.
- **비타민 B6**: 에너지 대사에 중요한 역할을 하며, 뇌 건강에도 기여합니다.
- **간편한 에너지 공급원**: 운동 후 빠른 에너지를 제공할 수 있어 스낵으로 인기가 많습니다.

### 결론
사과와 바나나 모두 영양가가 높고 건강에 이로운 과일입니다. 개인의 필요와 건강 목표에 따라 선택하는 것이 좋습니다. 예를 들어, 섬유소 섭취를 늘리고 싶다면 사과를, 칼륨과 에너지를 보충하고 싶다면 바나나를 선택하는 것이 좋습니다. 다양한 과일을 골고루 섭취하는 것이 가장 이상적입니다.