app.py

import gradio as gr
from duckduckgo_search import DDGS
from swarm import Swarm, Agent
from datetime import datetime
import re

current_date = datetime.now().strftime("%Y-%m")
client = Swarm()

def contains_chinese(text):
    """중국어 문자 포함 여부 확인"""
    return any('\u4e00' <= char <= '\u9fff' for char in text)

def get_news_articles(topic, date=None):
    print(f"Running DuckDuckGo news search for {topic}...")
    ddg_api = DDGS()
    search_query = f"{topic} {date if date else current_date}"
    results = ddg_api.news(search_query, max_results=10)
    if results:
        # "https://www.msn.com" URL을 포함하지 않은 뉴스만 필터링
        filtered_results = [
            result for result in results
            if "https://www.msn.com" not in result['url']
        ]
        
        if filtered_results:
            news_results = "\n\n".join([
                f"**제목**: {result['title']}\n**링크**: {result['url']}\n**내용**: {result['body']}" 
                for result in filtered_results
            ])
            return news_results
        else:
            return f"{topic}에 대한 뉴스를 찾을 수 없습니다."
    else:
        return f"{topic}에 대한 뉴스를 찾을 수 없습니다."

news_agent = Agent(
    name="News Assistant",
    instructions="""
    당신은 DuckDuckGo 검색을 사용하여 주어진 주제에 대한 오늘의 최신 뉴스 기사를 제공합니다.
    반드시 한국어로만 응답하십시오.
    중국어가 포함된 내용이 있다면 한국어로 번역하여 제공하십시오.
    항상 응답에 전체 URL을 포함하십시오.
    각 뉴스 기사에 대한 날짜를 포함하십시오.
    """,
    functions=[get_news_articles],
    model="qwen2.5:latest"
)

editor_agent = Agent(
    name="Editor Assistant",
    instructions="""
    당신은 뉴스 편집자입니다. 다음 규칙에 따라 뉴스를 한국어로 편집해주세요:
    1. 각 뉴스는 별도의 섹션으로 구분하여 작성
    2. 모든 원본 URL을 반드시 포함
    3. 각 기사의 날짜 포함
    4. 중요 내용은 ** 볼드체 처리
    5. URL은 [제목](URL) 형식으로 포함
    6. 반드시 한국어로 작성하고, 중국어가 포함된 내용은 한국어로 번역하여 제공
    7. 각 기사는 200단어 이내로 요약
    """,
    model="qwen2.5:latest"
)

def process_news_request(topic, progress=gr.Progress(track_tqdm=True)):
    try:
        progress(0, desc="시작하는 중...")
        
        progress(0.3, desc="뉴스 검색 중...")
        news_response = client.run(
            agent=news_agent,
            messages=[{"role": "user", "content": f"한국어로 {topic}에 대한 최신 뉴스를 찾아주세요."}],
        )
        
        raw_news = news_response.messages[-1]["content"]
        
        # 원본 뉴스에서 중국어가 포함된 항목 제거
        raw_news_list = raw_news.split("\n\n")
        filtered_news_list = [news for news in raw_news_list if not contains_chinese(news)]
        
        # MSN URL이 포함된 항목도 제거
        filtered_news_list = [news for news in filtered_news_list if "https://www.msn.com" not in news]
        
        filtered_news = "\n\n".join(filtered_news_list)

        progress(0.6, desc="뉴스 편집 중...")
        
        edited_news_response = client.run(
            agent=editor_agent,
            messages=[{"role": "user", "content": f"다음 뉴스를 한국어로 편집해주세요:\n{filtered_news}"}],
        )
        
        progress(1.0, desc="완료!")
        return edited_news_response.messages[-1]["content"]
    except Exception as e:
        return f"오류가 발생했습니다: {str(e)}\n\n디버그 정보: {type(e).__name__}"

with gr.Blocks(theme=gr.themes.Soft()) as demo:
    gr.Markdown("# 🤖 Swarm AI 뉴스 에이전트")
    
    with gr.Row():
        topic_input = gr.Textbox(
            label="검색할 주제를 입력하세요",
            placeholder="예: AI, SpaceX, Blockchain...",
            lines=1
        )
    
    with gr.Row():
        submit_btn = gr.Button("검색", variant="primary")
    
    with gr.Row():
        result_output = gr.Markdown(
            value="Swarm AI를 활용해서 Ollama qwen2.5 모델이 사용자가 요청한 최신 뉴스를 검색하고 한국어로 요약해드립니다.",
            label="AI가 정리한 뉴스"
        )
    
    examples = gr.Examples(
        examples=[
            ["인공지능"],
            ["테슬라"],
            ["OpenAI"],
            ["기후변화"]
        ],
        inputs=topic_input
    )
    
    submit_btn.click(
        fn=process_news_request,
        inputs=topic_input,
        outputs=result_output
    )
    
    topic_input.submit(
        fn=process_news_request,
        inputs=topic_input,
        outputs=result_output
    )

if __name__ == "__main__":
    demo.queue().launch(share=True)