youlbot/services/agent/tools.py

from datetime import date, datetime

from langchain_core.tools import tool


def make_vision_tool(vision_model, image_path: str):
    """현재 요청에 첨부된 이미지를 분석하는 도구."""

    @tool
    def analyze_image(prompt: str = "이 이미지를 한국어로 자세히 설명해줘.") -> str:
        """첨부된 이미지를 분석한다. 이미지 속 음식, 문서, 사람, 사물 등을 파악할 때 사용하세요."""
        return vision_model.analyze(image_path, prompt)

    return analyze_image


@tool
def get_current_date() -> str:
    """오늘 날짜를 반환합니다. 나이 계산, 날짜 비교 등 현재 날짜가 필요할 때 반드시 먼저 호출하세요."""
    return date.today().isoformat()


@tool
def web_search(query: str) -> str:
    """최신 뉴스, 금리, 육아 정책 등 실시간 정보가 필요할 때 사용하세요. 저장된 문서에 없는 최신 정보를 검색합니다."""
    from duckduckgo_search import DDGS
    with DDGS() as ddgs:
        results = list(ddgs.text(query, max_results=5))
    if not results:
        return "검색 결과가 없습니다."
    return "\n\n".join(
        f"[{r['title']}]\n{r['body']}\n출처: {r['href']}"
        for r in results
    )


def make_retriever_tool(retriever_service):
    """retriever_service.search()를 사용하는 검색 Tool (Reranker 자동 적용)."""

    @tool
    def search_documents(query: str) -> str:
        """등록된 문서(논문, 육아 가이드, 금융 자료 등)에서 관련 정보를 검색합니다.
        육아·금융 관련 질문이 오면 자신의 지식으로 답하기 전에 반드시 이 도구를 먼저 호출하세요.
        등록된 문서가 없거나 검색 결과가 없을 때만 자신의 학습 지식을 보조적으로 활용합니다."""
        docs = retriever_service.search(query)
        if not docs:
            return "관련 문서를 찾을 수 없습니다."
        return "\n\n".join(
            f"[문서 {i + 1}]\n{doc.page_content}" for i, doc in enumerate(docs)
        )

    return search_documents


def make_memory_tools(profile_repo, user_id: str = "default"):
    """사용자 정보 저장/조회 Tool 쌍을 반환한다."""

    @tool
    def remember_user_info(key: str, value: str) -> str:
        """사용자 정보를 영구 저장합니다. 다음 대화에도 기억해야 할 정보를 저장하세요.
        - 아이 생년월일은 전체 날짜로 저장하세요. 날짜를 모르면 연도만이라도 저장하세요.
          예: key='첫째_이름' value='신도율', key='첫째_생년월일' value='2020년 6월 19일'
          연도만 알 경우: key='첫째_생년' value='2020'
        - 기타 key 예시: 재정_목표, 거주지, 직업, 자녀수"""
        profile_repo.remember(key, value, user_id=user_id)
        return f"'{key}' 정보를 기억했습니다: {value}"

    @tool
    def recall_user_info(key: str) -> str:
        """이전 대화에서 저장한 사용자 정보를 조회합니다."""
        value = profile_repo.recall(key, user_id=user_id)
        return value if value is not None else f"'{key}'에 대한 저장된 정보가 없습니다."

    return remember_user_info, recall_user_info


def make_reminder_tools(reminder_repo, user_id: str = "default"):
    """알림 등록/조회 Tool 쌍을 반환한다."""

    @tool
    def set_reminder(remind_date: str, message: str) -> str:
        """특정 날짜에 텔레그램으로 알림을 보냅니다.
        예방접종, 병원 예약, 기념일 등 기억해야 할 날짜를 등록하세요.
        - remind_date: 알림 날짜 (YYYY-MM-DD 형식). 날짜를 모르면 get_current_date를 먼저 호출하세요.
        - message: 알림 내용 (구체적으로 작성)
        등록 시 D-7(7일 전), D-1(하루 전), D-0(당일) 세 번 알림이 발송됩니다."""
        try:
            parsed = datetime.strptime(remind_date, "%Y-%m-%d").date()
        except ValueError:
            return f"날짜 형식이 잘못되었습니다. YYYY-MM-DD 형식으로 입력해 주세요. (예: 2026-07-01)"
        reminder_repo.add(user_id, parsed, message)
        return f"알림이 등록되었습니다. {remind_date}에 '{message}' 알림을 보내드릴게요."

    @tool
    def list_reminders() -> str:
        """등록된 예정 알림 목록을 조회합니다. (향후 30일 이내)"""
        items = reminder_repo.get_upcoming(user_id, days_ahead=30)
        if not items:
            return "등록된 예정 알림이 없습니다."
        lines = [f"- {r['remind_date']}: {r['message']}" for r in items]
        return "등록된 알림 목록:\n" + "\n".join(lines)

    return set_reminder, list_reminders


def make_search_tool(retriever_service, source_buffer: list | None = None):
    """RetrieverService를 클로저로 감싼 문서 검색 Tool을 반환합니다.

    source_buffer가 주어지면 검색된 문서의 메타데이터(source, page)를 누적 저장합니다.
    """

    @tool
    def search_documents(query: str) -> str:
        """등록된 문서(논문, 육아 가이드, 금융 자료 등)에서 관련 정보를 검색합니다.
        육아·금융 관련 질문이 오면 자신의 지식으로 답하기 전에 반드시 이 도구를 먼저 호출하세요.
        등록된 문서가 없거나 검색 결과가 없을 때만 자신의 학습 지식을 보조적으로 활용합니다."""
        docs = retriever_service.search(query)

        if source_buffer is not None:
            for doc in docs:
                src = doc.metadata.get("source", "")
                page = doc.metadata.get("page", None)
                if src:
                    entry = {"source": src}
                    if page is not None:
                        entry["page"] = page + 1  # 0-indexed → 1-indexed
                    if entry not in source_buffer:
                        source_buffer.append(entry)

        if not docs:
            return "관련 문서를 찾을 수 없습니다."
        return "\n\n".join(
            f"[문서 {i + 1}]\n{doc.page_content}" for i, doc in enumerate(docs)
        )

    return search_documents