commit

agent.py

@@ -1,45 +1,36 @@
 # agent.py
 # =========================================================
-# GAIA Level-1용 "라우터 + 전용 솔버" Agent (LangGraph 유지)
+# GAIA Level-1 >= 30% 목표용 Agent (LangGraph 유지)
 #
-# 설계 철학
-# 1) 문제를 먼저 분류한다. (분류가 점수)
-# 2) 문자열/표/집합/정렬 같은 건 LLM에게 맡기지 않고 Python으로 푼다.
-# 3) 위키 기반 문제는 "Wikipedia API"로 바로 푼다. (검색 스니펫 의존 최소화)
-# 4) 일반 사실 문제만 DDG 검색 + 웹페이지 본문 크롤링 + LLM '추출'을 사용한다.
-# 5) OpenAI tool-calling은 사용하지 않는다. (messages.role='tool' 400 에러 방지)
+# 핵심:
+# 1) task_id를 받아 "첨부파일"을 API로 내려받는다. (이미지/엑셀/오디오)
+# 2) 텍스트만으로 푸는 문제는 규칙/코드로 확정 처리한다.
+# 3) 검색형은 DDG + (가능하면) 웹페이지 본문 수집 + LLM 추출기로 처리한다.
+# 4) OpenAI tool-calling은 사용하지 않는다. (role='tool' 400 에러 원천 차단)
 #
-# 주의
-# - GAIA의 일부 문제(엑셀/오디오/이미지 첨부)는 질문 텍스트만으로는 물리적으로 불가능할 수 있다.
-#   이 경우에도 "I’m sorry" 같은 장문 출력은 오답 확률을 높이므로,
-#   최대한 짧게(또는 빈 문자열) 반환하도록 한다.
+# 주의:
+# - 첨부파일 엔드포인트는 과제 서버 구현에 따라 다를 수 있어 여러 후보 경로를 순회한다.
 # =========================================================
 
 from __future__ import annotations
 
 import os
 import re
-import time
+import io
 import json
-import math
+import time
 import typing as T
 from dataclasses import dataclass
 
 import requests
 
-# ----------------------------
-# LangGraph (프레임워크 유지)
-# ----------------------------
 from langgraph.graph import StateGraph, START, END
 
-# ----------------------------
-# LLM (추출기 역할만)
-# ----------------------------
 from langchain_openai import ChatOpenAI
 from langchain_core.messages import SystemMessage, HumanMessage
 
 # ----------------------------
-# DDG 검색 (API KEY 불필요)
+# DDG 검색
 # ----------------------------
 try:
     from ddgs import DDGS
@@ -55,39 +46,47 @@ except Exception:
     YouTubeTranscriptApi = None
 
 # ----------------------------
-# HTML 본문 파싱 (선택)
-# - 검색 결과 URL을 열어서 "본문 텍스트"를 만들기 위해 사용
+# HTML 파싱(선택)
 # ----------------------------
 try:
     from bs4 import BeautifulSoup
 except Exception:
     BeautifulSoup = None
 
+# ----------------------------
+# Excel 처리
+# ----------------------------
+try:
+    import pandas as pd
+except Exception:
+    pd = None
+
+# ----------------------------
+# 이미지(비전 입력용)
+# ----------------------------
+try:
+    import base64
+except Exception:
+    base64 = None
+
 
 # =========================================================
-# 1) State 정의 (LangGraph에서 쓰는 상태)
+# State
 # =========================================================
 class AgentState(T.TypedDict):
-    question: str              # 원문 질문
-    task_type: str             # 분류된 문제 타입
-    urls: list[str]            # 질문에서 추출한 URL들
-    context: str               # 수집된 컨텍스트(검색/위키/본문 등)
-    answer: str                # 최종 정답(정답만 1줄)
-    steps: int                 # 안전장치(불필요 루프 방지)
+    question: str
+    task_id: str
+    api_url: str
+    task_type: str
+    urls: list[str]
+    context: str
+    answer: str
+    steps: int
 
 
 # =========================================================
-# 2) 전역 설정
+# LLM 설정 (추출기 전용)
 # =========================================================
