연구내용
개요
- Chainlit은 오픈소스 파이썬 패키지로, 빠르게 프로덕션 준비가 된 대화형 AI를 개발을 도와줌
목표
- Chainlit을 사용해 최소 기능 제품(MVP) 수준의 데모 챗봇을 개발하는 과정
- LangChain 라이브러리와의 통합을 통해 문서에서 정보를 추출하고 사용자의 질문에 답변하는 챗봇을 만드는 방법 설명
시스템 설정 및 초기화
시스템 환경 설정
Chainlit과 필요한 라이브러리를 설치하고 환경을 설정함. 이 과정은 빠른 통합과 플랫폼 독립성을 보장함
1
2
pip install chainlit langchain openai chroma
코드 구조
챗봇의 기본 코드는 Python을 기반으로 하며, 다음과 같은 주요 구성 요소로 이루어짐
- 문서 로더: PDF와 텍스트 파일에서 데이터를 로드
- 텍스트 분할기: 긴 텍스트를 일정한 크기로 나눠 처리
- 임베딩 및 벡터 스토어: OpenAI 임베딩을 사용해 텍스트 데이터를 벡터화하고 검색 가능한 벡터 스토어에 저장
- 대화형 검색 체인: 사용자 입력에 따라 관련 정보를 검색하고 응답을 생성
초기화 코드 설명
필요한 라이브러리와 모듈을 불러옴. 여기에는 문서 로더, 텍스트 분할기, 임베딩 생성기, 벡터 스토어, 그리고 Chainlit 관련 모듈이 포함됨
문서 처리 및 챗봇 실행
파일 처리 및 문서 검색 설정
사용자가 업로드한 파일을 처리하고 검색 가능한 데이터로 변환함. Chainlit의 강력한 기능을 활용하여 사용자로부터 파일을 받고 이를 기반으로 검색 가능한 문서 집합을 생성함
파일 처리 함수
1
2
3
4
5
6
7
8
9
10
11
12
def process_file(file: AskFileResponse):
if file.type == "text/plain":
Loader = TextLoader
elif file.type == "application/pdf":
Loader = PyPDFLoader
loader = Loader(file.path)
documents = loader.load()
docs = text_splitter.split_documents(documents)
return docs
사용자가 업로드한 PDF 또는 텍스트 파일을 로드하여 텍스트로 변환하고 특정 크기로 나눈 문서 목록을 반환함
문서 검색 초기화
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
def get_docsearch(file: AskFileResponse):
docs = process_file(file)
cl.user_session.set("docs", docs)
collection_name = file.id
if collection_name in collections:
docsearch = Chroma(
collection_name=collection_name, embedding_function=embeddings
)
else:
docsearch = Chroma.from_documents(
docs, embeddings, collection_name=collection_name
)
collections.add(collection_name)
return docsearch
문서를 처리하고 벡터 스토어에 저장하여 사용자가 질문할 때 검색할 수 있도록 설정함
챗봇 대화 시작
사용자와의 상호작용을 관리하는 부분임. Chainlit의 기능을 활용하여 챗봇 세션을 관리하고 사용자 입력에 따라 적절한 응답을 생성함
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
@cl.on_chat_start
async def start():
files = None
while files is None:
files = await cl.AskFileMessage(
content=welcome_message,
accept=["text/plain", "application/pdf"],
max_size_mb=20,
timeout=180,
).send()
file = files[0]
msg = cl.Message(content=f"Processing `{file.name}`...")
await msg.send()
docsearch = await cl.make_async(get_docsearch)(file)
message_history = ChatMessageHistory()
memory = ConversationBufferMemory(
memory_key="chat_history",
output_key="answer",
chat_memory=message_history,
return_messages=True,
)
chain = ConversationalRetrievalChain.from_llm(
ChatOpenAI(model_name="gpt-4o-mini", temperature=0, streaming=True),
chain_type="stuff",
retriever=docsearch.as_retriever(),
memory=memory,
return_source_documents=True,
)
msg.content = f"`{file.name}` processed. You can now ask questions!"
await msg.update()
cl.user_session.set("chain", chain)
사용자 입력 처리
사용자가 입력한 질문을 받아 대화형 검색 체인을 통해 응답을 생성함
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
@cl.on_message
async def main(message: cl.Message):
chain = cl.user_session.get("chain") # type: ConversationalRetrievalChain
cb = cl.AsyncLangchainCallbackHandler()
async with cl.Step(name="ConversatinalRetrievalChain", type="retriever") as step:
step.input = message.content
res = await chain.ainvoke(message.content, callbacks=[cb])
for source in res["source_documents"]:
await step.stream_token(str(source))
answer = res["answer"]
source_documents = res["source_documents"] # type: List[Document]
text_elements = [] # type: List[cl.Text]
if source_documents:
for source_idx, source_doc in enumerate(source_documents):
source_name = f"source_{source_idx}"
text_elements.append(
cl.Text(
content=source_doc.page_content, name=source_name, display="side"
)
)
source_names = [text_el.name for text_el in text_elements]
if source_names:
answer += f"\\nSources: {', '.join(source_names)}"
else:
answer += "\\nNo sources found"
await cl.Message(content=answer, elements=text_elements).send()
사용자 입력을 기반으로 적절한 응답을 생성하고 관련 소스 문서를 함께 제공함. 이러한 구조는 사용자가 질문을 했을 때 문서 내에서 정보를 추출하고 그에 대한 답변을 생성하여 사용자의 이해를 도움
Step 클래스 활용 연구
Chainlit의 Step 클래스는 컨텍스트 관리자(Context Manager)로, 체인릿 앱에서 단계(Step)를 생성하는 데 사용됨. Step은 컨텍스트 관리자에 진입할 때 생성되고, 종료할 때 클라이언트에 업데이트됨
Step 클래스의 주요 매개변수
- name: 단계의 이름을 지정
- type: 단계의 유형을 지정하여 모니터링 및 디버깅에 유용
- elements: 단계에 첨부할 요소의 리스트
- root: 기본적으로 단계는 이전 단계/메시지 아래에 중첩됨. 이를
True로 설정하면 루트 단계가 됨 - language: 출력의 언어를 지정
Step 사용 예시
기본 사용
1
2
3
4
5
6
7
@cl.on_message
async def main():
async with cl.Step(name="Test") as step:
step.input = "hello"
step.output = "world"
스트림 출력
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
@cl.on_message
async def main(msg: cl.Message):
root = True
async with cl.Step(name="gpt4", type="llm", root=root) as step:
step.input = msg.content
stream = await client.chat.completions.create(
messages=[{"role": "user", "content": msg.content}],
stream=True,
model="gpt-4",
temperature=0,
)
async for part in stream:
delta = part.choices[0].delta
if delta.content:
await step.stream_token(delta.content)
단계 중첩
1
2
3
4
5
6
7
8
9
10
11
12
@cl.on_chat_start
async def main():
async with cl.Step(name="Parent step") as parent_step:
parent_step.input = "Parent step input"
async with cl.Step(name="Child step") as child_step:
child_step.input = "Child step input"
child_step.output = "Child step output"
parent_step.output = "Parent step output"
단계 업데이트
1
2
3
4
5
6
7
8
9
10
11
12
@cl.on_chat_start
async def main():
async with cl.Step(name="Parent step") as step:
step.input = "Parent step input"
step.output = "Parent step output"
await cl.sleep(2)
step.output = "Parent step output updated"
await step.update()
단계 제거
1
2
3
4
5
6
7
8
9
10
11
@cl.on_chat_start
async def main():
async with cl.Step(name="Parent step") as step:
step.input = "Parent step input"
step.output = "Parent step output"
await cl.sleep(2)
await step.remove()
Chainlit의 Step 클래스는 단계별 처리를 통해 복잡한 대화 흐름을 관리하고 디버깅할 수 있는 강력한 도구임. 이를 통해 대화의 중간 단계를 시각화하고, 출력 결과에 대한 중간 상태를 이해할 수 있음
결과 화면
결론
Chainlit을 사용하면 빠르게 MVP 수준의 챗봇을 개발할 수 있음. 이 시스템은 PDF 및 텍스트 파일을 처리하여 사용자 질문에 대한 정확한 답변을 제공하며, 이를 통해 사용자는 Chainlit의 강력한 기능을 활용하여 다양한 대화형 AI 애플리케이션을 구축할 수 있음