1. 라이브러리들을 임포트(import)하는 섹션
from langchain_community.llms import Ollama # Ollama LLM 사용
from langchain_core.prompts import PromptTemplate # 프롬프트 템플릿
from langgraph.graph import StateGraph, END # LangGraph 상태 머신
from typing import TypedDict # 타입 정의용
import sqlite3 # SQLite 사용
from langchain_community.document_loaders import TextLoader # 텍스트 문서 로더
from langchain.text_splitter import CharacterTextSplitter # 문서 분할 도구
from langchain_community.vectorstores import Chroma # Chroma 벡터스토어
from langchain.embeddings import OllamaEmbeddings # Ollama 임베딩 생성
from langchain.chains.combine_documents import create_stuff_documents_chain # 문서 기반 QA 체인
TextLoader, CharacterTextSplitter, Chroma, OllamaEmbeddings, create_stuff_documents_chain 등은 외부 지식(문서)을 검색하여 LLM의 응답을 보강하는 RAG 패턴을 구현하기 위한 것입니다.
2. RAG를 위한 벡터스토어
# 0. RAG 벡터스토어 생성 (최초 1회 실행 후 주석처리 가능)
def build_rag_chroma():
with open("rag_data/medical.txt", "w", encoding="utf-8") as f: # 텍스트 파일 저장
f.write("""감기는 콧물, 기침, 인후통 등의 가벼운 증상이 있는 바이러스성 감염병입니다.\n
독감은 인플루엔자 바이러스 감염으로 고열, 근육통, 기침, 피로가 발생합니다.\n
코로나19는 고열, 마른기침, 후각상실 등을 유발하며 중증으로 발전할 수 있습니다.""")
loader = TextLoader("rag_data/medical.txt", encoding="utf-8") # 텍스트 로드
docs = loader.load() # 문서 리스트 로드
splitter = CharacterTextSplitter(chunk_size=300, chunk_overlap=20) # 텍스트 분할기
split_docs = splitter.split_documents(docs) # 문서 나누기
embeddings = OllamaEmbeddings(model="exaone3.5:2.4b") # 임베딩 생성기
db = Chroma.from_documents(split_docs, embedding=embeddings, persist_directory="./rag_chroma") # Chroma 저장
db.persist() # 저장
build_rag_chroma() # 최초 실행 시만 주석 해제
medical.txt라는 간단한 텍스트 파일을 직접 생성하여 RAG에 사용할 데이터를 준비합니다. 실제 시나리오에서는 이 부분이 외부 데이터베이스, API, 파일 시스템 등에서 데이터를 가져오는 복잡한 과정으로 대체될 수 있습니다.TextLoader로 텍스트 파일을 로드하고, CharacterTextSplitter를 사용하여 이를 의미 있는 작은 단위(청크)로 나눕니다. 이는 긴 문서 전체를 한 번에 임베딩하거나 LLM에 전달하기 어렵기 때문에 효율적인 검색과 처리를 위해 필수적인 과정입니다.exaone3.5:2.4b)을 사용하여 분할된 각 문서 청크의 의미를 나타내는 벡터(숫자 배열)를 생성합니다. 이 벡터들은 고차원 공간에서 텍스트의 의미적 유사성을 표현합니다.
Chroma라는 벡터 데이터베이스에 저장됩니다. persist_directory 옵션을 통해 이 데이터를 로컬 디스크에 영구적으로 저장하여, 한 번 구축해두면 다음 실행 시에는 다시 구축할 필요 없이 로드하여 사용할 수 있게 됩니다.
3. SQLite 데이터베이스를 설정
# 1. SQLite 샘플 데이터베이스 생성 (최초 1회 실행 후 주석처리 가능)
def build_sqlite_db():
conn = sqlite3.connect("disease_knowledge.db") # DB 연결
cursor = conn.cursor() # 커서 생성
cursor.execute("""
CREATE TABLE IF NOT EXISTS diseases (
name TEXT PRIMARY KEY,
care TEXT
)
""") # 테이블 생성
sample_data = [
("감기", "수분을 충분히 섭취하고 휴식을 취하세요. 필요 시 해열제를 복용하세요."),
("독감", "타미플루 등 항바이러스제를 초기 투여하고 휴식이 필요합니다."),
("코로나19", "자가격리와 함께 해열제 복용 및 호흡기 증상 관리가 필요합니다."),
]
cursor.executemany("INSERT OR IGNORE INTO diseases VALUES (?, ?)", sample_data) # 데이터 삽입
conn.commit() # 커밋
conn.close() # 연결 종료
build_sqlite_db() # 최초 실행 시만 주석 해제
disease_knowledge.db라는 SQLite 데이터베이스 파일을 생성하고, diseases라는 테이블을 만듭니다. 이 테이블은 name (질병명)과 care (치료/관리 방법) 두 개의 컬럼을 가집니다. name이 PRIMARY KEY로 지정되어 각 질병이 고유하게 식별됩니다.sample_data 리스트에 정의하고, executemany를 사용하여 효율적으로 테이블에 삽입합니다. INSERT OR IGNORE는 중복 데이터 삽입을 방지하는 유용한 기능입니다.
conn.commit()을 통해 변경 사항을 데이터베이스 파일에 반영하므로, 한 번 실행되면 데이터가 영구적으로 저장됩니다.4. LangGraph 에이전트의 "상태(State)"를 정의
# 2. 상태 정의
class AgentState(TypedDict): # 상태 타입 정의
query: str # 사용자 질의
symptoms: str # 추출된 증상
disease_candidates: str # 질병 후보
disease_info: str # 질병 설명 정보 (RAG)
result: str # 최종 응답
AgentState에 정의된 필드들은 에이전트가 어떤 정보를 필요로 하고, 각 처리 단계에서 어떤 정보를 생성하며, 최종적으로 어떤 정보를 사용자에게 제공할 것인지를 명확하게 보여줍니다.
query: 사용자의 초기 입력.symptoms:query에서 파싱된 정보 (첫 번째 LLM 호출의 결과일 수 있음).disease_candidates:symptoms를 기반으로 예측되거나 검색된 질병 목록.disease_info:disease_candidates중 하나 또는 여러 질병에 대해 RAG를 통해 가져온 상세 정보.result: 최종적으로 사용자에게 보여줄 응답.
typing 모듈에 있는 TypedDict를 사용함으로써, 딕셔너리의 키와 값의 타입이 미리 정의되어 코드 작성 시 오류를 줄이고 다른 개발자가 코드를 이해하는 데 도움을 줍니다. 이는 에이전트 상태 관리에 대한 명확한 계약을 제공합니다.5. Ollama LLM을 초기화하고, 첫 번째 에이전트 노드(extractor_agent)를 정의
# 3. LLM 초기화
llm = Ollama(model="exaone3.5:2.4b") # Ollama 모델 로딩
# 4. 에이전트 정의
extractor_prompt = PromptTemplate.from_template("""
사용자의 질문에서 증상에 해당하는 단어 또는 구를 추출하세요.
결과는 쉼표로 구분된 문자열로 출력하세요.
질문: {query}
""") # 증상 추출 프롬프트
# 사용자의 질문에서 증상에 해당되는 단어를 LLM이 추출해줄거임
def extractor_agent(state: AgentState): # 증상 추출 함수
chain = extractor_prompt | llm # 프롬프트 체인
symptoms = chain.invoke({"query": state["query"]}) # LLM 실행
return {**state, "symptoms": symptoms.strip()} # 상태에 추가
# {query:"어쩌구",
# symptoms:"기침, 목, 열",
# disease_candidates: null,
# disease_info: null,
# result: null }
Ollama(model="exaone3.5:2.4b")를 통해 에이전트의 "두뇌" 역할을 할 LLM을 로드합니다. 이 LLM은 이후 단계에서 정보 추출, 판단, 답변 생성 등 다양한 언어 관련 작업을 수행하게 됩니다.extractor_prompt 정의: 이 PromptTemplate는 LLM에게 특정 임무, 즉 사용자 질문에서 "증상"을 추출하도록 명확히 지시합니다. 결과 형식을 "쉼표로 구분된 문자열"로 지정하여 후속 처리가 용이하도록 합니다. 이는 LLM이 단순히 자유 형식의 텍스트를 생성하는 것이 아니라, 구조화된 출력을 내도록 유도하는 중요한 프롬프트 엔지니어링 기법입니다.
extractor_agent 함수:
- 이 함수는 LangGraph의 "노드"가 될 후보이며,
AgentState를 입력으로 받아서 처리하고AgentState를 반환하는 전형적인 구조를 가집니다. extractor_prompt | llm은 LangChain Expression Language (LCEL)를 사용하여 프롬프트와 LLM을 연결한 간단한 처리 체인입니다. 이 체인은query를 입력으로 받아llm을 통해symptoms를 생성합니다.return {**state, "symptoms": symptoms.strip()}는 기존state의 모든 내용을 유지하면서symptoms필드만 새로 추출된 값으로 업데이트하는 파이썬의 딕셔너리 언패킹(**) 문법을 사용합니다. 이는 상태를 불변적으로(immutable) 관리하는 좋은 패턴입니다.
6. 추출된 증상을 바탕으로 가능성 있는 질병 후보를 추정하는 두 번째 에이전트 노드(matcher_agent)를 정의
matcher_prompt = PromptTemplate.from_template("""
다음 증상 목록을 바탕으로 가장 가능성 높은 질병 이름 3개를 쉼표로 추정하세요.
증상: {symptoms}
""") # 질병 후보 추정 프롬프트
# 기침, 목, 열에 대한 질병 목차를 llm이 생각해서 내뱉음
def matcher_agent(state: AgentState): # 질병 후보 추정
chain = matcher_prompt | llm
candidates = chain.invoke({"symptoms": state["symptoms"]})
return {**state, "disease_candidates": candidates.strip()}
# {query:"어쩌구",
# symptoms:"기침, 목, 열",
# disease_candidates: "감기, 코로나, 식중독",
# disease_info: null,
# result: null }
matcher_prompt 정의: 이 프롬프트는 LLM에게 명확한 지침을 제공합니다. 단순히 증상을 나열하는 것이 아니라, "가장 가능성 높은 질병 이름 3개를 쉼표로 추정"하도록 요구하여 LLM의 출력을 특정 형식과 개수로 제한합니다. 이는 이후 단계에서 질병 후보 목록을 처리하기 용이하게 합니다.
matcher_agent 함수:
- 이 함수는
extractor_agent에서 추출한symptoms를 입력으로 받아서,matcher_prompt를 통해 LLM에 전달합니다. - LLM은 학습된 지식을 바탕으로 해당 증상과 연관성이 높은 질병들을 추론하여 반환합니다.
- 마찬가지로
return {**state, "disease_candidates": candidates.strip()}패턴을 사용하여AgentState의disease_candidates필드를 업데이트합니다.
7. 증상 > 질병 RAG 기능을 구현하는 rag_agent 노드를 정의
# 벡터스토어에서 사용자의 증상("기침, 목, 열")에 대해서 리트리벌 검색을 함.
def rag_agent(state: AgentState): # RAG 검색 기반 질병 설명 생성
embeddings = OllamaEmbeddings(model="exaone3.5:2.4b")
# 벡터스토어 스탠바이
vectorstore = Chroma(persist_directory="./rag_chroma", embedding_function=embeddings)
# 리트리벌 생성
retriever = vectorstore.as_retriever()
# rag에 쓸 프롬프트 작성
rag_prompt = PromptTemplate.from_template("""
사용자의 증상: {symptoms}
다음 참고 문서를 바탕으로 관련 질병 정보를 요약해 주세요:
{context}
""")
# rag 체인 생성
rag_chain = create_stuff_documents_chain(llm, rag_prompt)
# 벡터스토어에서 "기침, 목, 열"과 비슷한 문서'틀'을 가지고옴
docs = retriever.get_relevant_documents(state["symptoms"])
# rag_promplt에 증상과, 관련 문서(문구)를 넣어서 llm한테 프롬프트를 날림 그 대답을 변수에 대입.
rag_response = rag_chain.invoke({"symptoms": state["symptoms"], "context": docs})
return {**state, "disease_info": rag_response.strip()}
# {query:"어쩌구",
# symptoms:"기침, 목, 열",
# disease_candidates: "감기,코로나,식중독",
# disease_info: "증상과 질병 정보를 보아하니 당신은 즉각 병원에 가봐야 할 것 같아요",
# result: null }
build_rag_chroma() 함수로 구축하고 영구 저장했던 Chroma 벡터스토어를 로드합니다. 이는 실제 데이터를 조회할 준비를 하는 과정입니다.vectorstore.as_retriever()를 통해 검색 도구(retriever)를 만듭니다. 이 리트리버는 주어진 질의(여기서는 사용자의 증상)와 가장 유사한 문서 청크들을 벡터스토어에서 찾아 반환하는 역할을 합니다.rag_prompt는 LLM에게 단순한 질의응답을 넘어서, 제공된 symptoms와 검색된 context (문서 내용)를 바탕으로 정보를 "요약"하도록 지시합니다. 이는 LLM이 관련 정보를 단순히 반복하는 것이 아니라, 이해하고 재구성하여 의미 있는 답변을 만들도록 유도합니다.
create_stuff_documents_chain은 여러 검색된 문서를 LLM의 컨텍스트 윈도우에 효율적으로 맞추어 전달하는 표준적인 방법(stuff방식)을 사용합니다.retriever.get_relevant_documents(state["symptoms"])를 통해, 사용자의 현재 증상(state["symptoms"])과 관련된 문서들을 벡터스토어에서 실시간으로 검색합니다.
8. SQLite 데이터베이스에서 질병 > 대처 방안을 조회하는 sql_agent 노드를 정의
# DB에서 조회하는 노드
def sql_agent(state: AgentState): # DB에서 질병 대처방안 조회
# sql 파일 읽음
conn = sqlite3.connect("disease_knowledge.db")
cursor = conn.cursor()
# matcher_agent 함수에서 아까 llm이 질병 후로 내뱉지 않았나? 그걸 state에서 가져와서 콤마 기준으로 파싱해서
# 해당 질병이 디비에 있는지 where절로 치유 방법 조회 할거임.
names = [n.strip() for n in state["disease_candidates"].split(",")]
results = []
for name in names:
cursor.execute("SELECT care FROM diseases WHERE name = ?", (name,))
row = cursor.fetchone()
# 질병마다 치유방법 있는지 없는지 가져와서 빈 리스트에 넣음.
if row:
results.append(f"{name} 대처법: {row[0]}")
else:
results.append(f"{name}: 대처 정보 없음")
conn.close()
# 질병 정보에다가 리스트를 join함수로 해서 하나의 구절로 다시 대입
# 증상과 질병 정보를 보아하니 당시은 즉각 병원에 가봐야 할 것 같아요.
# + "감기:잠 푹자기\n코로나:백신처방"
return {**state, "disease_info": state["disease_info"] + "\n" + "\n".join(results)}
# {query:"어쩌구",
# symptoms:"기침, 목, 열",
# disease_candidates: "감기,코로나,식중독",
# disease_info: "증상과 질병 정보를 보아하니 당신은 즉각 병원에 가봐야 할 것 같아요\n감기:잠 푹자기\n코로나:백신처방",
# result: null }
disease_knowledge.db에 연결하고 SQL 쿼리 실행을 위한 커서를 얻습니다.matcher_agent)에서 LLM이 추론한 쉼표로 구분된 질병 후보 문자열(state["disease_candidates"])을 개별 질병 이름 리스트로 분리합니다.care (대처법) 정보를 조회합니다. INSERT OR IGNORE를 사용했던 build_sqlite_db처럼, 여기서는 WHERE name = ?를 사용하여 정확한 질병 이름을 기준으로 조회합니다.results 리스트에 "질병명 대처법: [내용]" 또는 "질병명: 대처 정보 없음" 형식으로 저장합니다.
results 리스트를 줄바꿈 문자로 연결하여, 기존 disease_info (RAG 결과) 뒤에 추가합니다. 이로써 disease_info 필드는 RAG를 통한 일반적인 질병 설명과 DB에서 가져온 구체적인 대처법 정보를 모두 포함하게 됩니다.9. 모든 정보를 취합하여 최종 답변을 생성하는 노드와, 정의된 노드들을 LangGraph로 연결하여 전체 에이전트 워크플로우를 구성하고 실행
# 여태까지 얻은 정보로 마지막 노드에서 최종 질의 만드는 곳임. 이 부분이 최종 대답을 얻기 위한 과정임.
# state에 저장된 키:밸류 중 원하는 요소만 쏙쏙 빼서 질의에 넣으면 됨
# 아래는 state에서 sympthms랑, disease, candidates, disease_info의 최종 결과만 사용하네?
answer_prompt = PromptTemplate.from_template("""
사용자의 증상은 다음과 같습니다: {symptoms}
예측된 질병 후보: {disease_candidates}
관련 정보 및 대처방안:
{disease_info}
위 내용을 바탕으로 사용자에게 알기 쉽게 설명해주세요.
""") # 최종 응답 생성 프롬프트
def answer_agent(state: AgentState): # 응답 생성 에이전트
chain = answer_prompt | llm # 프롬프트와 LLM을 연결하여 실행 체인 구성
answer = chain.invoke({
"symptoms": state["symptoms"],
"disease_candidates": state["disease_candidates"],
"disease_info": state["disease_info"]
})
return {**state, "result": answer.strip()}
# 5. LangGraph 정의
from langgraph.graph import StateGraph # LangGraph 구성 요소
graph = StateGraph(AgentState) # 그래프 정의
graph.add_node("extractor", extractor_agent) # 노드 추가
graph.add_node("matcher", matcher_agent)
graph.add_node("rag", rag_agent)
graph.add_node("sql_lookup", sql_agent)
graph.add_node("answer", answer_agent)
graph.set_entry_point("extractor") # 시작 노드 설정
graph.add_edge("extractor", "matcher") # 노드 간 연결 정의
graph.add_edge("matcher", "rag")
graph.add_edge("rag", "sql_lookup")
graph.add_edge("sql_lookup", "answer")
graph.add_edge("answer", END) # 종료 노드 설정
app = graph.compile() # 그래프 컴파일
print("============================== LangGraph 구조:")
app.get_graph().print_ascii() # 구조 출력
# 6. 실행 예시
query = "기침이 심하고 목이 아프고 열이 납니다" # 사용자 질문
result = app.invoke({"query": query}) # 실행
print("============================== 최종 응답:")
print(result["result"]) # 결과 출력
answer_agent 함수:
- 이 함수는 에이전트의 최종 목표인 "사용자에게 의미 있는 답변 제공"을 담당합니다.
- 이전 노드들(
extractor,matcher,rag,sql_agent)에서AgentState에 차곡차곡 쌓아 올린symptoms,disease_candidates,disease_info를 한데 모아answer_prompt에 채워 넣습니다. - LLM은 이 풍부한 컨텍스트를 바탕으로 최종적으로 사용자에게 친절하고 유익한 설명을 생성하게 됩니다.
return {**state, "result": answer.strip()}를 통해 최종 답변을AgentState의result필드에 저장합니다.
LangGraph 정의 및 구성:
StateGraph(AgentState)는AgentState를 상태의 스키마로 사용하여 그래프를 초기화합니다.graph.add_node()를 통해 앞서 정의된 모든 에이전트 함수(extractor_agent,matcher_agent,rag_agent,sql_agent,answer_agent)를 개별적인 처리 노드로 그래프에 추가합니다.graph.set_entry_point("extractor")는 워크플로우가 "extractor" 노드부터 시작됨을 명시합니다.graph.add_edge()는 각 노드의 실행 순서를 정의합니다.extractor->matcher->rag->sql_lookup->answer의 선형적인 흐름을 만듭니다.graph.add_edge("answer", END)는 "answer" 노드 실행 후에 그래프가 종료됨을 나타냅니다.app = graph.compile()은 정의된 그래프를 실행 가능한 형태로 컴파일하며, 이를 통해app.invoke()와 같은 메서드를 사용하여 에이전트를 호출할 수 있게 됩니다.app.get_graph().print_ascii()는 그래프의 논리적 흐름을 시각적으로 확인할 수 있게 해주는 유용한 디버깅 도구입니다.
실행 예시:
query = "기침이 심하고 목이 아프고 열이 납니다"와 같은 실제 사용자 질문을app.invoke({"query": query})를 통해 에이전트에 전달합니다.app.invoke()는extractor노드부터 시작하여add_edge로 정의된 순서대로 각 노드를 실행하고AgentState를 업데이트하며 전달합니다.- 최종적으로
result["result"]에는answer_agent가 생성한 통합된 답변이 담기게 되며, 이를 출력하여 사용자에게 보여줍니다.
10. 결과
============================== 최종 응답:
### 사용자 증상 및 가능한 질병 요약 및 대처 방안
**증상 요약:**
- **기침**
- **목 아픔**
- **발열**
**가능한 질병 후보 및 특징:**
1. **감기 (Common Cold):**
- **주요 증상:** 기침, 목 아픔, 콧물이 주로 동반됩니다.
- **발열:** 경미하거나 거의 없을 수 있습니다.
- **특징:** 비교적 가볍고 자주 발생하는 호흡기 감염입니다.
- **대처방안:**
- **수분 섭취:** 충분한 수분 섭취로 탈수를 예방하세요.
- **휴식:** 충분한 휴식을 취하고 면역력을 높이는 방법을 고려하세요.
- **해열제:** 필요하다면 해열제를 복용하여 불편함을 완화하세요.
2. **독감 (Influenza):**
- **주요 증상:** 기침, 발열(38°C 이상), 심한 목 아픔, 피로감이 일반적입니다.
- **특징:** 감기보다 더 심각하고 빠른 진행 속도를 보일 수 있습니다.
- **대처방안:**
- **의료 상담:** 의사와 상담하여 적절한 치료법(예: 항바이러스제 복용)을 결정하세요.
- **휴식 및 수분 섭취:** 충분한 휴식을 취하고 수분을 충분히 섭취하세요.
...
- **휴식 및 수분 섭취:** 모든 경우에 있어 충분한 휴식과 수분 섭취는 회복에 도움이 됩니다.
- **증상 모니터링:** 증상이 지속되거나 악화될 경우 즉시 의료 서비스를 이용하세요.
**중요:** 위 정보는 일반적인 가이드라인이며, 개인의 건강 상태에 따라 적합한 조치가 달라질 수 있습니다. 특히 의심 증상이 있는 경우 전문가의 진단을 받는 것이 가장 안전하고 효과적인 방법입니다.
Tags:
클라우드
