Merge branch 'dev' into feat/text-news-mcp-server

Author: gsgh3016

.env.example

@@ -13,6 +13,7 @@ LANGGRAPH_PORT=2024
 # You can get it from the OpenAI website (https://platform.openai.com/).
 OPENAI_API_KEY=sk...
 
+
 # Google API Key (예시에는 없지만, 필요하다면 주석 추가)
 GOOGLE_API_KEY=...
 
@@ -22,3 +23,7 @@ UPSTAGE_API_KEY=...
 # News API Key - required to use the news API.
 # You can get it from the News API website (https://newsapi.org/).
 NEWS_API_KEY=...
+
+# Groq API Key - used to access Groq LLMs such as Mixtral or LLaMA models.
+# Sign up and get your key from https://console.groq.com/keys
+GROQ_API_KEY=grq...

agents/text/modules/chains.py

@@ -5,21 +5,24 @@
 
 """
 
+
 from langchain_core.output_parsers import StrOutputParser
 from langchain_core.runnables import (
     RunnableLambda,
+    RunnableMap,
     RunnablePassthrough,
     RunnableSerializable,
 )
 
 from agents.text.mcp.mcp_client import scrape_news
-from agents.text.modules.models import get_openai_model
+from agents.text.modules.models import get_groq_model, get_openai_model
 from agents.text.modules.persona import PERSONA
 from agents.text.modules.prompts import (
     get_extraction_prompt,
     get_instagram_text_prompt,
     get_news_scraping_query_prompt,
     get_topic_from_news_prompt,
+    get_persona_match_prompt,
 )
 
 
@@ -119,3 +122,89 @@ def set_instagram_text_chain() -> RunnableSerializable:
         | model  # LLM 모델 호출
         | StrOutputParser()  # 결과를 문자열로 변환
     )
+
+
+def set_instagram_text_format_check_chain() -> RunnableLambda:
+    """
+    인스타그램 포맷(2200자 이하) 검사를 위한 체인을 반환합니다.
+
+    Returns:
+        RunnableLambda: 텍스트 길이를 검사하는 실행 체인
+    """
+    return RunnableLambda(lambda x: len(x["text"]) <= 2200)
+
+
+def set_sensitive_text_check_chain() -> RunnableLambda:
+    def is_text_safe(x):
+        model = get_groq_model("meta-llama/llama-guard-4-12b")
+        try:
+            response = model.invoke(x["text"])
+            return "safe" in response.content.lower()
+        except Exception as e:
+            print(f"[ERROR] llama-guard request failed: {e}")
+        return False
+
+    return RunnableLambda(is_text_safe)
+
+
+def set_text_persona_match_check_chain() -> RunnableLambda:
+    def check_persona_match(x):
+        model = get_openai_model()
+
+        text = x["text"]
+        persona = x.get("persona", {})
+
+        # 다양한 타입의 persona_description 처리: dict, str, list
+        if isinstance(persona, dict):
+            persona_description = "\n".join([f"{k}: {v}" for k, v in persona.items()])
+        elif isinstance(persona, list):
+            persona_description = "\n".join([str(p) for p in persona])
+        else:
+            persona_description = str(persona)
+
+        prompt_template = get_persona_match_prompt()
+        prompt = prompt_template.format(
+            persona_description=persona_description, text=text
+        )
+
+        try:
+            response = model.invoke(prompt).content.strip().upper()
+            return "YES" in response
+        except Exception as e:
+            print(f"[ERROR] Persona check failed: {e}")
+            return False
+
+    return RunnableLambda(check_persona_match)
+
+
+def set_text_content_check_chain() -> RunnableSerializable:
+    return (
+        RunnablePassthrough.assign(
+            text=lambda x: x if isinstance(x, str) else x.get("instagram_text", ""),
+            persona=lambda x: (
+                x.get("persona_extracted", {}) if isinstance(x, dict) else {}
+            ),
+        )
+        | RunnableMap(
+            {
+                "format_check_passed": set_instagram_text_format_check_chain(),
+                "safety_check_passed": set_sensitive_text_check_chain(),
+                "persona_check_passed": set_text_persona_match_check_chain(),
+            }
+        )
+        | RunnableLambda(
+            lambda results: {
+                "text_content_checker_result": {
+                    "success": all(results.values()),
+                    "reason": [k for k, v in results.items() if not v],
+                    "content_check_passed": all(results.values()),
+                    **results,
+                    "message": (
+                        "Text content is valid."
+                        if all(results.values())
+                        else "Text content failed validation checks."
+                    ),
+                }
+            }
+        )
+    )

agents/text/modules/models.py

@@ -3,6 +3,7 @@
 기본적으로 사용할 모델 인스턴스를 설정하고 생성하고 반환시킵니다.
 """
 
+from langchain_groq import ChatGroq
 from langchain_openai import ChatOpenAI
 
 
