데이터 수집 DART → 데이터 저장 로컬 → embedding, vector db kb→ langchain → llm
from langchain.schema import Document
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings.openai import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.llms import OpenAI
from langchain.chains import RetrievalQA
Python
복사
url_for_company = "https://opendart.fss.or.kr/api/list.json" # xml도 가능함. json이 편해서 이걸로 봤음.
params = {
"crtfc_key": dart_api_key,
"corp_code": "00126380",
"bgn_de": "20240101",
"end_de": "20250630",
"pblntf_detail_ty": "A001" # 사업보고서 조회 시 이 파라미터를 반드시 붙여줘야 함.
}
res = requests.get(url, params=params).json()
print(json.dumps(res, indent=2, ensure_ascii=False))
"""
{
"status": "000",
"message": "정상",
"page_no": 1,
"page_count": 10,
"total_count": 2,
"total_page": 1,
"list": [
{
"corp_code": "00126380",
"corp_name": "삼성전자",
"stock_code": "005930",
"corp_cls": "Y",
"report_nm": "사업보고서 (2024.12)",
"rcept_no": "20250311001085", # 여기에서 나온 문서번호를 rcept_no에 입력해야 함.
"flr_nm": "삼성전자",
"rcept_dt": "20250311",
"rm": "연"
},
{
"""
############
rcept_no = "20250311001085" # 문서 번호 별도 조회
# url_for_company에서 company code를 기반으로 검색해야함. 그러면 회사와 문서 이름 문서 번호가 나옴
url = "https://opendart.fss.or.kr/api/document.xml"
params = {
"crtfc_key": dart_api_key,
"rcept_no": rcept_no
}
response = requests.get(url, params=params)
# soup = BeautifulSoup(response.content, "html.parser", from_encoding="utf-8") # Content Type이 대체로 xml, msdownload 일 경우 압축 해제 처리를 해줘야 함
if "xml" in response.headers.get("Content-Type", "").lower():
soup = BeautifulSoup(response.content, "xml")
text = soup.get_text("\n", strip=True)
elif "msdownload" in response.headers.get("Content-Type", "").lower() or \
"zip" in response.headers.get("Content-Type", "").lower():
with zipfile.ZipFile(io.BytesIO(response.content)) as z:
print("ZIP 내부 파일:", z.namelist())
xml_name = z.namelist()[0]
xml_data = z.read(xml_name)
soup = BeautifulSoup(xml_data, "xml")
text = soup.get_text("\n", strip=True)
else:
text = None
print(text[:1000])
"""
"""
encoding = tiktoken.encoding_for_model("gpt-4o")
tokens = encoding.encode(text)
print(tokens) # 56517 # 190원
Python
복사
samsung.txt
ext_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000, # 한 덩어리당 토큰 수
chunk_overlap=100, # 겹치는 부분
length_function=len,
separators=["\n\n", "\n", " ", ""]
)
split_docs = text_splitter.split_documents(docs)
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(split_docs, embeddings)
llm = OpenAI(temperature=0.7)
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff",
retriever=vectorstore.as_retriever(search_kwargs={"k": 3}),
return_source_documents=True
)
Python
복사
test_questions = [
"삼성전자의 최근 재무상황은 어떤가요?",
"회사의 주요 사업 내용을 요약해주세요",
"최근 공시에서 주목할 만한 내용이 있나요?"
]
for question in test_questions:
result = qa_chain({"query": question})
print(f"\n질문: {question}")
print(f"답변: {result['result']}\n")
Python
복사
my_question = input("여기에 질문을 입력하세요:")
result = qa_chain({"query": my_question})
print(result['result'])
Python
복사
