LangChain은 다양한 언어 모델과의 상호작용을 간편하게 할 수 있는 프레임워크이다. 이 중에서도 테스트나 데모 목적으로 가짜 응답을 생성할 수 있는 도구로 FakeListLLM
과 FakeStreamingListLLM
을 제공한다. 이 두 도구는 LLM(대형 언어 모델)의 응답을 시뮬레이션할 수 있으며, 스트리밍 여부에 따라 다르게 동작한다.
여기에서는 FakeListLLM
과 FakeStreamingListLLM
의 차이점, 스트리밍을 사용하는 것과 사용하지 않는 것의 예제, 그리고 sleep
을 활용한 스트리밍 응답의 지연 효과를 알아보자.
1. FakeListLLM
- 기본적인 가짜 응답 생성
FakeListLLM
은 가짜 응답을 생성하기 위한 기본적인 도구이다. 이 도구는 가짜 응답 리스트를 순서대로 반환하며, 동기적으로 작동한다.
예제: FakeListLLM
사용하기
from langchain.llms.fake import FakeListLLM
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain_core.language_models import FakeStreamingListLLM
from langchain_core.output_parsers import StrOutputParser
# 가짜 응답 리스트 정의
fake_responses = [
"서울은 대한민국의 수도입니다. 대한민국의 정치, 경제, 문화의 중심지로서 중요한 역할을 하고 있습니다. 추가로 궁금한 사항이 있으시면 말씀해 주세요!"
]
# LLMChain 생성
fake_llm = FakeListLLM(
responses=fake_responses,
)
template = "{country}의 수도는?"
prompt = PromptTemplate.from_template(template=template)
chain = prompt | fake_llm | StrOutputParser()
response = chain.invoke({"country": "서울"})
print(response)
출력 예시:
서울은 대한민국의 수도입니다. 대한민국의 정치, 경제, 문화의 중심지로서 중요한 역할을 하고 있습니다. 추가로 궁금한 사항이 있으시면 말씀해 주세요!
FakeListLLM
은 단순히 리스트에 정의된 가짜 응답을 반환한다. 모든 응답이 한 번에 반환되므로 스트리밍이 필요하지 않은 간단한 테스트 시나리오에 적합하다.
2. FakeStreamingListLLM
- 스트리밍 방식의 가짜 응답 생성
FakeStreamingListLLM
은 스트리밍 방식으로 가짜 응답을 반환하는 도구이다. 이는 FakeListLLM
과 달리, 응답을 여러 조각으로 나누어 순차적으로 반환한다. 이 방식은 대용량 응답을 처리하거나, 실시간 데이터를 제공하는 시나리오를 테스트할 때 유용하다.
예제: FakeStreamingListLLM
사용하기
import asyncio
from langchain.prompts import PromptTemplate
from langchain_core.language_models import FakeStreamingListLLM
from langchain_core.output_parsers import StrOutputParser
# 가짜 응답 리스트 정의
fake_responses = [
"서울은 대한민국의 수도입니다. 대한민국의 정치, 경제, 문화의 중심지로서 중요한 역할을 하고 있습니다. 추가로 궁금한 사항이 있으시면 말씀해 주세요!"
]
# LLMChain 생성
fake_llm = FakeStreamingListLLM(responses=fake_responses)
template = "{country}의 수도는?"
prompt = PromptTemplate.from_template(template=template)
chain = prompt | fake_llm | {"result": StrOutputParser()}
async def astream():
async for chunk in chain.astream({"country": "서울"}):
print(chunk)
asyncio.run(astream())
출력 예시:
{'result': '서'}
{'result': '울'}
{'result': '은'}
{'result': ' '}
{'result': '대'}
{'result': '한'}
{'result': '민'}
{'result': '국'}
{'result': '의'}
...
{'result': '주'}
{'result': '세'}
{'result': '요'}
{'result': '!'}
FakeStreamingListLLM
은 응답을 스트리밍 방식으로 반환하여, 실시간 응답을 처리하는 것처럼 동작한다. 이는 특히 사용자 인터페이스(UI)에서 실시간 응답을 처리하거나, 대규모 데이터 처리를 시뮬레이션하는 데 유용하다.
3. sleep
을 사용한 스트리밍 지연 효과
실제 LLM 응답에서는 네트워크 지연이나 처리 시간이 발생할 수 있다. 이를 시뮬레이션하기 위해 sleep
을 사용하여 응답 간의 지연 효과를 추가할 수 있다.
예제: sleep
을 활용한 스트리밍 지연 효과
import asyncio
from langchain.prompts import PromptTemplate
from langchain_core.language_models import FakeStreamingListLLM
from langchain_core.output_parsers import StrOutputParser
# 가짜 응답 리스트 정의
fake_responses = [
"서울은 대한민국의 수도입니다. 대한민국의 정치, 경제, 문화의 중심지로서 중요한 역할을 하고 있습니다. 추가로 궁금한 사항이 있으시면 말씀해 주세요!"
]
# LLMChain 생성
fake_llm = FakeStreamingListLLM(responses=fake_responses, sleep=0.1)
template = "{country}의 수도는?"
prompt = PromptTemplate.from_template(template=template)
chain = prompt | fake_llm | {"result": StrOutputParser()}
async def astream():
async for chunk in chain.astream({"country": "서울"}):
print(chunk)
asyncio.run(astream())
출력 예시:
{'result': '서'}
(1초 후)
{'result': '울'}
(1초 후)
{'result': '은'}
(1초 후)
...
{'result': '!'}
이 예제에서는 각 응답 조각 사이에 0.1초의 지연을 추가했다. 이와 같은 방법을 사용하면 실제 시스템에서 발생할 수 있는 지연 시간을 시뮬레이션할 수 있으며, 이를 통해 더 현실적인 테스트를 수행할 수 있다.
4. 언제 FakeListLLM
과 FakeStreamingListLLM
을 사용할까?
FakeListLLM
:- 간단한 테스트나 동기적으로 처리할 수 있는 응답 시나리오에 적합하다.
- 전체 응답을 한 번에 받을 때 유용하다.
FakeStreamingListLLM
:- 대용량 응답을 처리하거나, 스트리밍 방식의 실시간 응답을 테스트할 때 유용하다.
- 사용자 인터페이스와의 통합 테스트에서 응답을 부분적으로 보여줄 때 유용하다.