from datetime import date 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_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