본 기사는 Zenrows blog에 처음 게시되었습니다.

이 튜토리얼에서는 봇 보호 비즈니스 디렉토리를 읽고, 페이지에 나열된 모든 회사를 추출하며, 각 회사 정보를 해당 웹사이트에서 보강하고, 이상적인 고객 프로필(ICP)과 비교하여 점수를 매기는 리드 생성 에이전트를 구축합니다. Python 3.10 이상 버전, OpenAI API 키, 그리고 Zenrows API 키가 필요합니다.

OpenAI Agents SDK는 에이전트 루프를 실행하며, @function_tool 데코레이터는 일반 Python 함수를 도구(tools)로 변환합니다. Zenrows Fetch는 보통 문제가 되는 페이지 자체를 가져오는 단계를 처리합니다. 테스트 실행에서 하나의 디렉토리 페이지가 78개의 회사를 생성했고, 에이전트는 각 회사에 대해 20초 이내에 점수를 매겼습니다.

완성된 코드는 GitHub에서 확인할 수 있으니 클론하여 따라 해 보세요.

전제 조건 (Prerequisites)

• Python 3.10 이상 버전이 필요합니다.
• 에이전트, 추출 도구(extraction tool), 점수 매기기 도구(scoring tool) 모두 사용하는 OpenAI API 키가 필요합니다. OpenAI developer dashboard에서 발급받으세요.
• 웹 페이지의 전체 내용을 검색하려면 Zenrows API 키가 필요합니다. signup page에서 만드세요.

의존성(dependencies)을 설치하세요.

python3 -m pip install openai-agents openai requests pydantic python-dotenv

SDK가 에이전트를 실행하고, requests는 Zenrows 호출을 담당하며, pydantic은 도구 스키마를 정의하고, python-dotenv는 키를 로드합니다. 프로젝트 루트에 .env 파일을 만들고 여기에 OPENAI_API_KEY와 ZENROWS_API_KEY를 추가한 후, 크리덴셜이 버전 관리에 올라가지 않도록 .gitignore에 .env를 추가하세요.

에이전트 구조 (How the agent is structured)

작업은 네 가지 도구로 나뉘며, 각각 하나의 작업을 수행합니다.

Tool	입력	출력	분리된 이유
fetch_page	URL	Markdown 형식의 페이지 내용	네트워크에 접근하는 유일한 작업이므로, 재시도 및 오류 처리를 한 곳에서 관리할 수 있음
...
From an ICP to ranked leads

작업을 분리하면 재시도 로직이 지역적으로 유지됩니다. 리드 중 하나에서 정보 보강(enrichment)에 실패하더라도, 에이전트는 해당 도구를 재시도하고 나머지 실행 과정은 온전히 유지합니다. 각 작업의 출력물 또한 다음 작업으로 전달하기에 충분히 작게 유지됩니다.

에이전트는 단지 두 개의 조합된 도구만 인식합니다. discover_leads는 fetch_page와 extract_leads를 실행하고, qualify_lead는 enrich_lead와 score_lead를 실행합니다. 만약 에이전트가 정보 보강된 리드를 두 개의 별도 도구 사이에 전달한다면, 모든 페이지 발췌본을 도구 인수로 재현해야 하며, 이는 대규모 디렉토리에서 컨텍스트 창(context window)을 초과하게 됩니다.

1. Zenrows Fetch 도구 구축

여기서 나오는 모든 코드 블록은 tools.py라는 하나의 파일에 들어갑니다. 각 블록을 같은 파일에 추가하세요.

import os
import re
import requests
...

_fetch_page가 로직을 담고 있으며, fetch_page는 @function_tool로 이를 감싸 에이전트가 호출할 수 있게 합니다. 이 데코레이터(decorator)는 함수를 직접 호출할 수 없는 FunctionTool 객체로 대체하므로, 로직을 일반 함수로 유지하면 테스트 가능합니다.

실패는 텍스트 문자열 형태로 반환되므로, 하나의 잘못된 URL이 실행 전체를 중단시키지 않습니다. 또한 각 메시지는 모델에게 재시도할 가치가 있는지 여부까지 알려줍니다.

mode=auto를 사용하면 Zenrows가 각 타겟이 필요로 하는 접근 설정을 자동으로 선택합니다. 간단한 페이지는 가장 가벼운 경로로 로드되며, 보호된(protected) 타겟만 추가적인 단계를 거칩니다. 이는 중요한데, 보호된 사이트들은 오류 메시지로 차단하는 경우가 드물기 때문입니다. Cloudflare-보호 페이지는 본문에 챌린지 페이지를 포함하며 200 상태 코드를 반환할 수 있고, 이 경우 에이전트는 그 챌린지를 리드가 없는 유효한 페이지로 간주합니다.

response_type=markdown은 탐색 메뉴와 일반적인 내용을 제거하여 모델이 추출하기에 더 깨끗한 입력을 제공합니다. 만약 디렉토리가 API 형태의 엔드포인트라면, response_type=json을 사용하고 변환 과정은 건너뛰세요. 아래 블록의 세 가지 상수는 다음 섹션에서 추출 단계에 속합니다.

2. 디렉토리 페이지에서 리드 추출하기

추출 코드를 tools.py에 추가하세요.

# 엄격한(strict) 도구 스키마는 빈 딕셔너리(bare dicts)를 거부하므로, 모든 도구의 입력과 출력은 모델이어야 합니다
class Lead(BaseModel):
    company: str
