2024년 3월 31일: 든든전세주택 시각화 프로젝트 업데이트
오늘은 기존에 진행 중이던 든든전세주택 시각화 프로젝트에 새로운 데이터가 업데이트되어 이를 반영하는 작업을 시작했다. 최근 데이터 양이 크게 증가하면서 웹 크롤링 과정에서 예상보다 많은 시간이 소요되는 문제에 직면했다.
크롤링 최적화 고려사항
초기에는 비동기 처리를 통한 효율화를 고려했으나, 동일 사이트에 대한 크롤링이라는 특성을 고려하여 배치 단위 처리 방식으로 전환하기로 결정했다. 이는 서버에 과도한 부하를 주지 않으면서도 효율적인 데이터 수집이 가능한 접근법이다.
GitHub 프로젝트에서 이 문제를 해결하기 위해 다음과 같은 배치 처리 함수를 구현했습니다:
# 배치 처리 방식으로 이미지 링크 수집
def get_all_img_links_batch(href_ids, dt='20241007', batch_size=5):
results = []
for i in tqdm(range(0, len(href_ids), batch_size)):
batch = href_ids[i:i + batch_size]
for href_id in batch:
img_link = get_img_link(href_id, dt)
results.append(img_link)
# 배치 처리 후 잠시 대기
time.sleep(1.5)
return results
이 함수는 웹사이트에 과도한 요청을 방지하기 위해 배치 단위로 처리하고 각 배치 사이에 적절한 대기 시간을 두는 방식을 구현했습니다.
시각화 방식 업데이트
기존 코드 중 일부가 최신 라이브러리 버전과 호환되지 않는 문제가 발견되어 시각화 방식을 변경하기로 했다. 더불어 LLM 멀티모달 기능을 활용하여 구조도 이미지에서 방 개수와 같은 정보를 자동으로 추출하는 기능을 추가하기로 계획했다.
모델 테스트 결과
Gemini-2.5, 2.0 Flash, Flash Lite 모델을 테스트해본 결과, 일부 정보 추출에서 정확도가 떨어지는 경우가 발견되었다. 이를 개선하기 위해 두 가지 접근법을 고려 중이다:
- 프롬프트 엔지니어링을 통한 정확도 향상
- RPM(Request Per Minute) 제한을 고려한 루프 시간 조정
2024년 4월 1일: 데이터 기반 제안서 작성 방법론 개발
오늘은 제안서 작업 시 효율적인 데이터 수집 및 정리 방법론을 개발했다.
LLM을 활용한 리서치 프로세스
- 패턴 딥리서치 접근법: Gemini와 ChatGPT를 활용하여 자료를 상세히 조사한 후, 해당 내용을 요구사항에 맞게 명사형 종결 형식으로 재구성하는 방식을 개발했다.
- 언어 모델 특성화 활용: 한국어 글쓰기 능력이 비교적 우수한 Claude를 활용하여 최종 내용 재구성 작업을 진행했다.
효율적인 문서 구조화 방법
전체적인 흐름을 먼저 설정하는 것이 중요하다는 점을 발견했다. Napkin AI를 활용하여 프로젝트 흐름도를 먼저 작성한 후, 이를 컨텍스트로 제공하여 LLM에게 내용 재구성을 요청했을 때 더 우수한 결과를 얻을 수 있었다.
이력서 최적화 프로세스
MCP(멀티모달 기능)를 최대한 활용하여 개인 이력서를 효과적으로 수정하는 방법을 개발했다:
- YouTube MCP 기능을 활용해 경력기술서 작성 관련 영상 내용을 참고
- 개인 GitHub 내용을 정리하여 이력서 PDF와 함께 분석
- 구조적 구성을 위해 마크다운 형식으로 변환하거나 Notion에 정리 후 PDF로 변환
2024년 4월 2-3일: 든든전세 대시보드 업그레이드와 LangGraph 구현
LangGraph를 활용한 구조도 정보 추출
기존에 계획했던 방 구조도에서 정보를 추출하는 기능을 LangGraph를 활용하여 구현했다. GitHub 프로젝트에서는 다음과 같이 구현되어 있습니다:
from langgraph.graph import StateGraph, START, END
from .nodes import download_image, save_image, describe_image, check_image_description, get_default_llm
from .edges import decide_to_generate, decide_to_regenerate
from .models import GraphState
from langchain_core.language_models.chat_models import BaseChatModel
from typing import Optional, Dict, List, Any, TypedDict
# 병렬 노드 반환 타입 정의
class SaveImageBranch(TypedDict):
"""이미지 저장 작업의 병렬 분기 결과"""
save_result: None # 이미지 저장 결과 (상태를 변경하지 않음)
def make_workflow(llm: Optional[BaseChatModel] = None) -> StateGraph:
"""
LLM을 인자로 받아 방 구조 분석 워크플로우를 생성하는 함수
이미지 다운로드 후 저장은 병렬로 진행하고, 이미지 분석은 메인 흐름에서 진행
Args:
llm (BaseChatModel, optional): 사용할 LLM 모델. 기본값은 None이며,
None인 경우 기본 모델 사용
Returns:
StateGraph: 컴파일된 LangGraph 워크플로우
"""
# llm이 None인 경우 기본 모델 설정
llm_instance = llm or get_default_llm()
# Define a new graph
workflow = StateGraph(GraphState)
# Define the nodes
workflow.add_node("download_image", download_image)
# 이미지 저장 병렬 노드: 상태를 변경하지 않고 별도 작업만 수행
def save_image_branch(state: GraphState) -> Dict[str, Dict[str, Any]]:
"""다운로드된 이미지를 저장하는 병렬 노드"""
# 이미지 저장 실행
save_image(state)
# 상태 변경 없이 병렬 노드 결과 반환
return {"save_result": None}
# 병렬 노드 추가: 다운로드 이미지를 계속 사용하면서 저장도 병렬로 수행
workflow.add_node("save_image_branch", save_image_branch)
# LLM을 사용하는 노드들에 llm_instance 전달
workflow.add_node("describe_image", lambda state: describe_image(state, llm=llm_instance))
workflow.add_node("check_image_description", lambda state: check_image_description(state, llm=llm_instance))
# Add edges
workflow.add_edge(START, "download_image")
# 이미지 다운로드 후 병렬 처리
# 1. 이미지 저장 브랜치
workflow.add_edge("download_image", "save_image_branch")
# 이미지 저장 브랜치는 결과를 반환하지만 메인 워크플로우에 영향을 주지 않음
# 2. 메인 경로: 다운로드 → 이미지 설명
workflow.add_conditional_edges(
"download_image",
decide_to_generate,
{
"end": END,
"generate": "describe_image"
}
)
workflow.add_edge("describe_image", "check_image_description")
workflow.add_conditional_edges(
"check_image_description",
decide_to_regenerate,
{
"nextstep": END,
"regenerate": "describe_image",
"end": END
}
)
# Compile
graph = workflow.compile()
return graph
이 코드는 이미지 다운로드, 저장, 분석, 검증의 전체 워크플로우를 정의하고 있습니다. 특히 병렬 처리를 통해 이미지 처리 효율성을 높이고, 조건부 엣지를 통해 처리 결과에 따라 다른 경로로 진행하는 흐름 제어를 구현했습니다.
기술적 제약사항 및 해결 방안
토큰 수 제한을 고려했을 때, 구조도를 직접 전송하는 멀티모달 방식은 OpenAI 모델 사용에 제약이 있다고 판단했다. 이에 따라 RPM(Request Per Minute) 제한을 고려한 시스템 구조를 다음과 같이 설계했다:
- 구조도 다운로드
- 병렬 처리:
- 구조도 로컬 저장
- 구조도에서 정보 추출
- LLM을 활용한 추출 정보 검증
- 검증 결과에 따른 반복 처리 여부 결정
방 구조도 이미지 분석 시 정확도를 높이기 위해 검증 단계를 추가했는데, 이는 GitHub의 edges.py 파일에서 다음과 같이 구현되어 있습니다:
def decide_to_regenerate(state: GraphState) -> Literal["nextstep", "regenerate", "end"]:
"""
Determines whether to regenerate an answer, proceed to next step, or end the workflow.
Args:
state (GraphState): The current graph state
Returns:
str: Decision for next node to call
"""
# 현재 regenerate_count 출력 (디버깅용)
current_count = state.get('regenerate_count', 0)
logger.info(f"현재 재생성 횟수: {current_count}")
# 재시도 최대 횟수 정의
max_regenerations = 3
# 최대 재시도 횟수 초과 여부 확인 (어떤 조건에서든 우선 확인)
if current_count >= max_regenerations:
logger.error(f"최대 재시도 횟수({max_regenerations})에 도달했습니다. 워크플로우를 종료합니다.")
return "end"
# 에러 여부 확인
if state.get('error'):
logger.warning(f"에러 발생: {state['error']}")
return "regenerate"
# 검증 결과 확인
accuracy = state.get('description_accuracy')
if not accuracy:
logger.warning("검증 결과가 없습니다. 재생성합니다.")
return "regenerate"
# 모든 검증 통과 확인
if (accuracy.num_room_accuracy and
accuracy.num_balcony_accuracy and
accuracy.num_wc_accuracy and
accuracy.description_adequacy):
logger.info("모든 검증이 통과되었습니다.")
return "nextstep"
# 검증 실패로 재생성 진행
logger.warning(f"검증 실패. 재생성합니다. (시도 {current_count + 1}/{max_regenerations})")
return "regenerate"
이 코드는 방 구조 분석 결과의 정확도를 검증하고, 필요에 따라 재분석을 결정하는 로직입니다. 최대 3번까지 재시도를 허용하며, 방의 개수, 발코니 수, 화장실 수, 그리고 전반적인 설명의 적절성을 모두 검증합니다.
현재 진행 상황과 과제
구조도 다운로드 과정이 안정적인 API가 아닌 크롤링 방식에 의존하고 있어, 웹사이트에 과도한 부하를 주지 않도록 주의하면서 작업을 진행하고 있다. 또한 응답 속도 문제로 인해 데이터 처리에 상당한 시간이 소요되고 있다.
이미지 다운로드 부분은 GitHub 프로젝트에서 다음과 같이 구현되어 있습니다:
def download_image(img_url, save_path, filename, max_retries=3, timeout=30):
for attempt in range(max_retries):
try:
# SSL 검증 비활성화 및 타임아웃 설정
response = requests.get(img_url, verify=False, timeout=timeout)
response.raise_for_status()
os.makedirs(save_path, exist_ok=True)
file_path = os.path.join(save_path, filename)
with open(file_path, 'wb') as f:
f.write(response.content)
return True
except requests.Timeout:
if attempt < max_retries - 1:
print(f"타임아웃 발생 ({filename}), {attempt + 1}/{max_retries} 재시도")
time.sleep(2) # 타임아웃 발생 시 더 긴 대기
else:
print(f"최대 타임아웃 재시도 횟수 초과 ({filename})")
return False
except Exception as e:
if attempt < max_retries - 1:
print(f"다운로드 시도 {attempt + 1}/{max_retries} 실패: {e}")
time.sleep(1)
else:
print(f"최대 재시도 횟수 초과 ({filename}): {e}")
return False
def download_images_batch(df, batch_size=2, save_path='downloaded_images', max_retries=3,
timeout=30, delay_between_batches=3):
downloaded_paths = []
# 이미지 다운로드
for i in tqdm(range(0, df.shape[0], batch_size), desc="이미지 다운로드 중"):
batch = df.loc[i:i + batch_size, :]
for row in batch.itertuples():
img_url = row.img
if img_url:
filename = f"{row.번호}.jpg"
if download_image(img_url, save_path, filename, max_retries=max_retries, timeout=timeout):
downloaded_paths.append(os.path.join(save_path, filename))
else:
print(f"이미지 다운로드 실패: {filename}")
# 배치 처리 후 대기
time.sleep(delay_between_batches)
return downloaded_paths
이 코드는 이미지 다운로드를 배치 단위로 처리하며, 재시도 로직과 타임아웃 처리를 포함하여 견고한 다운로드 프로세스를 구현했습니다.
최근 대상 매물 수가 100개에서 500개로 크게 증가함에 따라, 현재는 코드 정리와 테스트에 집중하고 있으며, 전체 시스템의 스케일링 방안을 검토 중이다.
지하철 역 거리 및 통근시간 계산 기능
GitHub 프로젝트에서는 카카오 API를 활용하여 각 매물과 가장 가까운 지하철역까지의 거리와 예상 통근 시간을 계산하는 기능도 구현되어 있습니다:
def find_near_subway_station(x, y, max_distance = 3000):
time.sleep(0.2)
results = kakaomap.search_by_category('SW8', y, x, 3000) # 위경도 바꿔어서 입력
if len(results.get('documents'))!=0:
near_result = results.get('documents')[0]
return near_result.get('distance'), near_result.get('place_name')
else:
print('no result')
def calculate_transit_time(self, origin_y, origin_x, dest_y, dest_x):
time.sleep(0.2)
url = "https://apis-navi.kakaomobility.com/v1/directions"
headers = {"Authorization": f"KakaoAK {self.api_token}"}
params = {
"origin": f"{origin_y},{origin_x}",
"destination": f"{dest_y},{dest_x}",
"priority": "RECOMMEND",
"car_fuel": "GASOLINE",
"car_hipass": True,
"alternatives": False,
"road_details": False,
"roadevent":2
}
try:
response = requests.get(url, headers=headers, params=params)
if response.status_code == 200:
result = response.json()
return result['routes'][0]['summary']['duration'] / 60, result['routes'][0]['summary']['distance']
return None
except:
return None
이 코드들은 카카오맵 API를 활용하여 지리적 정보를 풍부하게 하는 기능을 구현했습니다. 각 매물로부터 가장 가까운 지하철역을 찾고, 특정 목적지(예: 회사)까지의 예상 통근 시간을 계산하여 사용자가 주택 선택 시 중요한 정보를 제공합니다.
다음 단계
향후 계획은 다음과 같다:
- 크롤링 프로세스 최적화를 통한 데이터 수집 시간 단축
- LLM 기반 정보 추출의 정확도 향상을 위한 프롬프트 엔지니어링
- 대량 데이터 처리를 위한 분산 처리 시스템 구축 검토
- 시각화 대시보드의 사용자 인터페이스 개선
이번 연구를 통해 LLM과 데이터 파이프라인을 결합한 효율적인 정보 추출 및 시각화 시스템의 가능성을 확인할 수 있었으며, 추후 더 많은 도메인에 적용할 수 있는 방안을 모색할 예정이다.
'사이드 프로젝트 연구일지' 카테고리의 다른 글
| 연구 일지: MCP 서버를 활용한 Tmap API 연동 및 멀티모달 AI 활용 연구(250407~250411) (1) | 2025.04.14 |
|---|---|
| 프롬프트 엔지니어링 / RAG 연구일지: MCP 탐색(feat. cursor, claude) (0) | 2025.03.31 |
| 프롬프트 엔지니어링 / RAG 연구일지 (0) | 2025.03.25 |