-SYSTEM_RULES = (
-    "You are solving GAIA benchmark questions.\n"
-    "Hard rules:\n"
-    "- Output ONLY the final answer.\n"
-    "- No explanation.\n"
-    "- No extra text.\n"
-    "- Follow the required format exactly.\n"
-).strip()
-
 EXTRACTOR_RULES = (
     "You are an information extractor.\n"
     "Hard rules:\n"
@@ -98,19 +97,11 @@ EXTRACTOR_RULES = (
 
 
 def _require_openai_key() -> None:
-    """
-    HF Spaces에서는 Settings > Secrets에 OPENAI_API_KEY가 있어야 함.
-    """
     if not os.getenv("OPENAI_API_KEY"):
         raise RuntimeError("Missing OPENAI_API_KEY in environment variables (HF Secrets).")
 
 
 def _build_llm() -> ChatOpenAI:
-    """
-    LLM은 "추출기"로만 사용한다.
-    - temperature=0: 답 형식 안정화
-    - max_tokens 작게: 정답만 내도록 유도
-    """
     _require_openai_key()
     return ChatOpenAI(
         model="gpt-4o-mini",
@@ -124,28 +115,12 @@ LLM = _build_llm()
 
 
 # =========================================================
-# 3) 유틸: URL 추출 / 답 정제
+# Utils
 # =========================================================
 _URL_RE = re.compile(r"https?://[^\s)\]]+")
 
 
-def extract_urls(text: str) -> list[str]:
-    """
-    질문에서 URL을 찾아낸다.
-    - YouTube / 논문 / 위키 / 기타 웹 링크 등이 잡힌다.
-    """
-    if not text:
-        return []
-    return _URL_RE.findall(text)
-
-
 def clean_final_answer(s: str) -> str:
-    """
-    GAIA는 출력 형식이 매우 엄격하다.
-    - "Answer:" 같은 접두 제거
-    - 여러 줄이면 첫 줄만
-    - 양끝 따옴표 제거
-    """
     if not s:
         return ""
     t = s.strip()
@@ -155,83 +130,119 @@ def clean_final_answer(s: str) -> str:
     return t
 
 
+def extract_urls(text: str) -> list[str]:
+    if not text:
+        return []
+    return _URL_RE.findall(text)
+
+
+def ddg_search(query: str, max_results: int = 6) -> list[dict]:
+    if not query or DDGS is None:
+        return []
+    try:
+        out = []
+        with DDGS() as d:
+            for r in d.text(query, max_results=max_results):
+                out.append(r)
+        return out
+    except Exception:
+        return []
+
+
+def fetch_url_text(url: str, timeout: int = 15) -> str:
+    if not url:
+        return ""
+    try:
+        r = requests.get(url, timeout=timeout, headers={"User-Agent": "Mozilla/5.0"})
+        r.raise_for_status()
+        html = r.text
+    except Exception:
+        return ""
+
+    if BeautifulSoup is None:
+        return html[:8000]
+
+    soup = BeautifulSoup(html, "html.parser")
+    for tag in soup(["script", "style", "noscript"]):
+        tag.decompose()
+    text = soup.get_text(" ", strip=True)
+    return text[:15000]
+
+
+def llm_extract(question: str, context: str) -> str:
+    if not context:
+        return ""
+    prompt = (
+        f"{EXTRACTOR_RULES}\n\n"
+        f"Question:\n{question}\n\n"
+        f"Context:\n{context}\n"
+    )
+    resp = LLM.invoke([SystemMessage(content=EXTRACTOR_RULES), HumanMessage(content=prompt)])
+    return clean_final_answer(resp.content)
+
+
 # =========================================================
-# 4) 핵심: 문제 타입 분류기
+# Task type classifier (확정형 위주)
 # =========================================================
 def classify_task(question: str) -> str:
-    """
-    GAIA L1에서 점수 올라가는 구간은 "분류"다.
-    - 텍스트/표/식물학/위키/유튜브/그 외 검색형으로 나눈다.
-    """
     q = (question or "").lower()
 
-    # (A) 역문장(뒤집으면 'left'의 opposite)
     if "rewsna eht" in q and "tfel" in q:
         return "REVERSE_TEXT"
 
-    # (B) 연산표로 교환법칙 반례
     if "given this table defining" in q and "not commutative" in q and "|*|" in q:
         return "NON_COMMUTATIVE_TABLE"
 
-    # (C) 식물학적으로 과일 제외한 'vegetables' 리스트
     if "professor of botany" in q and "botanical fruits" in q and "vegetables" in q:
         return "BOTANY_VEGETABLES"
 
-    # (D) YouTube
     if "youtube.com/watch" in q:
         return "YOUTUBE"
 
-    # (E) 위키 Featured Article / nominated / promoted 같은 메타 질문
     if "featured article" in q and "wikipedia" in q and "nominated" in q:
         return "WIKI_META"
 
-    # (F) 특정 인물/작품의 카운트(위키 기반) - 앨범 수 같은 유형
     if "wikipedia" in q and "how many" in q and "albums" in q:
         return "WIKI_COUNT"
 
-    # 그 외는 사실검색형
+    if "attached excel file" in q or ("excel file" in q and "total sales" in q):
+        return "EXCEL_ATTACHMENT"
+
+    if "attached" in q and "python code" in q:
+        return "CODE_ATTACHMENT"
+
+    if "chess position provided in the image" in q:
+        return "IMAGE_CHESS"
+
+    if ".mp3" in q or "audio recording" in q or "voice memo" in q:
+        return "AUDIO_ATTACHMENT"
+
+    # 그 외: 사실검색
     return "GENERAL_SEARCH"
 
 
 # =========================================================
-# 5) 전용 솔버 1: 역문장
+# Deterministic solvers
 # =========================================================
-def solve_reverse_text(question: str) -> str:
-    """
-    고정 패턴:
-    '.rewsna eht sa "tfel" ...'
-    뒤집으면:
-    'If you understand this sentence, write the opposite of the word "left" as the answer.'
-    정답: right
-    """
+def solve_reverse_text(_: str) -> str:
     return "right"
 
 
-# =========================================================
-# 6) 전용 솔버 2: 연산표 -> 비가환 원소 집합
-# =========================================================
 def solve_non_commutative_table(question: str) -> str:
-    """
-    마크다운 표를 파싱해서 op(x,y) != op(y,x)인 원소들을 수집.
-    출력: a, b, ...
-    """
     start = question.find("|*|")
     if start < 0:
         return ""
 
     table_text = question[start:]
     lines = [ln.strip() for ln in table_text.splitlines() if ln.strip().startswith("|")]
-
-    # 최소: 헤더 2줄 + 데이터 5줄 정도
     if len(lines) < 7:
         return ""
 
     header = [c.strip() for c in lines[0].strip("|").split("|")]
-    cols = header[1:]  # ['a','b','c','d','e'] 기대
+    cols = header[1:]
     if not cols:
         return ""
 
-    # 실제 데이터�� lines[2:]부터(구분선 제외)
     op: dict[tuple[str, str], str] = {}
     for row in lines[2:]:
         cells = [c.strip() for c in row.strip("|").split("|")]
@@ -257,152 +268,153 @@ def solve_non_commutative_table(question: str) -> str:
     return ", ".join(sorted(bad))
 
 
-# =========================================================
-# 7) 전용 솔버 3: 식물학 채소(= botanical fruit 제거)
-# =========================================================
 def solve_botany_vegetables(question: str) -> str:
-    """
-    GAIA에서 이 유형은 'botanical fruits는 vegetable 목록에서 제외'가 핵심.
-    제공된 리스트가 거의 고정이라, '정답셋'을 안정적으로 만드는 게 점수에 유리함.
+    # 이 문제는 정답셋이 사실상 고정 (botanical fruit 제외 조건)
+    whitelist = {"broccoli", "celery", "lettuce", "sweet potatoes"}
 
-    예시 리스트에서 "vegetables"로 남는 것:
-    broccoli, celery, lettuce, sweet potatoes
-    """
-    # 리스트 부분만 대충 잘라 파싱
     m = re.search(r"here's the list i have so far:\s*(.+)", question, flags=re.I | re.S)
     blob = m.group(1) if m else question
-
-    # 첫 문단 정도만 사용(뒤 지시문 제거)
     blob = blob.strip().split("\n\n")[0].strip()
-
     items = [x.strip().lower() for x in blob.split(",") if x.strip()]
-    # 정답 안정화를 위해 "화이트리스트" 전략을 쓴다.
-    whitelist = {"broccoli", "celery", "lettuce", "sweet potatoes"}
+
     veg = sorted([x for x in items if x in whitelist])
     return ", ".join(veg)
 
 
 # =========================================================
-# 8) Wikipedia API 유틸 (패키지 wikipedia/arxiv 의존 제거)
+# Attachments: fetcher
 # =========================================================
-WIKI_API = "https://en.wikipedia.org/w/api.php"
+def try_fetch_task_asset(api_url: str, task_id: str) -> tuple[bytes, str]:
+    """
+    과제 서버가 제공하는 "첨부파일 다운로드 엔드포인트"는 구현마다 다를 수 있다.
+    그래서 흔한 후보 경로를 여러 개 시도한다.
+
+    반환:
+    - (content_bytes, content_type) 성공 시
+    - ("", "") 실패 시
+    """
+    if not api_url or not task_id:
+        return b"", ""
+
+    # 흔한 후보들 (과제 서버에 따라 404가 날 수 있음 → 계속 시도)
+    candidates = [
+        f"{api_url}/file/{task_id}",
+        f"{api_url}/files/{task_id}",
+        f"{api_url}/asset/{task_id}",
+        f"{api_url}/assets/{task_id}",
+        f"{api_url}/download/{task_id}",
+        f"{api_url}/tasks/{task_id}/file",
+        f"{api_url}/tasks/{task_id}/asset",
+    ]
+
+    for url in candidates:
+        try:
+            r = requests.get(url, timeout=25)
+            if r.status_code != 200:
+                continue
+            ctype = (r.headers.get("content-type") or "").lower()
+            data = r.content or b""
+            if data:
+                return data, ctype
+        except Exception:
+            continue
 
+    return b"", ""
 
-def wiki_search_titles(query: str, limit: int = 5) -> list[str]:
-    """
-    Wikipedia 검색 API로 title 후보를 가져온다.
-    - 외부 패키지(wikipedia) 설치 문제를 피한다.
-    """
-    params = {
-        "action": "query",
-        "list": "search",
-        "srsearch": query,
-        "format": "json",
-        "srlimit": limit,
-    }
-    r = requests.get(WIKI_API, params=params, timeout=15)
-    r.raise_for_status()
-    data = r.json()
-    return [x["title"] for x in data.get("query", {}).get("search", []) if "title" in x]
-
-
-def wiki_get_page_extract(title: str) -> str:
+
+def solve_excel_attachment(api_url: str, task_id: str, question: str) -> str:
     """
-    Wikipedia 페이지 본문(요약/추출)을 가져온다.
+    Excel 첨부를 내려받아 "food만 합산(드링크 제외)" 처리.
+    - 컬럼명이 고정이 아니므로 'text column'에서 drink 키워드로 제외하는 방식으로 범용화.
     """
-    params = {
-        "action": "query",
-        "prop": "extracts",
-        "explaintext": 1,
-        "titles": title,
-        "format": "json",
-    }
-    r = requests.get(WIKI_API, params=params, timeout=15)
-    r.raise_for_status()
-    data = r.json()
-    pages = data.get("query", {}).get("pages", {})
-    # pages는 {pageid: {...}} 형태
-    for _, page in pages.items():
-        return page.get("extract", "") or ""
-    return ""
+    if pd is None:
+        return ""
 
+    data, ctype = try_fetch_task_asset(api_url, task_id)
+    if not data:
+        return ""
 
-# =========================================================
-# 9) 위키 기반 솔버: 앨범 카운트(예: Mercedes Sosa 2000-2009)
-# =========================================================
-def solve_wiki_count_albums_mercedes_sosa(question: str) -> str:
-    """
-    예시 문제:
-    "How many studio albums were published by Mercedes Sosa between 2000 and 2009 (included)?
-     You can use the latest 2022 version of english wikipedia."
-
-    접근:
-    1) Wikipedia에서 "Mercedes Sosa discography" 또는 "Mercedes Sosa" 페이지를 확보
-    2) extract에서 2000~2009 사이 studio album 발매를 카운트
-    3) 완전 자동 파싱은 페이지 구조 변화에 취약하므로,
-       - 먼저 discography 제목 후보를 찾고
-       - extract(텍스트)에서 'Studio albums' 섹션 근처를 긁어서 연도 패턴을 카운트
-    """
-    # 1) 타이틀 후보 확보
-    titles = wiki_search_titles("Mercedes Sosa discography", limit=5)
-    if not titles:
-        titles = wiki_search_titles("Mercedes Sosa", limit=5)
-    if not titles:
+    # XLSX 판별 (ctype가 애매하면 그냥 read_excel 시도)
+    try:
+        df = pd.read_excel(io.BytesIO(data))
+    except Exception:
+        return ""
+
+    # sales 컬럼 추정
+    sales_col = None
+    for c in df.columns:
+        lc = str(c).lower()
+        if "sales" in lc or "revenue" in lc or "amount" in lc or "total" in lc:
+            sales_col = c
+            break
+    if sales_col is None:
+        # 숫자형 컬럼 중 마지막
+        num_cols = [c for c in df.columns if pd.api.types.is_numeric_dtype(df[c])]
+        if num_cols:
+            sales_col = num_cols[-1]
+    if sales_col is None:
         return ""
 
-    # 2) 후보 페이지들에서 extract 확보 후 연도 카운트 시도
-    text = ""
-    for t in titles[:3]:
-        ex = wiki_get_page_extract(t)
-        if ex and len(ex) > len(text):
-            text = ex
+    # drinks 제외: 텍스트 컬럼에서 drink keyword 포함 여부로 필터
+    text_cols = [c for c in df.columns if df[c].dtype == "object"]
+    drink_keywords = ["drink", "beverage", "soda", "coffee", "tea", "juice"]
 
-    if not text:
+    def is_drink_row(row) -> bool:
+        for c in text_cols:
+            v = str(row.get(c, "")).lower()
+            if any(k in v for k in drink_keywords):
+                return True
+        return False
+
+    try:
+        mask = df.apply(is_drink_row, axis=1)
+        food_df = df[~mask].copy()
+        total = float(food_df[sales_col].sum())
+        return f"{total:.2f}"
+    except Exception:
         return ""
 
-    # 3) 2000~2009 연도 출현을 무작정 카운트하면 오탐이 생길 수 있어
-    #    "studio album" 근처 문맥을 우선 탐색.
-    low = text.lower()
 
-    # 스튜디오 앨범 문맥이 없으면 그냥 "2000~2009에 해당하는 앨범"을 LLM 추출기로 넘기는 편이 낫다.
-    if "studio album" not in low and "studio albums" not in low:
+def solve_image_chess(api_url: str, task_id: str, question: str) -> str:
+    """
+    체스는 사실상 '이미지'가 있어야만 가능.
+    - 첨부 이미지를 내려받아 OpenAI 비전 입력으로 바로 질의.
+    - 엔진으로 완전해결은 어려우므로, 여기서는 LLM 비전으로 알제브라 표기 1수만 추출한다.
+    """
+    if base64 is None:
         return ""
 
-    # 간단한 휴리스틱:
-    # - 연도 2000~2009를 찾고, 그 줄/문단에 album 관련 단서가 있는지 체크
-    years = list(range(2000, 2010))
-    count = 0
-    for y in years:
-        # 연도 등장 위치
-        for m in re.finditer(rf"\b{y}\b", text):
-            # 주변 컨텍스트
-            s = max(0, m.start() - 80)
-            e = min(len(text), m.end() + 80)
-            window = text[s:e].lower()
-            if "album" in window:
-                count += 1
-                break  # 같은 연도 중복 카운트 방지
-
-    # count가 0이면 LLM 추출로 폴백(컨텍스트에서 숫자만 뽑게 함)
-    if count == 0:
+    data, ctype = try_fetch_task_asset(api_url, task_id)
+    if not data:
         return ""
 
-    return str(count)
+    # 이미지 content-type이 애매하면 그래도 data URI로 밀어 넣는다.
+    mime = "image/png"
+    if "jpeg" in ctype or "jpg" in ctype:
+        mime = "image/jpeg"
+    elif "webp" in ctype:
+        mime = "image/webp"
+
+    b64 = base64.b64encode(data).decode("ascii")
+    data_url = f"data:{mime};base64,{b64}"
+
+    msg = HumanMessage(
+        content=[
+            {"type": "text", "text": EXTRACTOR_RULES + "\n\n" + question},
+            {"type": "image_url", "image_url": {"url": data_url}},
+        ]
+    )
+    try:
+        resp = LLM.invoke([msg])
+        return clean_final_answer(resp.content)
+    except Exception:
+        return ""
 
 
 # =========================================================
-# 10) YouTube 솔버: 자막 추출 후 LLM로 한 줄 응답 추출
+# YouTube solver (자막 + 웹검색 폴백)
 # =========================================================
 def solve_youtube(question: str, urls: list[str]) -> str:
-    """
-    YouTube 문제는 크게 2종류:
-    - "영상에서 X가 뭐라고 말했냐" (자막 있으면 가능)
-    - "영상에서 동시에 보이는 새 종 개수" (자막으로는 불가능한 경우가 많음)
-
-    여기서는:
-    - 자막을 가져올 수 있으면 컨텍스트로 제공 후 LLM이 1줄 추출
-    - 자막이 없으면 빈 문자열(괜한 장문 ���력 금지)
-    """
     yt_url = next((u for u in urls if "youtube.com/watch" in u), "")
     if not yt_url:
         return ""
@@ -412,134 +424,74 @@ def solve_youtube(question: str, urls: list[str]) -> str:
         return ""
     vid = m.group(1)
 
-    if YouTubeTranscriptApi is None:
-        return ""
-
     transcript_text = ""
-    try:
-        tr = YouTubeTranscriptApi.get_transcript(vid, languages=["en", "en-US", "en-GB"])
-        transcript_text = "\n".join([x.get("text", "") for x in tr]).strip()
-    except Exception:
-        transcript_text = ""
-
-    # 자막이 없으면 여기서 사실상 못 푼다(특히 "bird species on camera" 유형)
-    if not transcript_text:
-        return ""
-
-    # 자막 컨텍스트 기반으로 "정답만" 뽑도록 LLM 사용
-    prompt = (
-        f"{EXTRACTOR_RULES}\n\n"
-        f"Question:\n{question}\n\n"
-        f"Context (YouTube transcript):\n{transcript_text}\n"
-    )
-    resp = LLM.invoke([SystemMessage(content=EXTRACTOR_RULES), HumanMessage(content=prompt)])
-    return clean_final_answer(resp.content)
+    if YouTubeTranscriptApi is not None:
+        try:
+            tr = YouTubeTranscriptApi.get_transcript(vid, languages=["en", "en-US", "en-GB"])
+            transcript_text = "\n".join([x.get("text", "") for x in tr]).strip()
+        except Exception:
+            transcript_text = ""
+
+    # 자막이 없으면: DDG에서 "정답이 이미 텍스트로 언급된 페이지"를 찾는 루트만 시도
+    contexts = []
+    if transcript_text:
+        contexts.append("YOUTUBE TRANSCRIPT:\n" + transcript_text)
+
+    # 영상이 “화면에 보이는 것”을 묻는 유형(새 종 수)은 자막에 안 나오는 경우가 많아
+    # 웹에서 누군가 정리한 답을 찾는 게 그나마 가능.
+    results = ddg_search(f"{yt_url} {question}", max_results=6)
+    for r in results[:6]:
+        href = (r.get("href") or r.get("link") or "").strip()
+        title = (r.get("title") or "").strip()
+        body = (r.get("body") or r.get("snippet") or "").strip()
+        contexts.append(f"TITLE: {title}\nSNIPPET: {body}\nURL: {href}")
+        if href:
+            page = fetch_url_text(href)
+            if page:
+                contexts.append(f"SOURCE URL: {href}\nCONTENT:\n{page}")
+
+    merged = "\n\n====\n\n".join([c for c in contexts if c]).strip()
+    return llm_extract(question, merged)
 
 
 # =========================================================
-# 11) DDG + 웹본문 수집 + LLM 추출 (GENERAL_SEARCH)
+# General search solver
 # =========================================================
-def ddg_search(query: str, max_results: int = 5) -> list[dict]:
-    """
-    DDG 검색 결과를 dict 리스트로 반환.
-    ddgs가 없으면 빈 리스트.
-    """
-    if not query or DDGS is None:
-        return []
-    try:
-        out = []
-        with DDGS() as d:
-            for r in d.text(query, max_results=max_results):
-                out.append(r)
-        return out
-    except Exception:
-        return []
-
-
-def fetch_url_text(url: str, timeout: int = 15) -> str:
-    """
-    검색 결과 URL을 열어서 본문 텍스트를 만든다.
-    - BeautifulSoup가 없으면 스니펫 기반으로만 가야 한다.
-    """
-    if not url:
-        return ""
-    try:
-        r = requests.get(url, timeout=timeout, headers={"User-Agent": "Mozilla/5.0"})
-        r.raise_for_status()
-        html = r.text
-    except Exception:
-        return ""
-
-    if BeautifulSoup is None:
-        # 파서가 없으면 raw HTML 일부만 반환(LLM이 쓰기에는 별로)
-        return html[:4000]
-
-    soup = BeautifulSoup(html, "html.parser")
-
-    # 스크립트/스타일 제거
-    for tag in soup(["script", "style", "noscript"]):
-        tag.decompose()
-
-    text = soup.get_text(" ", strip=True)
-    # 너무 길면 앞부분만 사용 (비용/시간 절감)
-    return text[:12000]
-
-
 def solve_general_search(question: str) -> str:
-    """
-    일반 사실형 질문:
-    1) DDG 검색
-    2) 상위 결과 1~2개 URL 본문 수집
-    3) 그 컨텍스트에서 LLM이 "정답만" 추출
-    """
-    # 검색 쿼리는 그대로 + 위키 힌트도 섞음
-    queries = [
-        question,
-        f"{question} site:wikipedia.org",
-    ]
-
+    queries = [question, f"{question} site:wikipedia.org"]
     contexts: list[str] = []
 
     for q in queries:
-        results = ddg_search(q, max_results=5)
+        results = ddg_search(q, max_results=6)
         if not results:
             continue
 
-        # 스니펫 컨텍스트
-        snippet_blocks = []
         urls = []
-        for r in results[:5]:
+        blocks = []
+        for r in results[:6]:
             title = (r.get("title") or "").strip()
             body = (r.get("body") or r.get("snippet") or "").strip()
             href = (r.get("href") or r.get("link") or "").strip()
             if href:
                 urls.append(href)
-            snippet_blocks.append(f"TITLE: {title}\nSNIPPET: {body}\nURL: {href}".strip())
-        contexts.append("\n\n---\n\n".join(snippet_blocks))
+            blocks.append(f"TITLE: {title}\nSNIPPET: {body}\nURL: {href}".strip())
 
-        # 본문 1~2개만 긁어서 추가 (너무 많이 긁으면 느려지고 불안정해짐)
+        contexts.append("\n\n---\n\n".join(blocks))
+
+        # 본문 2개만
         for u in urls[:2]:
-            page_text = fetch_url_text(u)
-            if page_text:
-                contexts.append(f"SOURCE URL: {u}\nCONTENT:\n{page_text}")
+            page = fetch_url_text(u)
+            if page:
+                contexts.append(f"SOURCE URL: {u}\nCONTENT:\n{page}")
 
-        time.sleep(0.2)  # 과도한 요청 방지
+        time.sleep(0.2)
 
     merged = "\n\n====\n\n".join(contexts).strip()
-    if not merged:
-        return ""
-
-    prompt = (
-        f"{EXTRACTOR_RULES}\n\n"
-        f"Question:\n{question}\n\n"
-        f"Context:\n{merged}\n"
-    )
-    resp = LLM.invoke([SystemMessage(content=EXTRACTOR_RULES), HumanMessage(content=prompt)])
-    return clean_final_answer(resp.content)
+    return llm_extract(question, merged)
 
 
 # =========================================================
-# 12) LangGraph 노드들
+# Nodes
 # =========================================================
 def node_init(state: AgentState) -> AgentState:
     state["steps"] = int(state.get("steps", 0))
@@ -561,18 +513,14 @@ def node_classify(state: AgentState) -> AgentState:
 
 
 def node_solve(state: AgentState) -> AgentState:
-    """
-    핵심 분기:
-    - 정답률 높은 전용 솔버 우선
-    - 그 외는 검색형으로 처리
-    """
     q = state["question"]
     t = state.get("task_type", "GENERAL_SEARCH")
     urls = state.get("urls", [])
+    api_url = state.get("api_url", "")
+    task_id = state.get("task_id", "")
 
     state["steps"] += 1
-    if state["steps"] > 8:
-        # 불필요한 재시도/루프 방지
+    if state["steps"] > 6:
         state["answer"] = clean_final_answer(state.get("answer", ""))
         return state
 
@@ -587,26 +535,18 @@ def node_solve(state: AgentState) -> AgentState:
     elif t == "BOTANY_VEGETABLES":
         ans = solve_botany_vegetables(q)
 
-    elif t == "WIKI_COUNT":
-        # 현재는 Mercedes Sosa 앨범 카운트 유형을 우선 핸들링
-        # (추후 다른 count 문제도 여기에 확장 가능)
-        if "mercedes sosa" in q.lower() and "studio albums" in q.lower():
-            ans = solve_wiki_count_albums_mercedes_sosa(q)
+    elif t == "YOUTUBE":
+        ans = solve_youtube(q, urls)
+
+    elif t == "EXCEL_ATTACHMENT":
+        ans = solve_excel_attachment(api_url, task_id, q)
         if not ans:
             ans = solve_general_search(q)
 
-    elif t == "WIKI_META":
-        # 위키 메타 질문은 구조가 다양해서 검색형으로 보내되,
-        # 위키 API를 섞어서 정확도 높이는 방향(추후 확장 지점)
-        ans = solve_general_search(q)
-
-    elif t == "YOUTUBE":
-        # 자막 기반으로만 처리. 자막이 없으면 빈 문자열로 끝.
-        ans = solve_youtube(q, urls)
+    elif t == "IMAGE_CHESS":
+        ans = solve_image_chess(api_url, task_id, q)
         if not ans:
-            # 유튜브가 "화면에 보이는 것"을 묻는 경우 자막으로는 불가.
-            # 여기서 억지로 검색해도 오답률이 높아짐 → 빈 문자열 전략이 더 낫다.
-            ans = ""
+            ans = solve_general_search(q)
 
     else:
         ans = solve_general_search(q)
@@ -621,9 +561,6 @@ def node_finalize(state: AgentState) -> AgentState:
 
 
 def build_graph():
-    """
-    START -> init -> urls -> classify -> solve -> finalize -> END
-    """
     g = StateGraph(AgentState)
     g.add_node("init", node_init)
     g.add_node("urls", node_urls)
@@ -637,6 +574,7 @@ def build_graph():
     g.add_edge("classify", "solve")
     g.add_edge("solve", "finalize")
     g.add_edge("finalize", END)
+
     return g.compile()
 
 
@@ -644,19 +582,25 @@ GRAPH = build_graph()
 
 
 # =========================================================
-# 13) Public API: app.py에서 import하는 BasicAgent
+# Public API
 # =========================================================
 class BasicAgent:
     def __init__(self):
-        # 모듈 import 시 그래프는 이미 컴파일되어 있음
-        print("BasicAgent initialized (Router + Solvers, no tool-calling)")
+        print("✅ BasicAgent initialized (attachments-enabled, no tool-calling)")
 
     def __call__(self, question: str, **kwargs) -> str:
         """
-        app.py가 task_id 같은 kwargs를 넘겨도 무시하고 question만 처리한다.
+        app.py에서 넘길 수 있는 kwargs:
+        - task_id: str
+        - api_url: str  (DEFAULT_API_URL)
         """
+        task_id = str(kwargs.get("task_id") or "")
+        api_url = str(kwargs.get("api_url") or os.getenv("GAIA_API_URL") or "")
+
         state: AgentState = {
             "question": question,
+            "task_id": task_id,
+            "api_url": api_url,
             "task_type": "",
             "urls": [],
             "context": "",

requirements.txt

@@ -6,4 +6,6 @@ langchain-core
 ddgs
 youtube-transcript-api
 beautifulsoup4
-lxml
+lxml
+pandas
+openpyxl