From 204567feae2489e05e0cf25f1703e2bdccb5bbac Mon Sep 17 00:00:00 2001 From: tomchccom Date: Wed, 1 Apr 2026 23:42:26 +0900 Subject: [PATCH 1/5] =?UTF-8?q?feat(ai):=20LangGraph=20=EC=A0=84=EB=9E=B5?= =?UTF-8?q?=20=ED=8F=89=EA=B0=80=20=ED=8C=8C=EC=9D=B4=ED=94=84=EB=9D=BC?= =?UTF-8?q?=EC=9D=B8=20=EA=B3=A0=EB=8F=84=ED=99=94=20=EB=B0=8F=20=EC=95=88?= =?UTF-8?q?=EC=A0=95=EC=84=B1=20=ED=99=95=EB=B3=B4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - LangChain with_retry 체이닝 문법 오류 해결로 LLM 네트워크 단절 이슈 대응 - JD_URL_CACHE, fake-useragent, tenacity 익스포넨셜 백오프를 결합한 스크래핑 429 차단 방어막 구축 - GuideOutput(Pydantic) 객체 및 Self-Correction 프롬프트 추가 (문항 의도에 따른 전략 강제 매핑 및 논리 일관성 검증) --- myeongsung/resume_strategist.py | 346 ++++++++++++++++++++++++++++++++ 1 file changed, 346 insertions(+) create mode 100644 myeongsung/resume_strategist.py diff --git a/myeongsung/resume_strategist.py b/myeongsung/resume_strategist.py new file mode 100644 index 0000000..bb29131 --- /dev/null +++ b/myeongsung/resume_strategist.py @@ -0,0 +1,346 @@ +import os +import json +import time +import random +from dotenv import load_dotenv +from typing import List, Dict, Any, TypedDict, Literal, Optional + +# .env 환경변수를 자동으로 불러옵니다. +load_dotenv() + +from pydantic import BaseModel, Field +from langchain_openai import ChatOpenAI +from langchain_core.prompts import ChatPromptTemplate +from langgraph.graph import StateGraph, START, END + +# ========================================== +# [Fix 2] 간이 캐싱 메모리 (동일 URL 크롤링 회피) +# ========================================== +JD_URL_CACHE: Dict[str, str] = {} + + +# ========================================== +# 0. Custom Exceptions +# ========================================== +from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type +class RateLimitException(Exception): + pass + +# ========================================== +# 1. State 정의 +# ========================================== +class AgentState(TypedDict): + jd_markdown: str + jd_url: Optional[str] + experiences: List[Dict[str, Any]] + prompts: List[str] + jd_context: Dict[str, Any] + placements: List[Dict[str, Any]] + remaining_indices: List[int] + errors: List[str] + +# ========================================== +# 2. 구조화된 출력을 위한 Pydantic 모델 +# ========================================== +class JDAnalysis(BaseModel): + opportunities: str = Field(description="Opportunities (O)") + threats: str = Field(description="Threats (T)") + +class StrategyScore(BaseModel): + SO: int = Field(ge=0, le=100) + ST: int = Field(ge=0, le=100) + WO: int = Field(ge=0, le=100) + WT: int = Field(ge=0, le=100) + +class ScoredExperience(BaseModel): + id: int = Field(...) + scores: StrategyScore = Field(...) + primary_strategy: Literal["SO", "ST", "WO", "WT"] = Field(...) + reasoning: str = Field(...) + +class ExperienceScoringList(BaseModel): + scored_experiences: List[ScoredExperience] = Field(...) + + +# ========================================== +# 3. LangGraph 노드 함수 구현 +# ========================================== + +def jd_ingestion_router(state: AgentState) -> Literal["Upstage_Parse_Node", "Cache_Hit_Node", "Web_Scraping_Node"]: + """Node: 파싱 노드 라우팅 로직 (Cache 우선적 확인)""" + if state.get("jd_markdown") and state.get("jd_markdown").strip(): + return "Upstage_Parse_Node" + elif state.get("jd_url") and state.get("jd_url").strip(): + url = state["jd_url"] + # [Fix 2] 이전에 들어온 URL인지 체크하여 Conditional Edge 분기 + if url in JD_URL_CACHE: + return "Cache_Hit_Node" + return "Web_Scraping_Node" + else: + return "Upstage_Parse_Node" + +def upstage_parse_node(state: AgentState) -> AgentState: + return state + +def cache_hit_node(state: AgentState) -> AgentState: + """Node: 이전에 저장된 JD 메모리를 재사용하여 통신 횟수를 절감함""" + url = state.get("jd_url") + if url in JD_URL_CACHE: + print(f"[*] {url} 발견! 캐시에서 결과(마크다운)를 불러옵니다 (크롤링 건너뜀).") + state["jd_markdown"] = JD_URL_CACHE[url] + return state + +# [Fix 3] 429 에러 밸생 시 지수 백오프 로직 (5초->10초 늘려가며 최대 3회 시도) +@retry( + stop=stop_after_attempt(3), + wait=wait_exponential(multiplier=5, min=5, max=20), + retry=retry_if_exception_type(RateLimitException) +) +def fetch_html_with_retry(url: str) -> str: + import requests + from fake_useragent import UserAgent + + # [Fix 1] fake-useragent와 랜덤 Sleep 처리 + ua = UserAgent() + headers = { + "User-Agent": ua.random, + "Accept-Language": "ko-KR,ko;q=0.9,en-US;q=0.8,en;q=0.7", + "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8" + } + + delay = random.uniform(1.5, 3.5) + print(f"[*] 봇 탐지 회피용 대기 진행... ({delay:.2f}초 Sleep)") + time.sleep(delay) + + response = requests.get(url, headers=headers, timeout=15) + + if response.status_code == 429: + print(f"⚠️ [HTTP 429 에러] 대상 서버에서 너무 많은 요청을 감지함! 지수 백오프(Exponential Backoff) 재시도 작동 준비...") + raise RateLimitException(f"429 Too Many Requests: {url}") + + response.raise_for_status() + return response.text + +def web_scraping_node(state: AgentState) -> AgentState: + url = state.get("jd_url") + if not url: + return state + + try: + from bs4 import BeautifulSoup + print(f"[*] {url} 라이브 웹 스크래핑 시도...") + + # 재시도/헤더/지연로직이 포함된 안전한 페치(fetch) 함수 호출 + html_text = fetch_html_with_retry(url) + + soup = BeautifulSoup(html_text, "html.parser") + for tag in soup(["script", "style", "noscript", "header", "footer", "nav", "aside"]): + tag.extract() + + raw_text = soup.get_text(separator="\n", strip=True) + if len(raw_text) > 30000: + raw_text = raw_text[:30000] + + llm = ChatOpenAI(model="gpt-4o", temperature=0) + prompt = ChatPromptTemplate.from_messages([ + ("system", "당신은 웹 페이지에서 채용 정보만 추출하는 전문 데이터 엔지니어입니다. 제공된 텍스트에서 회사 소개, 주요 업무, 자격 요건, 우대 사항, 복지 혜택에 해당하는 내용만 마크다운 형식으로 요약하세요. 채용과 관련 없는 광고, 사이트 메뉴, 법적 고지 등은 반드시 제외하십시오."), + ("user", "원문 텍스트:\n{raw_text}") + ]) + + clean_md = (prompt | llm).with_retry(stop_after_attempt=3).invoke({"raw_text": raw_text}).content + state["jd_markdown"] = clean_md + + # [Fix 2] 분석 성공 시 캐시(딕셔너리)에 덮어쓰기 저장 + JD_URL_CACHE[url] = clean_md + + except Exception as e: + state["errors"].append(f"[Web Scraping Error] {str(e)}") + state["jd_markdown"] = f"# 스크래핑 및 LLM 정제 실패\n\nURL: {url}\n오류 내용: {str(e)}" + + return state + + +def jd_structural_analyzer(state: AgentState) -> AgentState: + llm = ChatOpenAI(model="gpt-4o", temperature=0) + + prompt = ChatPromptTemplate.from_messages([ + ("system", "당신은 채용 공고의 행간을 읽는 전략가입니다. 제공된 마크다운 텍스트에서 다음을 도출하세요.\n" + "Opportunities (O): 직무의 비전, 핵심 우대사항, 기업의 성장 동력.\n" + "Threats (T): 직무 수행의 난관, 기술적 복잡성, 업계의 페인 포인트.\n" + "결과는 JSON 구조로 저장하세요."), + ("user", "JD Markdown:\n{jd_markdown}") + ]) + + chain = (prompt | llm.with_structured_output(JDAnalysis)).with_retry(stop_after_attempt=3) + try: + result = chain.invoke({"jd_markdown": state.get("jd_markdown", "")}) + state["jd_context"] = result.model_dump() + except Exception as e: + state["errors"].append(f"[JD Analysis API Error] {str(e)}") + + return state + + +def swot_strategy_scorer(state: AgentState) -> AgentState: + llm = ChatOpenAI(model="gpt-4o", temperature=0) + + prompt = ChatPromptTemplate.from_messages([ + ("system", "JD의 O, T와 각 경험의 S(강점), W(약점)를 대조하여 4대 전략 점수를 매기세요.\n" + "SO: 강점으로 기회 선점 / ST: 강점으로 위협 돌파 / WO: 약점을 기회로 상쇄 / WT: 약점을 인정하고 보완.\n" + "가장 점수가 높은 '극단적 전략'을 해당 경험의 대표 전략으로 정의하세요."), + ("user", "JD Context:\nOpportunities: {opportunities}\nThreats: {threats}\n\nExperiences:\n{experiences}") + ]) + + chain = (prompt | llm.with_structured_output(ExperienceScoringList)).with_retry(stop_after_attempt=3) + try: + experiences_json = json.dumps(state["experiences"], ensure_ascii=False) + result = chain.invoke({ + "opportunities": state.get("jd_context", {}).get("opportunities", ""), + "threats": state.get("jd_context", {}).get("threats", ""), + "experiences": experiences_json + }) + + score_map = {item.id: item for item in result.scored_experiences} + for exp in state["experiences"]: + score_data = score_map.get(exp["id"]) + if score_data: + exp["scores"] = score_data.scores.model_dump() + exp["primary_strategy"] = score_data.primary_strategy + exp["strategy_reasoning"] = score_data.reasoning + + except Exception as e: + state["errors"].append(f"[SWOT Scoring API Error] {str(e)}") + + return state + + +def sequential_strategic_placer(state: AgentState) -> AgentState: + llm = ChatOpenAI(model="gpt-4o", temperature=0.2) + experiences = state["experiences"] + prompts = state["prompts"] + + remaining_indices = state.get("remaining_indices", []) + if not remaining_indices: + remaining_indices = list(range(len(experiences))) + + placements = [] + priority_weight = {"상": 2, "중": 1, "하": 0} + + # [Fix] 가이드라인과 매칭 논리를 엄격히 통제할 구조화된 출력 모델 설계 + class GuideOutput(BaseModel): + reasoning: str = Field(description="전략 선정 이유. 반드시 '분석 결과 [선택한 전략명] 전략이 가장 적합합니다'라는 문장으로 시작할 것") + writing_guide: str = Field(description="실제 자소서 작성 가이드라인 및 핵심 키워드 정리") + + for prompt_text in prompts: + if not remaining_indices: + placements.append({ + "question": prompt_text, + "experience_title": "경험 부족", + "selected_strategy": "N/A", + "reasoning": "할당 가능한 여유 경험이 부족합니다.", + "writing_guide": "N/A" + }) + continue + + # [Fix] 문항의 의도를 최우선으로 고려하는 제약 조건 추가 + intent_prompt = ChatPromptTemplate.from_messages([ + ("system", "다음 자소서 문항의 직무 연관성과 '질문 의도'를 최우선으로 분석하여 가장 적절한 전략 방향(SO, ST, WO, WT 중 택1) 하나만 답변하세요.\n" + "제약 조건: 약점/극복 문항 -> WO 또는 WT (성장 가능성 어필) / 성과/도전/문제해결 문항 -> SO 또는 ST (해결사 능력 어필).\n" + "결과는 반드시 'SO', 'ST', 'WO', 'WT' 중 하나의 문자열로만 출력하세요. 기본값은 'SO'입니다."), + ("user", "자소서 문항: {prompt_text}") + ]) + + try: + target_strategy = (intent_prompt | llm).with_retry(stop_after_attempt=3).invoke({"prompt_text": prompt_text}).content.strip() + if target_strategy not in ["SO", "ST", "WO", "WT"]: + target_strategy = "SO" + except Exception: + target_strategy = "SO" + + best_exp_idx = -1 + max_score = -1 + best_priority_val = -1 + + for idx in remaining_indices: + exp = experiences[idx] + score = exp.get("scores", {}).get(target_strategy, 0) + p_val = priority_weight.get(exp.get("priority", "하"), 0) + + if max_score != -1 and abs(max_score - score) < 5: + if p_val > best_priority_val: + max_score = score + best_exp_idx = idx + best_priority_val = p_val + elif score > max_score: + max_score = score + best_exp_idx = idx + best_priority_val = p_val + + if best_exp_idx != -1: + best_exp = experiences[best_exp_idx] + remaining_indices.remove(best_exp_idx) + + # [Fix] Self-Correction을 유도하는 프롬프트 적용 + guide_prompt = ChatPromptTemplate.from_messages([ + ("system", "자소서 작성 가이드라인과 매칭 논리를 작성해주세요.\n" + "1. reasoning 필드는 서두에 반드시 '분석 결과 [{target_strategy}] 전략이 가장 적합합니다'라는 문장을 강제로 포함하여 AI 스스로 자신의 논리적 일관성을 확인(Self-Correction)하세요. 결론 내린 전략 명칭은 반드시 {target_strategy} 와 100% 일치해야 합니다.\n" + "2. writing_guide 필드는 경험의 내용과 매칭 논리를 연결하여, 서술 시 강조해야 할 핵심 키워드 및 흐름을 분석하세요."), + ("user", "문항: {prompt_text}\n전략: {target_strategy}\n경험 명: {exp_title}\n경험 내용: {exp_content}\n경험의 원래 평가이유: {reasoning}") + ]) + try: + guide_chain = (guide_prompt | llm.with_structured_output(GuideOutput)).with_retry(stop_after_attempt=3) + guide_result = guide_chain.invoke({ + "prompt_text": prompt_text, + "target_strategy": target_strategy, + "exp_title": best_exp["title"], + "exp_content": best_exp["content"], + "reasoning": best_exp.get("strategy_reasoning", "") + }) + final_reasoning = guide_result.reasoning + final_guide = guide_result.writing_guide + except Exception: + final_reasoning = f"분석 결과 [{target_strategy}] 전략이 가장 적합합니다. (세부 매칭 논리 생성 실패)" + final_guide = "가이드 생성 실패" + + placements.append({ + "question": prompt_text, + "experience_id": best_exp["id"], + "experience_title": best_exp["title"], + "selected_strategy": target_strategy, + "reasoning": final_reasoning, + "writing_guide": final_guide + }) + + state["placements"] = placements + state["remaining_indices"] = remaining_indices + return state + + +# ========================================== +# 4. LangGraph 파이프라인 컴파일 +# ========================================== +def create_workflow() -> Any: + workflow = StateGraph(AgentState) + + # 노드 부착 + workflow.add_node("Upstage_Parse_Node", upstage_parse_node) + workflow.add_node("Cache_Hit_Node", cache_hit_node) + workflow.add_node("Web_Scraping_Node", web_scraping_node) + + workflow.add_node("JD_Structural_Analyzer", jd_structural_analyzer) + workflow.add_node("SWOT_Strategy_Scorer", swot_strategy_scorer) + workflow.add_node("Sequential_Strategic_Placer", sequential_strategic_placer) + + # 라우팅 + workflow.add_conditional_edges(START, jd_ingestion_router) + + # 순차 플로우 연결 + workflow.add_edge("Upstage_Parse_Node", "JD_Structural_Analyzer") + workflow.add_edge("Cache_Hit_Node", "JD_Structural_Analyzer") + workflow.add_edge("Web_Scraping_Node", "JD_Structural_Analyzer") + + workflow.add_edge("JD_Structural_Analyzer", "SWOT_Strategy_Scorer") + workflow.add_edge("SWOT_Strategy_Scorer", "Sequential_Strategic_Placer") + workflow.add_edge("Sequential_Strategic_Placer", END) + + return workflow.compile() From 6dd4b88ad62dec1574bc1bae8e2f15824f5ff2a8 Mon Sep 17 00:00:00 2001 From: tomchccom Date: Wed, 1 Apr 2026 23:42:26 +0900 Subject: [PATCH 2/5] =?UTF-8?q?test(mock):=20=EC=82=AC=EC=9A=A9=EC=9E=90?= =?UTF-8?q?=20=EC=8B=A4=EC=A0=9C=20=EA=B2=BD=ED=97=98=20=EB=8D=B0=EC=9D=B4?= =?UTF-8?q?=ED=84=B0=EB=A5=BC=20=ED=99=9C=EC=9A=A9=ED=95=9C=20API=20?= =?UTF-8?q?=ED=85=8C=EC=8A=A4=ED=8A=B8=20=ED=81=B4=EB=9D=BC=EC=9D=B4?= =?UTF-8?q?=EC=96=B8=ED=8A=B8=20=EB=A1=9C=EC=A7=81=20=EC=99=84=EC=84=B1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- myeongsung/mock_data_generator.py | 116 ++++++++++++++++++++++++++++++ 1 file changed, 116 insertions(+) create mode 100644 myeongsung/mock_data_generator.py diff --git a/myeongsung/mock_data_generator.py b/myeongsung/mock_data_generator.py new file mode 100644 index 0000000..d1e2a0e --- /dev/null +++ b/myeongsung/mock_data_generator.py @@ -0,0 +1,116 @@ +import json +import requests +import io +import time + +MOCK_JD_MARKDOWN = """ +# AI 플랫폼 백엔드 엔지니어 경력 채용 + +## 직무 비전 및 기회 (Opportunities) +- 대규모 언어 모델(LLM)과 RAG를 결합한 사내 및 기업용 AI 에이전트 서비스 구축 +- Vector DB (Qdrant, Pinecone 등) 설계 및 Hybrid Search 파이프라인 최적화 +- Spring Boot, FastAPI, Node.js 등을 활용한 안정적인 멀티 마이크로서비스 아키텍처(MSA) 운영 경험 제공 + +## 직무 수행의 난관 (Threats) +- 비정형 도메인 데이터 검색 시 나타나는 오분류 문제와 RAG의 Hallucination(환각) 리스크 최소화 역량 필수 +- 대용량 데이터 및 영상 분석 시 서버의 인프라 비용과 응답 지연(Latency)이 기하급수적으로 늘어나는 문제 +""" + +MOCK_EXPERIENCES = [ + { + "id": 1, + "title": "Fin-agent: 금융 특화 AI 에이전트 및 Hybrid Retrieval 구축", + "content": "미래에셋 공모전에서 금융 질의응답 신뢰성을 높이기 위해 Sparse(의미) 65% + Dense(유사도) 35% 가중치를 적용한 Hybrid Retrieval 시스템을 구축함. 계열사 오분류 문제를 해결하기 위해 Reranker와 LLM 검증 단계를 추가하여 검색 정확도를 극대화함.", + "tags": ["LangGraph", "RAG", "Hybrid Retrieval", "Python"], + "priority": "상" + }, + { + "id": 2, + "title": "Memoralaxy: GNN 기반 지식 추천 및 DAG 구조화", + "content": "비정형 문서에서 지식을 추출하고 GraphSAGE 모델을 통해 추천 시스템을 구축함. 특히 DFS 알고리즘을 활용해 관계 데이터의 사이클을 제거하고 강제로 DAG(방향성 비순환 그래프) 구조를 확보하여 학습 경로의 논리적 선후 관계를 보장함.", + "tags": ["GNN", "GraphSAGE", "Algorithm", "Python"], + "priority": "상" + }, + { + "id": 3, + "title": "YAR-YAR_BE: 비용 효율적 AI 릴스 분석 시스템", + "content": "숏폼 영상 분석 시 발생하는 고비용 문제를 해결하기 위해 FFmpeg로 핵심 프레임(3~5장)만 추출하여 분석하는 로직을 설계함. Spring-FastAPI-MySQL 멀티 컨테이너 환경을 Docker Compose와 GitHub Actions로 CI/CD 구축함.", + "tags": ["Spring Boot", "FastAPI", "Docker", "FFmpeg", "비용최적화"], + "priority": "상" + }, + { + "id": 4, + "title": "stopping: 판례 기반 스토킹 판별 RAG 시스템", + "content": "Qdrant Vector DB에 실제 판례를 저장하고 사용자의 상황과 유사한 데이터를 증강(RAG)하여 GPT-3.5가 법적 근거를 제시하도록 설계함. Spring 서버와 Python 임베딩 로직을 연동하여 실시간 유사도 추출 기능을 구현함.", + "tags": ["Vector DB", "Qdrant", "RAG", "Spring Boot"], + "priority": "중" + }, + { + "id": 5, + "title": "TOMO: 소셜 모임 서비스 인프라 및 CI/CD 운영", + "content": "Nginx 리버스 프록시 및 HTTPS 환경을 구축하고 AWS/Oracle Cloud를 병행하여 VM 인스턴스를 운영함. GitHub Actions를 통해 자동 배포 파이프라인을 구축하고 실제 유저 피드백 기반 릴리즈를 경험함.", + "tags": ["Infrastructure", "CI/CD", "Nginx", "AWS"], + "priority": "중" + }, + { + "id": 6, + "title": "Nomad: WebSocket 기반 실시간 채팅 및 AOP 예외처리", + "content": "WebSocket을 이용한 실시간 채팅과 세션 로그인을 구현함. 특히 AOP를 도입하여 서비스 단의 예외를 전역적으로 핸들링하고 클라이언트에게 표준화된 에러 응답을 전달하는 구조를 설계함.", + "tags": ["WebSocket", "AOP", "Spring Boot", "Java"], + "priority": "하" + } +] + +MOCK_PROMPTS = [ + "문항 1: 지원 직무와 관련하여 본인이 직면했던 가장 어려운 기술적 도전은 무엇이며, 이를 어떻게 해결했는지 기술하십시오.", + "문항 2: 기존의 방식에서 벗어나 효율성을 높이거나 비용을 절감했던 창의적인 문제 해결 경험이 있다면 기술하십시오.", + "문항 3: 본인이 가진 기술적 약점은 무엇이며, 이를 보완하기 위해 어떤 노력을 기울이고 있는지 실제 프로젝트 사례를 들어 기술하십시오." +] + +def run_test(): + url = "http://127.0.0.1:8000/analyze-and-place" + + print("\\n=== [실제 명성님의 데이터를 활용한 API 테스트 실행] ===") + + # 안정적인 테스트를 위해 사용자의 경험(RAG, AI 백엔드)과 정확히 매칭되는 가상의 JD PDF를 전송합니다. + dummy_pdf = io.BytesIO(MOCK_JD_MARKDOWN.encode("utf-8")) + files = { + "jd_pdf": ("mock_ai_jd.pdf", dummy_pdf, "application/pdf") + } + + data = { + "experiences_json": json.dumps(MOCK_EXPERIENCES, ensure_ascii=False), + "essay_prompts_json": json.dumps(MOCK_PROMPTS, ensure_ascii=False) + } + + print("🚀 FastAPI 서버로 분석을 요청합니다...") + start_time = time.time() + + try: + response = requests.post(url, files=files, data=data) + except Exception as e: + print("네트워크 또는 FastAPI 서버 연결 오류:", e) + return + + end_time = time.time() + + if response.status_code == 200: + result_json = response.json() + print(f"✅ 분석 완료! (소요 시간: {end_time - start_time:.2f}초)\\n") + print(json.dumps(result_json, ensure_ascii=False, indent=2)) + + # 보기 좋게 요약 출력 + print("\\n=======================================================") + for p in result_json.get("placements", []): + print(f"- [매칭 문항] : {p['question'][:30]}...") + print(f" [채택 경험] : {p['experience_title']}") + print(f" [전략 방향] : {p['selected_strategy']}") + print(f" [가이드라인]: {p['writing_guide']}") + print("-" * 55) + + else: + print(f"❌ 분석 실패 (Status Code: {response.status_code})") + print(response.text) + +if __name__ == "__main__": + run_test() From bcd8b973c5ac618824b41a120e84cedce029c947 Mon Sep 17 00:00:00 2001 From: tomchccom Date: Wed, 1 Apr 2026 23:42:26 +0900 Subject: [PATCH 3/5] =?UTF-8?q?chore:=20.gitignore=20=EB=B6=88=ED=95=84?= =?UTF-8?q?=EC=9A=94=ED=95=9C=20=EC=BA=90=EC=8B=9C=20=ED=99=95=EC=9E=A5?= =?UTF-8?q?=EC=9E=90=20=EA=B7=9C=EC=B9=99=20=EC=A0=95=EB=A6=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- myeongsung/.gitignore | 35 +++++++++++++++++++++++++++++++++++ 1 file changed, 35 insertions(+) create mode 100644 myeongsung/.gitignore diff --git a/myeongsung/.gitignore b/myeongsung/.gitignore new file mode 100644 index 0000000..7a85210 --- /dev/null +++ b/myeongsung/.gitignore @@ -0,0 +1,35 @@ +# Environments +.env +.env.* + +# MacOS +.DS_Store + +# Python +__pycache__/ +*.py[cod] +*$py.class +*.so + +# Virtual Environments +venv/ +.venv/ +env/ +ENV/ + +# Python packaging & distribution +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +*.egg-info/ +.installed.cfg +*.egg \ No newline at end of file From 106cf9b3794322471826ac11b562bd3e43df4bf6 Mon Sep 17 00:00:00 2001 From: tomchccom Date: Sun, 5 Apr 2026 12:10:18 +0900 Subject: [PATCH 4/5] feat(ai): fix SO/WO strategy bias in sequential_strategic_placer - Remove hardcoded 'SO' default in intent detection prompt - Expand intent categories from 2 to 6 types (achievement, crisis, growth, failure, values, motivation) - Extract _detect_intent_strategy() helper with improved prompt - Extract _score_based_fallback() helper using experience score totals instead of hardcoded 'SO' fallback - Add main_api.py FastAPI entrypoint - Add root .gitignore to exclude .cursor/ directory --- .gitignore | 1 + myeongsung/main_api.py | 110 ++++++++++++++++++++++++++++++++ myeongsung/resume_strategist.py | 100 +++++++++++++++++++++++++---- 3 files changed, 197 insertions(+), 14 deletions(-) create mode 100644 .gitignore create mode 100644 myeongsung/main_api.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..303a196 --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +.cursor/ diff --git a/myeongsung/main_api.py b/myeongsung/main_api.py new file mode 100644 index 0000000..be830fd --- /dev/null +++ b/myeongsung/main_api.py @@ -0,0 +1,110 @@ +from fastapi import FastAPI, UploadFile, File, Form, HTTPException, BackgroundTasks +from pydantic import BaseModel, Field, ValidationError +import json +from typing import List, Dict, Any, Optional +from dotenv import load_dotenv + +# 애플리케이션 시작 전 .env 환경변수를 자동으로 불러옵니다. +load_dotenv() + +from resume_strategist import create_workflow + +app = FastAPI( + title="Resume Strategist API", + description="JD 분석(PDF/URL) 및 경험 배치를 수행하는 LangGraph 기반 AI 에이전트 API", + version="1.1.0" +) + +workflow = create_workflow() + +class ExperienceInput(BaseModel): + id: int + title: str + content: str + tags: List[str] = [] + priority: str = Field(pattern="^(상|중|하)$", description="'상', '중', '하' 중 하나로 입력") + +class PlacementResult(BaseModel): + question: str + experience_id: Optional[int] = None + experience_title: str + selected_strategy: str + reasoning: str + writing_guide: str + +class PlacementResponse(BaseModel): + placements: List[PlacementResult] + errors: List[str] = [] + + +@app.post("/analyze-and-place", response_model=PlacementResponse) +async def analyze_and_place( + background_tasks: BackgroundTasks, + jd_pdf: Optional[UploadFile] = File(None, description="채용공고 원문 PDF 파일 (업스테이지 파싱용)"), + jd_url: Optional[str] = Form(None, description="채용공고 웹페이지 URL (웹 스크래핑용)"), + experiences_json: str = Form(..., description="사용자 경험 데이터 JSON 문자열"), + essay_prompts_json: str = Form(..., description="자소서 문항 리스트 JSON 문자열") +): + """ + JD PDF 혹은 URL 중 하나와, 경험 JSON 목록, 자소서 문항 배열을 받아 LangGraph를 이용해 자소서를 매핑합니다. + """ + + # [유효성 검사] PDF나 URL 중 최소 하나는 반드시 존재해야 함 + if not jd_pdf and not (jd_url and jd_url.strip()): + raise HTTPException( + status_code=400, + detail="jd_pdf (업로드 파일) 또는 jd_url 중 최소 하나는 필수적으로 제공되어야 합니다." + ) + + # 1. JSON 검증 + try: + raw_experiences = json.loads(experiences_json) + raw_prompts = json.loads(essay_prompts_json) + + validated_experiences = [] + for exp in raw_experiences: + validated_experiences.append(ExperienceInput(**exp).model_dump()) + + if not isinstance(raw_prompts, list): + raise ValueError("essay_prompts_json 필드는 문자열 배열 형태여야 합니다.") + + except json.JSONDecodeError: + raise HTTPException(status_code=400, detail="유효하지 않은 JSON 문자열입니다.") + except (ValidationError, ValueError) as e: + raise HTTPException(status_code=400, detail=f"입력 데이터 검증 실패: {str(e)}") + + + # 2. 우선순위 판별 및 처리 + # 두 값이 모두 들어올 경우 jd_pdf 분석 결과를 우선 사용 + jd_markdown = "" + if jd_pdf and jd_pdf.filename: + jd_content = await jd_pdf.read() + try: + # 실제 서비스시엔 바이너리(jd_content)를 Upstage API에 넘기고 반환된 마크다운을 씁니다. + jd_markdown = jd_content.decode("utf-8") + except UnicodeDecodeError: + jd_markdown = "# JD 텍스트 파싱 처리 (더미 마크다운. 실제론 Upstage API에서 넘어왔다고 가정)" + + + # 3. LangGraph 상태(State) 설정 + initial_state = { + "jd_markdown": jd_markdown, + "jd_url": jd_url, + "experiences": validated_experiences, + "prompts": raw_prompts, + "jd_context": {}, + "placements": [], + "remaining_indices": [], + "errors": [] + } + + # 4. 워크플로우 실행 + try: + final_state = workflow.invoke(initial_state) + except Exception as e: + raise HTTPException(status_code=500, detail=f"내부 파이프라인 실행 중 오류 발생: {str(e)}") + + return PlacementResponse( + placements=final_state.get("placements", []), + errors=final_state.get("errors", []) + ) diff --git a/myeongsung/resume_strategist.py b/myeongsung/resume_strategist.py index bb29131..31dafd3 100644 --- a/myeongsung/resume_strategist.py +++ b/myeongsung/resume_strategist.py @@ -214,6 +214,84 @@ def swot_strategy_scorer(state: AgentState) -> AgentState: return state +# ========================================== +# [개선] 문항 의도 기반 전략 감지 헬퍼 함수 +# - SO 기본값 편향 제거 +# - 6가지 문항 유형 커버 +# ========================================== +_STRATEGY_CHOICES = ["SO", "ST", "WO", "WT"] + +def _detect_intent_strategy(prompt_text: str, llm) -> tuple[str | None, bool]: + """자소서 문항 의도를 분석하여 최적 SWOT 전략을 반환합니다. + + Returns: + (target_strategy, fallback_used) + - target_strategy: 'SO' | 'ST' | 'WO' | 'WT' | None (None이면 폴백 필요) + - fallback_used: LLM 감지 실패 여부 + """ + intent_prompt = ChatPromptTemplate.from_messages([ + ("system", + "당신은 자소서 문항 의도 분석 전문가입니다.\n" + "아래 [문항 유형별 전략 매핑]을 기준으로 주어진 자소서 문항을 분석하여 " + "가장 적합한 SWOT 전략 하나를 선택하세요.\n\n" + "[문항 유형별 전략 매핑]\n" + "• 성과/도전/목표달성/리더십 문항 → SO\n" + " 예: '목표를 세우고 달성한 경험', '도전적인 사례', '리더로서의 경험', '성과를 낸 경험'\n\n" + "• 위기대응/경쟁상황/압박/기술적 난관 문항 → ST\n" + " 예: '어려운 상황에서 문제를 해결한 경험', '갈등·충돌을 해결한 경험', '실패 위기를 극복한 사례'\n\n" + "• 약점보완/성장/개선/협업·팀워크 문항 → WO\n" + " 예: '부족한 점을 보완한 경험', '피드백을 받아 성장한 경험', '팀원과 협력하여 성과를 낸 경험'\n\n" + "• 실패/한계 인정/반성/포기 경험 문항 → WT\n" + " 예: '가장 힘들었던 경험', '실패한 경험과 교훈', '포기했거나 한계를 직면한 경험'\n\n" + "• 가치관/신념/직업의식/인생관 문항 → SO\n" + " 예: '직업 가치관', '인생 좌우명', '가장 중요하게 여기는 것'\n\n" + "• 지원동기/직무이해/입사 후 포부 문항 → ST\n" + " 예: '지원 동기', '이 직무를 선택한 이유', '입사 후 목표 및 성장 계획'\n\n" + "중요: 반드시 SO, ST, WO, WT 중 정확히 하나만 출력하세요. " + "다른 텍스트, 설명, 구두점은 절대 포함하지 마세요."), + ("user", "자소서 문항: {prompt_text}") + ]) + + try: + raw = (intent_prompt | llm).with_retry(stop_after_attempt=3).invoke( + {"prompt_text": prompt_text} + ).content.strip().upper() + # 불필요한 구두점·공백 제거 후 유효성 검사 + cleaned = raw.replace(".", "").replace(",", "").replace("'", "").replace('"', "").strip() + if cleaned in _STRATEGY_CHOICES: + return cleaned, False + # LLM이 유효하지 않은 문자열 반환 → 폴백 필요 + return None, True + except Exception: + return None, True + + +def _score_based_fallback( + experiences: list, + remaining_indices: list, + priority_weight: dict, +) -> str: + """LLM 의도 감지 실패 시 경험 점수 합계 기반으로 최적 전략을 선택합니다. + + 각 후보 경험의 우선순위를 반영한 전략별 점수 총합을 계산, + 가장 높은 전략을 반환합니다. 점수가 모두 0이면 무작위 선택. + """ + strategy_totals: dict[str, float] = {s: 0.0 for s in _STRATEGY_CHOICES} + + for idx in remaining_indices: + exp = experiences[idx] + scores = exp.get("scores", {}) + # 우선순위 가중치: 상=3, 중=2, 하=1 (0이면 최소 1 보장) + p_val = priority_weight.get(exp.get("priority", "하"), 0) + 1 + for strategy in _STRATEGY_CHOICES: + strategy_totals[strategy] += scores.get(strategy, 0) * p_val + + if all(v == 0.0 for v in strategy_totals.values()): + return random.choice(_STRATEGY_CHOICES) + + return max(strategy_totals, key=lambda s: strategy_totals[s]) + + def sequential_strategic_placer(state: AgentState) -> AgentState: llm = ChatOpenAI(model="gpt-4o", temperature=0.2) experiences = state["experiences"] @@ -242,20 +320,14 @@ class GuideOutput(BaseModel): }) continue - # [Fix] 문항의 의도를 최우선으로 고려하는 제약 조건 추가 - intent_prompt = ChatPromptTemplate.from_messages([ - ("system", "다음 자소서 문항의 직무 연관성과 '질문 의도'를 최우선으로 분석하여 가장 적절한 전략 방향(SO, ST, WO, WT 중 택1) 하나만 답변하세요.\n" - "제약 조건: 약점/극복 문항 -> WO 또는 WT (성장 가능성 어필) / 성과/도전/문제해결 문항 -> SO 또는 ST (해결사 능력 어필).\n" - "결과는 반드시 'SO', 'ST', 'WO', 'WT' 중 하나의 문자열로만 출력하세요. 기본값은 'SO'입니다."), - ("user", "자소서 문항: {prompt_text}") - ]) - - try: - target_strategy = (intent_prompt | llm).with_retry(stop_after_attempt=3).invoke({"prompt_text": prompt_text}).content.strip() - if target_strategy not in ["SO", "ST", "WO", "WT"]: - target_strategy = "SO" - except Exception: - target_strategy = "SO" + # [개선] 6가지 문항 유형 기반 의도 감지 (SO 기본값 편향 제거) + target_strategy, fallback_used = _detect_intent_strategy(prompt_text, llm) + if fallback_used or target_strategy is None: + # LLM 감지 실패 시 경험 점수 합계 기반 폴백 (SO 하드코딩 제거) + target_strategy = _score_based_fallback(experiences, remaining_indices, priority_weight) + print(f"[*] 의도 감지 폴백 사용 → 점수 기반 전략: {target_strategy} (문항: '{prompt_text[:30]}...')") + else: + print(f"[*] 의도 감지 성공 → 전략: {target_strategy} (문항: '{prompt_text[:30]}...')") best_exp_idx = -1 max_score = -1 From aad3f59b1ca9f5c86497e89bde43283153de1b7c Mon Sep 17 00:00:00 2001 From: tomchccom Date: Sun, 5 Apr 2026 14:27:46 +0900 Subject: [PATCH 5/5] feat(ai): integrate STAR schema, persona-based framing & flat output (v1.2.0) - Adopted STAR schema for experience input (situation, task, action, result) - Implemented dynamic S/W framing based on user_persona (Fact-first principle) - Enforced Korean output for all AI-generated fields - Refactored response to flat structure for RDB compatibility (PostgreSQL/MySQL) - Enabled 1:N experience mapping & fixed Unicode escaping in API responses --- myeongsung/main_api.py | 85 ++++++++--- myeongsung/resume_strategist.py | 248 ++++++++++++++++++++------------ 2 files changed, 220 insertions(+), 113 deletions(-) diff --git a/myeongsung/main_api.py b/myeongsung/main_api.py index be830fd..beb63f9 100644 --- a/myeongsung/main_api.py +++ b/myeongsung/main_api.py @@ -1,7 +1,8 @@ -from fastapi import FastAPI, UploadFile, File, Form, HTTPException, BackgroundTasks +from fastapi import FastAPI, UploadFile, File, Form, HTTPException, BackgroundTasks, Response from pydantic import BaseModel, Field, ValidationError import json -from typing import List, Dict, Any, Optional +import uuid +from typing import List, Dict, Any, Optional, Union from dotenv import load_dotenv # 애플리케이션 시작 전 .env 환경변수를 자동으로 불러옵니다. @@ -12,25 +13,38 @@ app = FastAPI( title="Resume Strategist API", description="JD 분석(PDF/URL) 및 경험 배치를 수행하는 LangGraph 기반 AI 에이전트 API", - version="1.1.0" + version="1.2.0" ) workflow = create_workflow() +# ── STAR 경험 입력 스키마 ────────────────────────────────────── +class StarContent(BaseModel): + situation: str = Field(..., description="[S] 상황 - 어떤 배경/맥락에서 발생한 일인지") + task: str = Field(..., description="[T] 과제 - 내가 맡은 구체적 역할과 목표") + action: str = Field(..., description="[A] 행동 - 내가 취한 구체적 행동과 방법") + result: str = Field(..., description="[R] 결과 - 행동으로 얻은 성과 (수치 포함 권장)") + class ExperienceInput(BaseModel): - id: int - title: str - content: str - tags: List[str] = [] - priority: str = Field(pattern="^(상|중|하)$", description="'상', '중', '하' 중 하나로 입력") + id: Optional[str] = Field( + None, + description="경험 고유 ID (미입력 시 UUID 자동 생성)" + ) + title: str = Field(..., description="경험 제목") + priority: str = Field(..., pattern="^(상|중|하)$", description="경험 중요도: 상/중/하") + tags: List[str] = Field(default=[], description="기술/역량 태그 (선택, 추후 AI 자동 태깅)") + star: StarContent = Field(..., description="STAR 형식 경험 본문") +# ── 응답 스키마 (플랫 구조) ── class PlacementResult(BaseModel): - question: str - experience_id: Optional[int] = None - experience_title: str - selected_strategy: str - reasoning: str - writing_guide: str + essay_question: str = Field(..., description="자소서 문항 원문") + matched_experience_id: Optional[Union[str, int]] = Field(None, description="매핑된 경험 ID (문자열 혹은 숫자)") + matched_experience_title: str = Field(..., description="매핑된 경험 제목") + strategy: str = Field(..., description="선택된 SWOT 전략 (SO/ST/WO/WT/N/A)") + jd_targeting: str = Field(..., description="[JD 타겟팅] JD에서 설정한 O/T 근거") + dynamic_framing: str = Field(..., description="[동적 프레이밍] 페르소나 기반 S/W 해석") + strategy_derivation: str = Field(..., description="[전략 도출] 전략 선택 최종 논증") + writing_guide: str = Field(..., description="자소서 작성 가이드라인 및 핵심 키워드") class PlacementResponse(BaseModel): placements: List[PlacementResult] @@ -43,7 +57,8 @@ async def analyze_and_place( jd_pdf: Optional[UploadFile] = File(None, description="채용공고 원문 PDF 파일 (업스테이지 파싱용)"), jd_url: Optional[str] = Form(None, description="채용공고 웹페이지 URL (웹 스크래핑용)"), experiences_json: str = Form(..., description="사용자 경험 데이터 JSON 문자열"), - essay_prompts_json: str = Form(..., description="자소서 문항 리스트 JSON 문자열") + essay_prompts_json: str = Form(..., description="자소서 문항 리스트 JSON 문자열"), + user_persona: str = Form("", description="지원자 성향/가치관 (예: '빠른 실행과 피보팅을 중시하는 개발자'). 동적 S/W 프레이밍에 사용됩니다."), ): """ JD PDF 혹은 URL 중 하나와, 경험 JSON 목록, 자소서 문항 배열을 받아 LangGraph를 이용해 자소서를 매핑합니다. @@ -56,18 +71,39 @@ async def analyze_and_place( detail="jd_pdf (업로드 파일) 또는 jd_url 중 최소 하나는 필수적으로 제공되어야 합니다." ) - # 1. JSON 검증 + # 1. JSON 검증 및 STAR → 내부 포맷 변환 try: raw_experiences = json.loads(experiences_json) raw_prompts = json.loads(essay_prompts_json) - + validated_experiences = [] for exp in raw_experiences: - validated_experiences.append(ExperienceInput(**exp).model_dump()) - + parsed = ExperienceInput(**exp) + + # UUID 자동 생성 (미입력 시) + exp_id = parsed.id or str(uuid.uuid4()) + + # STAR → LLM용 content 문자열 변환 + s = parsed.star + content = ( + f"[상황] {s.situation}\n" + f"[과제] {s.task}\n" + f"[행동] {s.action}\n" + f"[결과] {s.result}" + ) + + validated_experiences.append({ + "id": exp_id, + "title": parsed.title, + "priority": parsed.priority, + "tags": parsed.tags, + "content": content, # 내부 LLM 처리용 + "star": s.model_dump(), # 원본 보존 (추후 DB 저장용) + }) + if not isinstance(raw_prompts, list): raise ValueError("essay_prompts_json 필드는 문자열 배열 형태여야 합니다.") - + except json.JSONDecodeError: raise HTTPException(status_code=400, detail="유효하지 않은 JSON 문자열입니다.") except (ValidationError, ValueError) as e: @@ -92,6 +128,7 @@ async def analyze_and_place( "jd_url": jd_url, "experiences": validated_experiences, "prompts": raw_prompts, + "user_persona": user_persona, "jd_context": {}, "placements": [], "remaining_indices": [], @@ -104,7 +141,13 @@ async def analyze_and_place( except Exception as e: raise HTTPException(status_code=500, detail=f"내부 파이프라인 실행 중 오류 발생: {str(e)}") - return PlacementResponse( + # [개선] 한글 유니코드 이스케이프 방지 (ensure_ascii=False 적용) + final_response = PlacementResponse( placements=final_state.get("placements", []), errors=final_state.get("errors", []) + ).model_dump() + + return Response( + content=json.dumps(final_response, ensure_ascii=False), + media_type="application/json" ) diff --git a/myeongsung/resume_strategist.py b/myeongsung/resume_strategist.py index 31dafd3..5e9e327 100644 --- a/myeongsung/resume_strategist.py +++ b/myeongsung/resume_strategist.py @@ -3,7 +3,7 @@ import time import random from dotenv import load_dotenv -from typing import List, Dict, Any, TypedDict, Literal, Optional +from typing import List, Dict, Any, TypedDict, Literal, Optional, Union # .env 환경변수를 자동으로 불러옵니다. load_dotenv() @@ -34,9 +34,10 @@ class AgentState(TypedDict): jd_url: Optional[str] experiences: List[Dict[str, Any]] prompts: List[str] + user_persona: str # 지원자 성향/가치관 (동적 S/W 프레이밍용) jd_context: Dict[str, Any] placements: List[Dict[str, Any]] - remaining_indices: List[int] + remaining_indices: List[int] # 하위 호환 유지 (신규 로직에서는 미사용) errors: List[str] # ========================================== @@ -183,23 +184,34 @@ def jd_structural_analyzer(state: AgentState) -> AgentState: def swot_strategy_scorer(state: AgentState) -> AgentState: llm = ChatOpenAI(model="gpt-4o", temperature=0) - + user_persona = state.get("user_persona", "") or "별도 성향 정보 없음" + prompt = ChatPromptTemplate.from_messages([ - ("system", "JD의 O, T와 각 경험의 S(강점), W(약점)를 대조하여 4대 전략 점수를 매기세요.\n" - "SO: 강점으로 기회 선점 / ST: 강점으로 위협 돌파 / WO: 약점을 기회로 상쇄 / WT: 약점을 인정하고 보완.\n" - "가장 점수가 높은 '극단적 전략'을 해당 경험의 대표 전략으로 정의하세요."), - ("user", "JD Context:\nOpportunities: {opportunities}\nThreats: {threats}\n\nExperiences:\n{experiences}") + ("system", + "JD의 O, T와 각 경험의 S(강점), W(약점)를 대조하여 4대 전략 점수를 매기세요.\n" + "SO: 강점으로 기회 선점 / ST: 강점으로 위협 돌파 / WO: 약점을 기회로 상쇄 / WT: 약점을 인정하고 보완.\n" + "가장 점수가 높은 '극단적 전략'을 해당 경험의 대표 전략으로 정의하세요.\n\n" + "[동적 S/W 프레이밍 규칙]\n" + "아래 User Persona를 S/W 분류의 해석 필터로 사용하세요.\n" + "단, Persona와 경험 팩트(content)가 충돌하면 반드시 팩트를 우선합니다.\n" + "예: Persona가 '끈기'이지만 경험에 중도 포기가 명시된 경우 → W로 분류, WT 점수를 높이세요.\n\n" + "한국어 구사: 모든 텍스트 출력(reasoning 등)을 반드시 한국어로 작성하세요."), + ("user", + "User Persona: {user_persona}\n" + "JD Context:\nOpportunities: {opportunities}\nThreats: {threats}\n\n" + "Experiences:\n{experiences}") ]) - + chain = (prompt | llm.with_structured_output(ExperienceScoringList)).with_retry(stop_after_attempt=3) try: experiences_json = json.dumps(state["experiences"], ensure_ascii=False) result = chain.invoke({ + "user_persona": user_persona, "opportunities": state.get("jd_context", {}).get("opportunities", ""), "threats": state.get("jd_context", {}).get("threats", ""), "experiences": experiences_json }) - + score_map = {item.id: item for item in result.scored_experiences} for exp in state["experiences"]: score_data = score_map.get(exp["id"]) @@ -207,10 +219,10 @@ def swot_strategy_scorer(state: AgentState) -> AgentState: exp["scores"] = score_data.scores.model_dump() exp["primary_strategy"] = score_data.primary_strategy exp["strategy_reasoning"] = score_data.reasoning - + except Exception as e: state["errors"].append(f"[SWOT Scoring API Error] {str(e)}") - + return state @@ -293,98 +305,150 @@ def _score_based_fallback( def sequential_strategic_placer(state: AgentState) -> AgentState: + """자소서 문항별 최적 경험을 전략적으로 배치합니다. + + 주요 변경사항: + - 1:N 매핑 허용: 동일 경험을 여러 문항에 재사용 가능 + - user_persona 기반 동적 S/W 해석 + - 팩트-페르소나 충돌 시 팩트 우선 (환각 방지) + - 3단 reasoning: [JD 타겟팅] / [동적 프레이밍] / [전략 도출] + - 적합 경험 없을 때 N/A 반환 (억지 매핑 금지) + """ llm = ChatOpenAI(model="gpt-4o", temperature=0.2) experiences = state["experiences"] prompts = state["prompts"] - - remaining_indices = state.get("remaining_indices", []) - if not remaining_indices: - remaining_indices = list(range(len(experiences))) - + user_persona = state.get("user_persona", "") or "별도 성향 정보 없음" + jd_context = state.get("jd_context", {}) placements = [] - priority_weight = {"상": 2, "중": 1, "하": 0} - - # [Fix] 가이드라인과 매칭 논리를 엄격히 통제할 구조화된 출력 모델 설계 - class GuideOutput(BaseModel): - reasoning: str = Field(description="전략 선정 이유. 반드시 '분석 결과 [선택한 전략명] 전략이 가장 적합합니다'라는 문장으로 시작할 것") - writing_guide: str = Field(description="실제 자소서 작성 가이드라인 및 핵심 키워드 정리") - + + # ── 구조화 출력 모델 ────────────────────────────────────────── + class StrategicPlacement(BaseModel): + experience_id: Optional[Union[str, int]] = Field( + None, + description="매핑된 경험 ID. 경험 목록 중 하나의 id 값 (문자열 혹은 숫자 가능). 적합한 경험이 없으면 null." + ) + selected_strategy: Literal["SO", "ST", "WO", "WT", "N/A"] = Field( + ..., + description="SWOT 전략. 적합한 경험이 없으면 N/A." + ) + jd_targeting: str = Field( + description="[JD 타겟팅] JD의 어떤 구체적 요구사항을 Opportunity(O) 또는 Threat(T)으로 설정했는지 명시." + ) + dynamic_framing: str = Field( + description="[동적 프레이밍] user_persona 기준으로 해당 경험이 왜 강점(S) 또는 약점(W)으로 해석되는지 명시. " + "페르소나와 팩트가 충돌하면 팩트를 우선하고 그 이유를 설명하세요." + ) + strategy_derivation: str = Field( + description="[전략 도출] 선택한 SWOT 전략이 이 문항과 JD 환경에서 왜 최선의 선택인지 전략적 가치를 논증." + ) + writing_guide: str = Field( + description="실제 자소서 작성 가이드라인: 경험 내용과 전략을 연결하여 강조할 키워드·서술 흐름 정리." + ) + + # ── 시스템 프롬프트 ─────────────────────────────────────────── + SYSTEM_PROMPT = ( + "당신은 개발자 채용을 위한 최고 수준의 이력서/자소서 전략 분석 에이전트입니다.\n" + "JD 분석, 지원자 경험, 지원자 성향을 종합하여 각 자소서 문항에 가장 강력한 전략적 매핑을 제공하세요.\n\n" + "# 핵심 지침\n" + "1. [유연한 매핑 - 1:N 허용]\n" + " - 하나의 경험 ID를 여러 문항에 중복 사용할 수 있습니다.\n" + " - 경험 데이터(content, tags)에 없는 사실(팀워크, 갈등 조율 등)을 절대 지어내지 마세요.\n" + " - 어떤 경험도 이 문항에 적합하지 않다면 experience_id=null, selected_strategy=N/A를 반환하세요.\n\n" + "2. [동적 S/W 프레이밍 - User Persona 기반]\n" + " - 경험을 S/W로 분류할 때 user_persona를 해석 필터로 사용하세요.\n" + " - 단, Persona와 경험 팩트(content)가 충돌하면 반드시 팩트를 우선하세요.\n" + " - 예: Persona='끈기'지만 content에 '중도 포기'가 명시 → W 분류, WT 전략 선택.\n\n" + "3. [SWOT 전략 정의]\n" + " - SO: 강점(S) + Persona → JD 핵심 요구사항(O) 완벽 충족\n" + " - ST: 강점(S) + Persona → JD 실무 허들/위협(T) 돌파\n" + " - WO: 약점(W) 인정 + JD 환경(O)에서의 성장 어필\n" + " - WT: 약점(W)과 위협(T) 직시, 현실적 보완책 제시\n\n" + "4. [reasoning 3단 논리 구조]\n" + " - jd_targeting: JD의 어떤 구체적 요구사항을 O/T로 설정했는가\n" + " - dynamic_framing: Persona 기준으로 이 경험이 왜 S/W인가 (팩트 충돌 시 팩트 우선)\n" + " - strategy_derivation: 선택한 전략이 왜 이 문항과 JD 환경에서 최선인가\n\n" + "# 한국어 출력 규칙 (가장 중요)\n" + "모든 텍스트 출력을 반드시 한국어로 작성하세요. " + "영어 사용을 엄격히 금지합니다." + ) + + # 경험 목록을 요약 (점수 포함해서 LLM이 참고하도록) + experiences_summary = json.dumps( + [ + { + "id": e["id"], + "title": e["title"], + "content": e["content"], + "tags": e.get("tags", []), + "priority": e.get("priority", ""), + "swot_scores": e.get("scores", {}), + "primary_strategy": e.get("primary_strategy", ""), + } + for e in experiences + ], + ensure_ascii=False, + indent=2, + ) + + placement_prompt = ChatPromptTemplate.from_messages([ + ("system", SYSTEM_PROMPT), + ("user", + "## 자소서 문항\n{prompt_text}\n\n" + "## JD 분석 결과\nOpportunities: {opportunities}\nThreats: {threats}\n\n" + "## User Persona\n{user_persona}\n\n" + "## 사용 가능한 경험 목록 (1:N 재사용 가능)\n{experiences}") + ]) + chain = ( + placement_prompt | llm.with_structured_output(StrategicPlacement) + ).with_retry(stop_after_attempt=3) + for prompt_text in prompts: - if not remaining_indices: + print(f"[*] 문항 분석 중: '{prompt_text[:40]}...'") + try: + result: StrategicPlacement = chain.invoke({ + "prompt_text": prompt_text, + "opportunities": jd_context.get("opportunities", ""), + "threats": jd_context.get("threats", ""), + "user_persona": user_persona, + "experiences": experiences_summary, + }) + + # experience_id → experience_title 조회 (타입 불일치 방지를 위해 str()로 변환 후 비교) + exp_title = "N/A" + if result.experience_id is not None: + matched = next( + (e for e in experiences if str(e["id"]) == str(result.experience_id)), None + ) + exp_title = matched["title"] if matched else f"Unknown (id={result.experience_id})" + print(f" → 전략: {result.selected_strategy} | 경험: '{exp_title}'") + else: + print(f" → 전략: N/A | 적합한 경험 없음") + placements.append({ - "question": prompt_text, - "experience_title": "경험 부족", - "selected_strategy": "N/A", - "reasoning": "할당 가능한 여유 경험이 부족합니다.", - "writing_guide": "N/A" + "essay_question": prompt_text, + "matched_experience_id": result.experience_id, + "matched_experience_title": exp_title, + "strategy": result.selected_strategy, + "jd_targeting": result.jd_targeting, + "dynamic_framing": result.dynamic_framing, + "strategy_derivation": result.strategy_derivation, + "writing_guide": result.writing_guide, }) - continue - - # [개선] 6가지 문항 유형 기반 의도 감지 (SO 기본값 편향 제거) - target_strategy, fallback_used = _detect_intent_strategy(prompt_text, llm) - if fallback_used or target_strategy is None: - # LLM 감지 실패 시 경험 점수 합계 기반 폴백 (SO 하드코딩 제거) - target_strategy = _score_based_fallback(experiences, remaining_indices, priority_weight) - print(f"[*] 의도 감지 폴백 사용 → 점수 기반 전략: {target_strategy} (문항: '{prompt_text[:30]}...')") - else: - print(f"[*] 의도 감지 성공 → 전략: {target_strategy} (문항: '{prompt_text[:30]}...')") - - best_exp_idx = -1 - max_score = -1 - best_priority_val = -1 - - for idx in remaining_indices: - exp = experiences[idx] - score = exp.get("scores", {}).get(target_strategy, 0) - p_val = priority_weight.get(exp.get("priority", "하"), 0) - - if max_score != -1 and abs(max_score - score) < 5: - if p_val > best_priority_val: - max_score = score - best_exp_idx = idx - best_priority_val = p_val - elif score > max_score: - max_score = score - best_exp_idx = idx - best_priority_val = p_val - - if best_exp_idx != -1: - best_exp = experiences[best_exp_idx] - remaining_indices.remove(best_exp_idx) - - # [Fix] Self-Correction을 유도하는 프롬프트 적용 - guide_prompt = ChatPromptTemplate.from_messages([ - ("system", "자소서 작성 가이드라인과 매칭 논리를 작성해주세요.\n" - "1. reasoning 필드는 서두에 반드시 '분석 결과 [{target_strategy}] 전략이 가장 적합합니다'라는 문장을 강제로 포함하여 AI 스스로 자신의 논리적 일관성을 확인(Self-Correction)하세요. 결론 내린 전략 명칭은 반드시 {target_strategy} 와 100% 일치해야 합니다.\n" - "2. writing_guide 필드는 경험의 내용과 매칭 논리를 연결하여, 서술 시 강조해야 할 핵심 키워드 및 흐름을 분석하세요."), - ("user", "문항: {prompt_text}\n전략: {target_strategy}\n경험 명: {exp_title}\n경험 내용: {exp_content}\n경험의 원래 평가이유: {reasoning}") - ]) - try: - guide_chain = (guide_prompt | llm.with_structured_output(GuideOutput)).with_retry(stop_after_attempt=3) - guide_result = guide_chain.invoke({ - "prompt_text": prompt_text, - "target_strategy": target_strategy, - "exp_title": best_exp["title"], - "exp_content": best_exp["content"], - "reasoning": best_exp.get("strategy_reasoning", "") - }) - final_reasoning = guide_result.reasoning - final_guide = guide_result.writing_guide - except Exception: - final_reasoning = f"분석 결과 [{target_strategy}] 전략이 가장 적합합니다. (세부 매칭 논리 생성 실패)" - final_guide = "가이드 생성 실패" + except Exception as e: + state["errors"].append(f"[Strategic Placement Error] {str(e)}") placements.append({ - "question": prompt_text, - "experience_id": best_exp["id"], - "experience_title": best_exp["title"], - "selected_strategy": target_strategy, - "reasoning": final_reasoning, - "writing_guide": final_guide + "essay_question": prompt_text, + "matched_experience_id": None, + "matched_experience_title": "오류", + "strategy": "N/A", + "jd_targeting": f"배치 처리 중 오류 발생: {str(e)}", + "dynamic_framing": "", + "strategy_derivation": "", + "writing_guide": "N/A", }) - + state["placements"] = placements - state["remaining_indices"] = remaining_indices return state