반응형
[Langchain] 계엄령 기념, 집밥 같은 랭체인 코드로 계엄령 뉴스 보기
계엄령 기념으로 계엄령 뉴스들을 몇개 선택한 후
마치 집밥을 먹는것 같은!
마치 기본적인 반찬들이 있는 것 같은!
랭체인 기본 코드를 작성했다.
Code
import os
from typing import List
import bs4
from dotenv import load_dotenv
from langchain.embeddings import CacheBackedEmbeddings
from langchain.retrievers import EnsembleRetriever
from langchain.storage import LocalFileStore
from langchain_chroma import Chroma
from langchain_community.document_loaders import WebBaseLoader
from langchain_community.retrievers import BM25Retriever
from langchain_core.documents import Document
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_groq import ChatGroq
from langchain_ollama import OllamaEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_demos.utils.cache_loader import LocalCacheLoader
from langchain_demos.utils.decorators import cacheable
from langchain_demos.utils.dev import green
DOCUMENT_CACHE_PATH = "coupdeta"
CHROMA_PATH = ".vector.coupdeta"
STORE_PATH = ".store.coupdeta"
load_dotenv()
@cacheable(DOCUMENT_CACHE_PATH, loader=LocalCacheLoader)
def load_documents() -> List[Document]:
recursive_text_splitter = RecursiveCharacterTextSplitter(
chunk_size=100,
chunk_overlap=30,
length_function=len,
is_separator_regex=False,
separators=["\n\n"],
)
loader = WebBaseLoader(
web_paths=(
"https://n.news.naver.com/mnews/article/050/0000082996",
"https://n.news.naver.com/mnews/article/050/0000082989",
"https://n.news.naver.com/mnews/article/014/0005277179",
"https://n.news.naver.com/mnews/article/050/0000082998",
"https://n.news.naver.com/mnews/article/050/0000082873",
"https://n.news.naver.com/mnews/article/014/0005277164",
"https://n.news.naver.com/mnews/article/014/0005277881",
),
bs_kwargs=dict(
parse_only=bs4.SoupStrainer("article", attrs={"id": ["dic_area"]}),
),
)
return loader.load_and_split(recursive_text_splitter)
def get_vector_db() -> Chroma:
embeddings = OllamaEmbeddings(
# model="mxbai-embed-large"
model="bge-m3",
)
cached_embeddings = CacheBackedEmbeddings.from_bytes_store(
underlying_embeddings=embeddings,
document_embedding_cache=LocalFileStore(STORE_PATH),
namespace=embeddings.model,
)
if os.path.exists(DOCUMENT_CACHE_PATH):
return Chroma(
embedding_function=cached_embeddings,
persist_directory=CHROMA_PATH,
)
return Chroma.from_documents(
documents=load_documents(),
embedding=cached_embeddings,
persist_directory=CHROMA_PATH,
)
def get_retriever() -> EnsembleRetriever:
vector_db = get_vector_db()
bm25_retriever = BM25Retriever.from_documents(
load_documents(),
search_kwargs={
"k": 5,
},
)
vector_retriever = vector_db.as_retriever(
search_type="mmr",
search_kwargs={
"k": 5,
# "fetch_k": 5, # MMR 초기 후보군 크기
# "lambda_mult": 0.8, # 다양성, 관련성 사이의 균형 조정 (0.5-0.8 범위 추천)
# "score_threshold": 0.8, # 유사도 점수가 0.8 이상인 문서만 반환
# "filter": {"metadata_field": "desired_value"}, # 메타 데이터 기반 필터링
},
)
return EnsembleRetriever(
retrievers=[bm25_retriever, vector_retriever],
weights=[0.6, 0.4],
)
def app_main():
llm = ChatGroq(
model="llama-3.1-70b-versatile",
temperature=0,
max_tokens=None,
max_retries=2,
timeout=None,
)
retriever = get_retriever()
prompt = ChatPromptTemplate.from_messages(
[
(
"system",
"""
# Instruction:
다음은 뉴스 데이터에서 추출한 텍스트입니다. 아래의 지침을 따라 요약하세요.
1. 텍스트에서만 제공된 정보를 기반으로 요약하세요. 외부의 추가 정보나 상식을 포함하지 마세요.
2. 내용은 객관적이고 중립적인 시각에서 작성하세요.
3. 뉴스에 언급된 숫자, 통계, 인용문 등 중요한 세부 정보를 반드시 포함하세요.
4. 주제와 관련 없는 세부 사항이나 부차적인 정보를 제거하고 핵심적인 내용을 간결하게 전달하세요.
5. 뉴스의 출처가 명확하지 않은 정보나 추측성 내용은 요약에 포함하지 마세요.
# News Article:
{articles}
# Response Format:
"전체 내용 요약 (50단어 이내)"
- "핵심 내용 1 (20단어 이내)" (source)
- "핵심 내용 2 (20단어 이내)" (source)
- "핵심 내용 3 (20단어 이내)" (source)
- ...
""".strip(),
),
("human", "# Question: {question}"),
]
)
chain = {"articles": retriever, "question": RunnablePassthrough()} | prompt | llm | StrOutputParser()
while True:
user_input = input("\n유저 입력: ")
for found in retriever.invoke(user_input):
green(found.page_content)
for token in chain.stream(user_input):
print(token, end="", flush=True)
if __name__ == '__main__':
app_main()Output
유저 입력: >? 계엄령 군인
"윤석열 대통령의 비상계엄 선포와 관련된 뉴스 요약"
- "계엄군 청년이 시민에게 고개를 숙여 사과하는 모습이 포착됐다." (https://n.news.naver.com/mnews/article/050/0000082989)
- "계엄군은 국회 본관 건물 진입을 시도했지만 곧이어 국회에서 비상계엄 해제 요구 결의안이 통과되면서 철수를 시작했다." (https://n.news.naver.com/mnews/article/050/0000082989)
- "계엄사령관 박안수 육군참모총장은 계엄사령부 포고령 제1호를 통해 정치활동, 집회, 시위 등을 금지했다." (https://n.news.naver.com/mnews/article/050/0000082873)
유저 입력: >? 계엄령 실패
"윤석열 대통령의 비상계엄령 선포와 해제에 따른 혼란과 논란"
- "윤석열 대통령, 44년 만에 비상계엄령 선포" "(https://n.news.naver.com/mnews/article/050/0000082996)"
- "계엄령 선포 후 재난문자 발송되지 않아 시민 혼란 가중" "(https://n.news.naver.com/mnews/article/050/0000082996)"
- "한덕수 국무총리, 비상계엄 사태와 관련해 내각 총사퇴 논의" "(https://n.news.naver.com/mnews/article/014/0005277164)"
- "김문수 고용노동부 장관, 비상계엄령 위헌 여부 판단하지 않았다고 밝혀" "(https://n.news.naver.com/mnews/article/014/0005277881)"Cache Util
import functools
from typing import Callable, Type
from langchain_demos.utils.cache_loader import LocalCacheLoader, CacheLoader
def cacheable(key: str, loader: Type[CacheLoader] = LocalCacheLoader):
def decorator(func: Callable):
@functools.wraps(func)
def wrapper(*args, **kwargs):
cache_loader = loader(key)
if cache_loader.is_cached():
return cache_loader.read()
cache_loader.write(result := func(*args, **kwargs))
return result
return wrapper
return decorator
import abc
import os
import pickle
from pathlib import Path
from typing import Any
import redis
class CacheLoader(abc.ABC):
def __init__(self, key: str):
self.key = key
@abc.abstractmethod
def is_cached(self) -> bool:
raise NotImplementedError()
@abc.abstractmethod
def read(self) -> Any:
raise NotImplementedError()
@abc.abstractmethod
def write(self, dataset: Any):
raise NotImplementedError()
class LocalCacheLoader(CacheLoader):
def __init__(self, key: str):
super().__init__(key)
def is_cached(self) -> bool:
return Path(self.key).exists()
def read(self) -> Any:
with open(self.key, 'rb') as file:
return pickle.load(file)
def write(self, dataset: Any):
with open(self.key, 'wb') as file:
pickle.dump(dataset, file)
class RedisCacheLoader(CacheLoader):
def __init__(self, key: str):
super().__init__(key)
self.client = redis.StrictRedis(
host=os.getenv("REDIS_HOST", "localhost"),
port=os.getenv("REDIS_PORT", 6379),
db=os.getenv("REDIS_DATABASE", 7),
)
self.expire_ttl = os.getenv("REDIS_EXPIRE_TTL", 60 * 60)
def is_cached(self) -> bool:
return self.client.exists(self.key)
def read(self) -> Any:
return pickle.loads(
self.client.get(self.key),
)
def write(self, dataset: Any):
self.client.set(
self.key,
pickle.dumps(dataset),
ex=self.expire_ttl,
)728x90
반응형