종합 가이드 - 목차
1. 개요 및 주요 기능
2. 환경별 상세 실행 모드
2.1 Mac OS 환경 (명함처리용)
2.2 리눅스 서버 환경 (대량 처리용)
3. 모델 선정 과정과 성능 평가
3.1 평가된 주요 모델들
3.2 최적화 시도 및 결과
4. 코드 동작 원리 및 주요 구성요소
4.1 URL 정보 추출 프로세스
4.2 병렬 처리 아키텍처
4.3 모델 추론 및 프롬프트 엔지니어링
5. 로깅 및 진행 상황 추적
6. 명령줄 인터페이스 및 설정 옵션
1. 개요 및 주요 기능
회사 웹사이트 URL에서 메타데이터와 핵심 정보를 추출하고, 고성능 언어 모델을 활용하여 간결한 영문 회사 소개를 자동으로 생성하는 고급 데이터 처리 파이프라인입니다. 이 스크립트는 2025년 4월 9일 최신 버전(v5.0)으로, 다양한 컴퓨팅 환경과 사용 사례를 지원하도록 설계되었습니다.
주요 기능으로는:
-
웹사이트 데이터 수집: 회사 URL에서 메타데이터(제목, 설명), 메인 페이지 콘텐츠, 주요 카테고리별 콘텐츠(예: About, Products, Services 등)를 지능적으로 추출합니다.
-
언어 모델 추론: 추출된 정보를 바탕으로 최신 LLM(Large Language Model)을 활용해 2-3문장의 간결하고 전문적인 회사 소개문을 생성합니다.
-
대규모 병렬 처리: 멀티 GPU, 멀티스레드 아키텍처를 통해 수만 개의 URL을 효율적으로 처리할 수 있습니다.
-
자동 리소스 최적화: 실행 환경(macOS 또는 Linux)과 가용 하드웨어(CPU, MPS, CUDA GPU)를 자동으로 감지하여 최적의 설정을 적용합니다.
-
체크포인트 및 복구 시스템: 장시간 실행 중 중단되더라도 진행 상황을 유지하고 처리를 재개할 수 있는 견고한 체크포인트 시스템을 갖추고 있습니다.
2. 환경별 상세 실행 모드
2.1 Mac OS 환경 (명함처리용)
macOS 환경에서는 Apple Silicon 프로세서의 Metal 가속을 활용하여 효율적인 처리가 가능합니다:
-
하드웨어 가속: Apple MPS(Metal Performance Shaders) 자동 감지 및 활용
-
사용 모델: LG ExaOne 3.5:32b
-
모델 특징: 한국어와 영어에 특화된 32B 파라미터 모델로, 한국어 회사명과 산업 용어에 대한 이해도가 뛰어남
-
실행 엔진: Ollama를 통한 로컬 실행 (메모리 효율성 최적화)
-
배치 크기: 8 (MPS 메모리 한계를 고려한 최적 설정)
-
메모리 사용: 약 24GB (M1 Max/M2 Max 이상 권장)
-
용도: 명함 데이터에서 추출한 회사 URL 처리에 최적화
-
처리 속도: 약 5,000건/일 (단일 머신 기준)
macOS 환경에서 MPS를 활용할 경우, 자동으로 다음 코드가 실행됩니다:
# macOS에서 MPS 가용성 확인 및 활성화
if torch.backends.mps.is_available():
device = torch.device("mps")
logger.info(f"Apple MPS 가속 사용 중: {device}")
# MPS용 최적화 설정
os.environ["PYTORCH_MPS_HIGH_WATERMARK_RATIO"] = "0.0" # 메모리 동적 관리
model = AutoModelForCausalLM.from_pretrained(
model_name,
token=huggingface_token,
device_map="auto", # 자동으로 MPS 감지
torch_dtype=torch.float16, # MPS에 최적화된 정밀도
)
logger.info(f"모델 로드 완료: MPS 가속 활성화됨")2.2 리눅스 서버 환경 (대량 처리용)
대규모 처리를 위한 Linux CUDA 환경 설정은 다음과 같습니다:
-
하드웨어 가속: NVIDIA CUDA 기반 멀티 GPU(GPU0, GPU1) 병렬 처리
-
사용 모델: Google Gemma 3 12b-it (instruction-tuned 버전)
-
모델 특징: 12B 파라미터, 기업 정보 요약 능력 우수, 다국어 지원
-
실행 엔진: Hugging Face Transformers 라이브러리 (최적화된 CUDA 지원)
-
배치 크기: 16 (NVIDIA RTX A6000 메모리에 최적화됨)
-
양자화 기법: bfloat16 (정확도와 메모리 효율성의 균형점)
-
GPU 메모리 활용: 60-80% (모델 안정성 보장을 위한 최적 사용률)
-
처리 속도: 약 50,000건/일 (2x NVIDIA A6000 기준)
-
병렬화 전략: ProcessPoolExecutor로 GPU별 별도 프로세스 + ThreadPoolExecutor로 각 GPU내 URL 병렬 처리
리눅스 환경에서의 멀티 GPU 활용 코드:
# 멀티 GPU 환경 설정 및 병렬 처리
if torch.cuda.is_available():
gpu_count = torch.cuda.device_count()
logger.info(f"사용 가능한 GPU: {gpu_count}개")
# 각 GPU별 사양 확인 및 로깅
for i in range(gpu_count):
device_name = torch.cuda.get_device_name(i)
mem_total = torch.cuda.get_device_properties(i).total_memory / (1024**3)
mem_reserved = torch.cuda.memory_reserved(i) / (1024**3)
logger.info(f"GPU{i}: {device_name} (총 메모리: {mem_total:.1f}GB, 현재 사용: {mem_reserved:.1f}GB)")3. 모델 선정 과정과 성능 평가
스크립트 개발 과정에서 다양한 모델이 테스트되었으며, 각 모델별 성능과 리소스 요구사항이 철저히 분석되었습니다:
3.1 평가된 주요 모델들
- LG ExaOne 3.5:32b
-
장점: 한국어와 영어에 특화된 뛰어난 성능, 한국 기업 산업군 이해도가 높음
-
단점: 32B 파라미터로 인한 높은 메모리 요구사항 (최소 24GB 이상)
-
결과: 명함 처리용으로 채택 (한국 회사명 인식 성능 우수)
- Google Gemma 3 27b
-
장점: 정확도가 매우 높고 복잡한 회사 설명도 정교하게 생성
-
단점: 단일 GPU로는 로드 불가능, 모델 병렬화 필요 (40GB+ 메모리 요구)
-
결과: 현재 GPU 사양으로는 활용이 어려워 기각
- Google Gemma 3 4b
-
장점: 메모리 요구사항이 낮아 저사양 GPU에서도 구동 가능 (8GB 이상)
-
단점: 복잡한 프롬프트 이해력 부족, B2B 산업군 설명 정확도 낮음
-
결과: 정확도 이슈로 기각
- Google Gemma 3 12b
-
장점: 성능과 리소스 요구사항의 균형이 좋음, 일관된 결과물 생성
-
단점: 배치 크기 제한 (24GB GPU 메모리에서 최대 배치 16)
-
결과: 서버 환경 대량 처리용으로 채택
3.2 최적화 시도 및 결과
- 양자화 실험:
-
INT8 양자화: 50% 메모리 감소 달성했으나 메모리 충돌 이슈 잦음
-
bfloat16: 메모리 효율과 정확도의 최적 균형점 (채택)
- Fast-Attention 테스트:
-
FlashAttention-2 적용 시도 (2025년 4월 기준)
-
최신 Gemma 3 모델과 호환성 이슈 발생
-
이전 세대 모델에서는 2배 이상 추론 속도 향상 확인
-
향후 업데이트 시 재평가 예정
- 최종 결정:
-
명함 처리용: LG ExaOne 3.5:32b (Ollama)
-
서버 대량처리용: Gemma 3 12b-it (bfloat16)
4. 코드 동작 원리 및 주요 구성요소
4.1 URL 정보 추출 프로세스
extract_company_info_from_url 함수는 웹사이트에서 정보를 추출하는 핵심 요소입니다:
def extract_company_info_from_url(url, depth=0, gpu_id=0):
"""URL에서 회사 정보를 추출합니다."""
logger.info(f"GPU{gpu_id} - URL 정보 추출 중: {url} (깊이: {depth})")
start_time = time.time()
# URL 정규화 (http/https 프로토콜 추가, 특수문자 처리 등)
normalized_url = normalize_url(url)
if not normalized_url:
elapsed = time.time() - start_time
logger.warning(f"GPU{gpu_id} - 잘못된 URL 형식: {url}")
return url, f"오류: 잘못된 URL 형식 - {url}", elapsed
# 텍스트 추출 (외부 모듈 company_url_metadata_extractor 활용)
try:
logger.info(f"GPU{gpu_id} - URL 문서 추출 시작: {normalized_url}")
company_data = get_text_from_url(normalized_url, depth=depth, max_links_per_category=2)
# 추출 결과 가공 - 구조화된 형식으로 변환
info_text = f"회사 URL: {normalized_url}\n\n"
# 메타데이터 추가 (타이틀, 설명 등)
meta = company_data.get('metadata', {})
if meta.get('title'):
info_text += f"웹사이트 제목: {meta.get('title')}\n"
if meta.get('description'):
info_text += f"웹사이트 설명: {meta.get('description')}\n"
# 메인 페이지 내용 추가 (최대 1500자 제한)
main_content = company_data.get('main_content', '')
if main_content:
info_text += f"\n메인 페이지 내용:\n{main_content[:1500]}\n"
# 카테고리별 내용 추가 (About, Products, Services 등)
categories = company_data.get('categories', {})
for category, data in categories.items():
if data:
category_text = f"\n[{category}]\n"
for item in data[:2]: # 각 카테고리에서 최대 2개 항목만 사용
page_title = item.get('title', '제목 없음')
page_content = item.get('content', '')
if page_content:
category_text += f"{page_title}:\n{page_content[:800]}\n\n"
info_text += category_text
elapsed = time.time() - start_time
logger.info(f"GPU{gpu_id} - URL 문서 추출 완료: {normalized_url} ({elapsed:.2f}초 소요)")
return url, info_text, elapsed
except Exception as e:
logger.error(f"GPU{gpu_id} - URL 처리 중 오류 발생: {url} - {str(e)}")
elapsed = time.time() - start_time
return url, f"오류: URL 처리 실패 - {str(e)}", elapsed이 함수는 다음과 같은 세부 과정을 거칩니다:
-
입력된 URL을 정규화하여 유효한 웹 주소로 변환
-
웹 사이트에서 다음 정보를 추출:
-
HTML 메타데이터 (title, description, og:title, keywords 등)
-
메인 페이지 텍스트 콘텐츠
-
회사 정보 관련 주요 페이지 (About, Products, Services 등)
-
추출된 정보를 구조화된 텍스트로 가공하여 반환
-
전체 처리 시간을 측정하고 로깅
URL 크롤링 깊이(depth)에 따라 다음과 같은 정보가 수집됩니다:
-
depth=0: 메인 페이지만 분석
-
depth=1: 메인 페이지 + 직접 링크된 주요 카테고리 페이지
-
depth=2: 메인 페이지 + 주요 카테고리 + 2차 링크 일부 (서브 카테고리)
-
depth=3: 메인 페이지 + 주요 카테고리 + 2차 + 3차 링크 (전체 사이트의 약 60% 분석)
4.2 병렬 처리 아키텍처
process_urls_parallel 함수는 여러 URL을 동시에 처리하는 병렬 아키텍처의 핵심입니다:
def process_urls_parallel(urls, url_column, output_dir, model_name="google/gemma-3-12b-it",
depth=0, max_workers=16, batch_size=16, total_urls=None, gpu_id=None):
"""ThreadPoolExecutor를 사용하여 병렬로 URL 처리 후 배치로 모델에 전달"""
# 모델과 토크나이저 로드 (GPU 환경에 따라 다른 설정 적용)
tokenizer = AutoTokenizer.from_pretrained(model_name, token=huggingface_token)
# GPU ID가 지정된 경우 해당 GPU 사용
if gpu_id is not None and torch.cuda.is_available():
device = f"cuda:{gpu_id}"
model = AutoModelForCausalLM.from_pretrained(
model_name,
token=huggingface_token,
device_map=device, # 특정 GPU 지정
torch_dtype=torch.bfloat16, # float16 대신 bfloat16 사용
)
else:
# GPU ID가 지정되지 않은 경우 자동 할당 (macOS MPS 포함)
model = AutoModelForCausalLM.from_pretrained(
model_name,
token=huggingface_token,
device_map="auto", # 자동으로 가용 GPU 사용
torch_dtype=torch.bfloat16,
)
# 병렬 URL 처리 시작
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
# URL별 작업 제출 (각 URL마다 별도 스레드로 처리)
future_to_url = {executor.submit(extract_company_info_from_url, url, depth, gpu_id): url for url in urls}
# 결과 수집 및 진행상태 표시
for future in concurrent.futures.as_completed(future_to_url):
# 결과 처리...
# 12개 단위로 배치 처리 및 중간 결과 저장
if len(company_info_list) % 12 == 0:
# 배치 처리 실행
batch_results = batch_process_company_info(
company_info_list[-12:], model, tokenizer, batch_size, gpu_id
)
# 체크포인트 저장
save_checkpoint(
all_results,
output_dir,
url_column,
checkpoint_count, # 처리 횟수를 저장
gpu_id # GPU ID 전달
)이 코드는 다음과 같은 병렬 처리 패턴을 구현합니다:
1. 멀티 프로세싱 (GPU별 분리):
-
ProcessPoolExecutor를 사용해 GPU마다 별도 프로세스 할당
-
각 GPU는 독립적인 메모리 공간에서 모델 로드 및 실행
-
Linux 환경에서 GPU0, GPU1 등 동시 활용 가능
2. 멀티 스레딩 (URL 병렬 처리):
-
각 GPU별 프로세스 내에서 ThreadPoolExecutor로 여러 URL 동시 처리
-
max_workers=16으로 16개 URL을 동시에 크롤링 및 데이터 추출
-
네트워크 I/O 작업이 많아 스레딩이 효율적
3. 배치 처리 (모델 추론 최적화):
-
16개 URL 단위로 수집된 데이터를 한 번에 모델에 전달
-
실제 추론은 batch_size=16 단위로 나누어 메모리 최적화
-
완료된 배치마다 체크포인트 저장으로 안정성 확보
4.3 모델 추론 및 프롬프트 엔지니어링
batch_process_company_info 함수는 언어 모델을 사용해 회사 정보를 요약하는 핵심 로직입니다:
def batch_process_company_info(company_info_list, model, tokenizer, batch_size=16, gpu_id=0):
"""Gemma 3 모델을 사용하여 회사 정보를 배치로 처리합니다."""
logger.info(f"GPU{gpu_id} - 배치 처리: {len(company_info_list)}개 URL 처리 시작 (배치 크기: {batch_size})")
results = {}
# 시스템 프롬프트 정의 - 회사 설명 전문가 역할
system_prompt = """
You are a corporate analyst writing standardized company profiles.
Create a concise, factual description of the company based on the website information provided.
FOCUS ON THESE ELEMENTS ONLY:
- Company's core business and main offerings
- Primary products/services
- Target markets or customer segments
- Industry position or unique value proposition
FORMATTING REQUIREMENTS:
- Write 2-3 sentences ONLY with maximum 75 words total
- Write as plain text without ANY formatting, symbols, or markdown
- No bullet points, hyphens, hashtags, or special characters
- No URLs or links
- No section headers or labels
- No repetition of content
STYLE REQUIREMENTS:
- Use formal, professional business language
- Write exclusively in English
- Start directly with company details (no introductory phrases)
- Do not use phrases like "company description" or "is a company that"
"""
# 배치 처리 수행
for i in range(0, len(company_info_list), batch_size):
batch = company_info_list[i:i+batch_size]
# 각 URL 처리
for url, company_info, process_time in batch:
# 오류 메시지인 경우 처리 스킵
if company_info.startswith("오류:"):
results[url] = {
"success": False,
"error": company_info,
"url_process_time": process_time
}
continue
# 모델 입력 프롬프트 구성
prompt = f"<|system|>\n{system_prompt}\n5. 로깅 및 진행 상황 추적
실제 코드에서는 Python의 기본 로깅 시스템을 사용하고 있으며, 진행 상황을 추적하는 간단한 메커니즘이 구현되어 있습니다.
5.1 로그 시스템 구성
스크립트는 Python의 기본 logging 모듈을 활용합니다:
import logging
# 로거 설정
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler("company_url_processor.log"),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)주요 로그 수준:
-
INFO: 일반 진행 상황 (URL 처리 시작/완료, 배치 처리 등)
-
WARNING: 경고 상황 (잘못된 URL 등 비치명적 오류)
-
ERROR: 오류 상황 (크롤링 실패, 모델 오류 등)
5.2 진행 상황 출력
현재 스크립트는 단순하게 처리된 URL 수를 카운트하여 기본적인 진행 상황을 출력합니다:
# URL 처리 진행 상황 출력
processed_count = len(all_results)
logger.info(f"처리된 URL: {processed_count}/{len(urls)} ({processed_count/len(urls):.1%})")5.3 체크포인트 및 복구 시스템
실제 코드에서는 멀티 GPU 환경을 고려한 체크포인트 시스템이 구현되어 있습니다. GPU ID별로 별도의 체크포인트 파일을 생성하여 병렬 처리의 진행 상황을 개별적으로 저장합니다:
def save_checkpoint(results, output_dir, url_column, checkpoint_index, gpu_id=None):
"""
중간 결과를 JSON 형식으로 저장합니다.
Args:
results (dict): 처리 결과
output_dir (pathlib.Path): 저장할 디렉토리
url_column (str): URL 참조 컬럼명
checkpoint_index (int): 체크포인트 인덱스
gpu_id (int, optional): GPU ID. 지정되면 GPU별 체크포인트 파일로 저장
"""
output_data = []
# 결과 데이터 구성
for url, result in results.items():
item = {
url_column: url
}
if result["success"]:
item["description"] = result["summary"]
item["error"] = None
else:
item["description"] = None
item["error"] = result["error"]
item["process_time"] = result.get("url_process_time", 0)
# 모델 처리 시간 추가
if "model_process_time" in result:
item["model_process_time"] = result["model_process_time"]
output_data.append(item)
# 디렉토리가 없으면 생성
os.makedirs(output_dir, exist_ok=True)
# GPU ID에 따른 체크포인트 파일명 생성
if gpu_id is not None:
output_file = output_dir / f"checkpoint_gpu{gpu_id}_{checkpoint_index}.json"
else:
output_file = output_dir / f"checkpoint_{checkpoint_index}.json"
# JSON 형식으로 저장
with open(output_file, 'w', encoding='utf-8') as f:
json.dump(output_data, f, ensure_ascii=False, indent=2)
# GPU ID 정보를 포함한 로그 메시지
logger.info(f"{'GPU'+str(gpu_id)+' - ' if gpu_id is not None else ''}중간 결과 저장 완료: {output_file.name} ({len(output_data)}개 항목)")이 함수의 핵심 기능은 다음과 같습니다:
- GPU별 체크포인트 파일:
-
GPU ID가 지정된 경우 checkpointgpu{gpu_id}{checkpoint_index}.json 형식으로 파일 생성
-
예: checkpoint_gpu0_1.json, checkpoint_gpu1_1.json 등
-
GPU ID가 없는 경우에는 checkpoint_{checkpoint_index}.json 형식 사용
- 로그 메시지 맞춤화:
-
GPU ID가 지정된 경우 로그 메시지 앞에 "GPU{gpu_id} - " 접두사 추가
-
예: "GPU0 - 중간 결과 저장 완료: checkpoint_gpu0_1.json (300개 항목)"
- 저장 데이터 형식:
-
URL별 처리 결과 (성공/실패)
-
성공 시 생성된 회사 설명
-
실패 시 오류 메시지
-
URL 처리 소요 시간 및 모델 추론 시간
이 체크포인트 시스템을 통해 멀티 GPU 환경에서 각 GPU의 작업 진행 상황을 독립적으로 저장하고 관리할 수 있습니다. 또한 처리 중 중단되는 경우 각 GPU별로 마지막 저장 지점부터 작업을 재개할 수 있어 대규모 URL 처리 작업의 안정성과 복원력을 향상시킵니다.
6. 명령줄 인터페이스 및 설정 옵션
6.1 기본 명령줄 인수
스크립트는 명령줄 인터페이스를 통해 다양한 옵션을 제공합니다:
parser = argparse.ArgumentParser(description="Gemma 3 회사 정보 변환기 v5 (멀티 GPU 지원 버전)")
parser.add_argument("--input", required=True, help="회사 URL이 포함된 CSV 파일")
parser.add_argument("--output", required=True, help="처리 결과를 저장할 CSV 파일")
parser.add_argument("--column", default="company_url", help="URL이 포함된 컬럼명 (기본값: company_url)")
parser.add_argument("--records", type=int, help="처리할 최대 레코드 수")
parser.add_argument("--depth", type=int, default=0, help="크롤링 깊이 (기본값: 0)")
parser.add_argument("--batch", type=int, default=16, help="배치 크기 (기본값: 16)")
parser.add_argument("--workers", type=int, default=16, help="병렬 작업자 수 (기본값: 16)")
parser.add_argument("--gpus", type=int, default=2, help="사용할 GPU 개수 (기본값: 2)")
parser.add_argument("--model", default="google/gemma-3-12b-it", help="사용할 모델 이름 (기본값: google/gemma-3-12b-it)")모든 옵션 사용 예시:
python company_url_summarizer.py \
--input company_database.csv \
--output company_descriptions.csv \
--column "company_website" \
--records 5000 \
--depth 1 \
--batch 32 \
--workers 24 \
--gpus 2 \
--model "google/gemma-3-12b-it"