...

엄격한 도구 스키마는 빈 딕셔너리를 거부하기 때문에, 모든 도구의 입력과 출력은 Pydantic 모델이어야 합니다. _unwrap_redirects는 모델이 페이지를 읽기 전에 디렉토리의 추적 링크(tracking links)를 각 회사 자체 도메인으로 재작성하고, _reject_source_domain은 디렉토리를 가리키는 모든 웹사이트를 공백 처리합니다.

전체 디렉토리 페이지는 신뢰할 수 있는 단일 추출 호출로는 너무 길기 때문에, _extract_leads는 이를 40,000자 청크로 나누고 2,000자의 중복(overlap)을 사용합니다. 이 중복은 경계에 걸쳐 있는 모든 리스팅을 하나의 청크 전체로 가져가며, _key는 중복이 만든 중복 항목들을 제거합니다.

discover_leads는 fetch_page의 원본 출력이 자체적으로 컨텍스트 길이(context length)를 초과할 수 있기 때문에, 검색(fetching)과 추출을 하나의 도구로 결합했습니다.

에이전트 없이 두 작업 모두 테스트해 보세요. 이 스크립트를 test_fetch_extract.py로 저장하세요.

# test_fetch_extract.py
import json
import os
...

프로젝트 루트에서 실행합니다.

python3 test_fetch_extract.py

IT 서비스 디렉토리를 대상으로 로컬에서 실행한 결과입니다.

using cached page, 543637 chars
457647 chars after unwrapping redirects, 85990 saved

...

페이지는 543,637자로 반환되었고, 리다이렉트(redirects)를 풀어서 처리하는 과정에서 85,990자를 절약했습니다. 카운트는 디렉토리 페이지가 변경되고 모델이 같은 페이지를 다르게 읽을 수 있기 때문에 실행마다 달라집니다.

3. 각 리드 풍부화 및 점수 매기기(Enrich and score each lead)

tools.py에 풍부화 코드를 추가하세요.

# matches any markdown link, used to read a company's own navigation
MARKDOWN_LINK = re.compile(r"[([^\)]+)](https?://[^\\)]+)")

# guessing paths costs a fetch per miss, so read the site's nav instead and
# follow only the links it actually has
SIGNAL_KEYWORDS = {
    "hiring": ["career", "job", "join", "hiring", "work with us", "we are hiring"],
    "services": ["service", "what we do", "solutions", "expertise", "capabilities"],
    "portfolio": ["portfolio", "case stud", "our work", "projects", "clients"],
    "about": ["about", "who we are", "our story", "team"],
    "contact": ["contact", "get in touch", "book a call", "let's talk", "talk to us"],
}

# dict[str, X] is rejected by strict schemas too, which is why signals is a list
class Signal(BaseModel):
    name: str
    found: bool
    url: str
    excerpt: str

class EnrichedLead(BaseModel):
    company: str
    website: str
    signals: list[Signal]

def _discover_links(markdown: str, website: str) -> dict:
    """find real urls for each signal by reading the homepage nav."""
    host = urlparse(website).netloc.replace("www.", "")
    homepage_path = urlparse(website).path.rstrip("/") or "/"
    found = {}

    for text, url in MARKDOWN_LINK.findall(markdown):
        parsed = urlparse(url)

회사 자체 도메인 링크만 추적합니다

    if parsed.netloc.replace("www.", "") != host:
        continue

    # 홈페이지에 대한 앵커는 이미 홈페이지 발췌 내용에 포함된 콘텐츠입니다
    if parsed.fragment and (parsed.path.rstrip("/") or "/") == homepage_path:
        continue

    # 프래그먼트를 제거합니다. 서버가 반환하는 내용은 절대 변하지 않기 때문입니다.
    clean_url = f"{parsed.scheme}://{parsed.netloc}{parsed.path}"

haystack = f"{text} {url}".lower()
for signal, keywords in SIGNAL_KEYWORDS.items():
if signal not in found and any(k in haystack for k in keywords):
found[signal] = clean_url

return found

def _enrich_lead(company: str, website: str) -> EnrichedLead:
"""회사 자체 사이트에서 신호(signals)를 수집합니다."""
# 도메인이 없는 리드라도 스코어링을 거치며, 알려진 정보가 적더라도 평가됩니다
if not website:
return EnrichedLead(company=company, website="", signals=[])

homepage = _fetch_page(website)
if homepage.startswith("FETCH_ERROR"):
    return EnrichedLead(company=company, website=website, signals=[])

# 홈페이지가 가장 정보력이 높은 단일 페이지이며, 원페이지 사이트의 경우
# 탐색 링크가 가리키는 모든 것을 담고 있습니다.
signals = [Signal(name="homepage", found=True, url=website, excerpt=homepage[:6000])]

for name, url in _discover_links(homepage, website).items():
    content = _fetch_page(url)
    ok = not content.startswith("FETCH_ERROR")
    signals.append(Signal(
        name=name,
        found=ok,
        url=url,
        # 발췌 내용은 작게 유지합니다. 스코어링이 모든 내용을 한 번에 읽기 때문입니다.
        excerpt=content[:2000] if ok else ""
    ))

return EnrichedLead(company=company, website=website, signals=signals)

@function_tool
def enrich_lead(company: str, website: str) -> EnrichedLead:
"""회사 자체 웹사이트에서 신호(signals)를 수집합니다.

Arg