@@ -17,3 +18,15 @@ def get_openai_model(temperature=0.7, top_p=0.9):
     """
     # OpenAI 모델 초기화 및 반환
     return ChatOpenAI(model="gpt-4o-mini", temperature=temperature, top_p=top_p)
+
+
+def get_groq_model(model_name="llama3-8b-8192", temperature=0.7, top_p=0.9):
+    """
+    Groq API를 사용하는 Llama3 기반 모델을 LangChain에서 가져옵니다.
+    사용 가능한 모델 예: "llama3-8b-8192", "llama3-70b-8192", "mixtral-8x7b-32768"
+    """
+    return ChatGroq(
+        model_name=model_name,
+        temperature=temperature,
+        top_p=top_p,
+    )

agents/text/modules/nodes.py

@@ -11,6 +11,7 @@
     set_extraction_chain,
     set_instagram_text_chain,
     set_topic_generation_news_chain,
+    set_text_content_check_chain,
 )
 from agents.text.modules.persona import PERSONA
 from agents.text.modules.state import TextState
@@ -89,3 +90,36 @@ def execute(self, state: TextState) -> dict:
             return {"response": result}
         except Exception as e:
             return {"response": f"뉴스 검색 중 오류가 발생했습니다: {str(e)}"}
+
+          
+class TextContentCheckNode(BaseNode):
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self.chain = set_text_content_check_chain()
+
+    def execute(self, state: TextState) -> dict:
+        instagram_text = state.get("instagram_text", "")
+
+        # instagram_text가 빈 칸(또는 공백)일 때는 모든 체크를 스킵하고 성공 결과 리턴
+        if not instagram_text or not instagram_text.strip():
+            result = {
+                "text_content_checker_result": {
+                    "success": True,
+                    "reason": [],
+                    "content_check_passed": True,
+                    "format_check_passed": True,
+                    "safety_check_passed": True,
+                    "persona_check_passed": True,
+                    "message": "Skipped checks because text_content is empty.",
+                }
+            }
+            state.update(result)
+            return result
+
+        input_data = {
+            "response": state.get("response", [""]),
+            "persona_extracted": state.get("persona_extracted", {}),
+        }
+        result = self.chain.invoke(input_data)
+        state.update(result)
+        return result

agents/text/modules/prompts.py

@@ -139,4 +139,22 @@ def get_topic_from_news_prompt():
     return PromptTemplate(
         template=prompt_template,
         input_variables=["news_article", "persona_details"],
+
+      
+      
+def get_persona_match_prompt() -> PromptTemplate:
+    """
+    Returns a prompt template to evaluate if a given text aligns with a provided persona.
+
+    The model must respond only with "YES" or "NO".
+    """
+    template = (
+        "The following is an Instagram text content. Please determine whether it aligns with the provided persona. "
+        "Reply only with 'YES' if it matches well, or 'NO' if it doesn't.\n\n"
+        "[Persona]\n{persona_description}\n\n"
+        "[Text]\n{text}"
+    )
+
+    return PromptTemplate(
+        template=template, input_variables=["persona_description", "text"]
     )

agents/text/modules/state.py

@@ -24,3 +24,4 @@ class TextState(TypedDict):
     response: Optional[Annotated[list, add_messages]] = (
         None  # 응답 메시지 목록 (add_messages로 주석되어 메시지 추가 기능 제공)
     )
+    text_content_checker_result: (dict)  # 텍스트 컨텐츠 검사 결과 전체를 담는 구조화된 필드

agents/text/pyproject.toml

@@ -11,6 +11,7 @@ readme = "README.md"
 requires-python = ">=3.13"
 dependencies = [
     "langchain-mcp-adapters>=0.0.9",
+    "langchain-groq>=0.3.2",
     "langchain-openai>=0.3.12",
     "mcp>=1.6.0",
     "newsapi-python>=0.2.7",

agents/text/workflow.py

@@ -5,6 +5,7 @@
     GenTextNode,
     PersonaExtractionNode,
     TopicFromNewsNode,
+    TextContentCheckNode,
 )
 from agents.text.modules.state import TextState
 
@@ -40,13 +41,18 @@ def build(self):
         # 텍스트 생성 노드 추가
         builder.add_node("text_generation", GenTextNode())
 
+        # 텍스트 컨텐츠 체커 노드 추가
+        builder.add_node("text_content_check", TextContentCheckNode())
+
         # 시작 노드에서 페르소나 추출 노드로 연결
         builder.add_edge("__start__", "topic_from_news")
         builder.add_edge("topic_from_news", "persona_extraction")
         # 페르소나 추출 노드에서 텍스트 생성 노드로 연결
         builder.add_edge("persona_extraction", "text_generation")
-        # 텍스트 생성 노드에서 종료 노드로 연결
-        builder.add_edge("text_generation", "__end__")
+        # 텍스트 생성 노드에서 텍스트 컨텐츠 체커 노드로 연결
+        builder.add_edge("text_generation", "text_content_check")
+        # 텍스트 컨텐츠 체커 노드에서 종료 노드로 연결
+        builder.add_edge("text_content_check", "__end__")
 
         # 조건부 에지 추가 예시
         # builder.add_conditional_edges(