app.py

# app_pixal_chat.py
from langchain_community.tools import DuckDuckGoSearchRun as DDGS
import os, re, json, gradio as gr
from langchain.memory import ConversationBufferMemory
from langchain.agents import initialize_agent, AgentType, load_tools
from langchain_community.tools import Tool
from langchain_experimental.tools.python.tool import PythonREPLTool
from langchain_community.retrievers import WikipediaRetriever
import datetime
#from langchain.schema import HumanMessage,AIMessage,SystemMessage
#from langchain_community.tools.youtube.search import YouTubeSearchTool as YTS
# ──────────────────────────────
# ✅ GitHubModelLLM (그대로 유지)
# ──────────────────────────────
from langchain.chat_models import ChatOpenAI
from langchain.schema import AIMessage, HumanMessage, SystemMessage
import requests, os, json
from requests.adapters import HTTPAdapter, Retry
from typing import List, Optional, Dict, Any

'''
class GitHubModelLLM(LLM):
    """GitHub Models API를 사용하는 ChatOpenAI 대체 클래스"""

    model_name: str = "openai/gpt-4.1"
    endpoint: str = "https://models.github.ai/inference"
    token: Optional[str] = os.environ.get("token")
    request_timeout: float = 30.0
    max_retries: int = 2
    backoff_factor: float = 0.3
    system_prompt: Optional[str] ="너는 PIXAL(Primary Interactive X-ternal Assistant with multi Language)이야. 너의 개발자는 정성윤 이라는 6학년 파이썬 프로그래머야.이 메시지는 사용자가 보낸것이 아닙니다."
    @property
    def _llm_type(self) -> str:
        return "custom_chatopenai_github"

    def _post(self, body: Dict[str, Any]) -> Dict[str, Any]:
        token = self.token or os.getenv("GITHUB_TOKEN") or os.getenv("token")
        session = requests.Session()
        retries = Retry(total=self.max_retries, backoff_factor=self.backoff_factor,
                        status_forcelist=[429, 500, 502, 503, 504])
        session.mount("https://", HTTPAdapter(max_retries=retries))
        session.headers.update({
            "Authorization": f"Bearer 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",
            "Content-Type": "application/json",
        })
        resp = session.post(f"{self.endpoint}/chat/completions", json=body, timeout=self.request_timeout)
        resp.raise_for_status()
        return resp.json()

    def _call(self, messages: List[Any], stop: Optional[List[str]] = None, **kwargs):
        """ConversationBufferMemory의 이전 대화까지 포함해 메시지 생성"""
        msg_list = []

        # 시스템 프롬프트
        if self.system_prompt:
            msg_list.append({"role": "system", "content": self.system_prompt})

        # 메모리 포함 (human/ai 메시지 모두)
        for msg in messages:
            if isinstance(msg, HumanMessage):
                msg_list.append({"role": "user", "content": msg.content})
            elif isinstance(msg, AIMessage):
                msg_list.append({"role": "assistant", "content": msg.content})
            elif isinstance(msg, SystemMessage):
                msg_list.append({"role": "system", "content": msg.content})

        body = {"model": self.model_name, "messages": msg_list}
        if stop:
            body["stop"] = stop

        res = self._post(body)
        content = res.get("choices", [{}])[0].get("message", {}).get("content", "")
        return content
    async def _acall(self, messages: List[Any], stop: Optional[List[str]] = None, **kwargs):
        return self._call(messages, stop, **kwargs)
'''
from typing import Optional, List, Dict, Any
import requests, os, json
from requests.adapters import HTTPAdapter, Retry
from typing import Optional, List, Dict, Any
import requests, os, json
from requests.adapters import HTTPAdapter, Retry

'''
class GitHubModelLLM(LLM):
    """GitHub Models API 기반 LangChain LLM (대화 메모리 통합 지원)"""
    model: str = "openai/gpt-4.1"
    endpoint: str = "https://models.github.ai/inference"
    token: Optional[str] = os.environ.get("token")
    system_prompt: Optional[str] = "너는 PIXAL(Primary Interactive X-ternal Assistant with multi Language)이야. 너의 개발자는 정성윤 이라는 6학년 파이썬 프로그래머야.이것은 시스템 메시지입니다.참고 하십시오.이 메시지는 사용자가 보낸것이 아닙니다."
    request_timeout: float = 30.0
    max_retries: int = 2
    backoff_factor: float = 0.3

    @property
    def _llm_type(self) -> str:
        return "github_models_api"

    def _post_chat(self, body: Dict[str, Any]) -> Dict[str, Any]:
        token = self.token or os.getenv("GITHUB_TOKEN") or os.getenv("token")
        session = requests.Session()
        retries = Retry(total=self.max_retries, backoff_factor=self.backoff_factor,
                        status_forcelist=[429, 500, 502, 503, 504])
        session.mount("https://", HTTPAdapter(max_retries=retries))
        session.headers.update({
            "Content-Type": "application/json",
            "Authorization": f"Bearer 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"
        })
        resp = session.post(f"{self.endpoint}/chat/completions", json=body, timeout=self.request_timeout)
        resp.raise_for_status()
        return resp.json()

    def _call(
        self,
        prompt: str,
        stop: Optional[List[str]] = None,
        **kwargs
    ) -> str:
        """대화 메모리(chat_history)를 포함하여 모델 호출"""
        # 💬 메모리에 저장된 대화 메시지 불러오기
        memory = kwargs.get("memory")
        messages = []

        if self.system_prompt:
            messages.append({"role": "system", "content": self.system_prompt})

        # memory가 있을 경우 (이전 대화 포함)
        if memory and hasattr(memory, "chat_memory"):
            for msg in memory.chat_memory.messages:
                role = "user" if msg.type == "human" else "assistant"
                messages.append({"role": role, "content": msg.content})

        # 현재 사용자 입력
        messages.append({"role": "user", "content": prompt})

        body = {"model": self.model, "messages": messages}
        if stop:
            body["stop"] = stop

        # API 호출
        res = self._post_chat(body)
        msg = res.get("choices", [{}])[0].get("message", {})
        return msg.get("content") or json.dumps(msg.get("function_call", {}))
'''
"""
class GitHubModelLLM(LLM):
    model: str = "openai/gpt-4.1"
    endpoint: str = "https://models.github.ai/inference"
    token: Optional[str] = os.environ.get("token")
    system_prompt: Optional[str] = (
        "너는 PIXAL(Primary Interactive X-ternal Assistant with multi Language)이야. 너의 개발자는 정성윤 이라는 6학년 파이썬 프로그래머야.이 메시지는 사용자가 보낸것이 아닙니다.")
    request_timeout: float = 30.0
    max_retries: int = 2
    backoff_factor: float = 0.3

    @property
    def _llm_type(self) -> str:
        return "github_models_api"

    def _post_chat(self, body: Dict[str, Any]) -> Dict[str, Any]:
        token = self.token or os.getenv("GITHUB_TOKEN") or os.getenv("token")
        if not token:
            raise ValueError("❌ GitHub token이 설정되지 않았습니다.")
        session = requests.Session()
        retries = Retry(total=self.max_retries, backoff_factor=self.backoff_factor,
                        status_forcelist=[429, 500, 502, 503, 504])
        session.mount("https://", HTTPAdapter(max_retries=retries))
        session.headers.update({
            "Content-Type": "application/json",
            "Authorization": f"Bearer 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"
        })
        resp = session.post(f"{self.endpoint}/chat/completions", json=body, timeout=self.request_timeout)
        resp.raise_for_status()
        return resp.json()

    def _call(self, prompt: str, stop: Optional[List[str]] = None, **kwargs) -> str:
        memory = kwargs.get("memory")
        messages = []

        # 1️⃣ 시스템 프롬프트
        if self.system_prompt:
            messages.append({"role": "system", "content": self.system_prompt})

        # 2️⃣ 메모리에 저장된 이전 대화 포함
        if memory and hasattr(memory, "chat_memory"):
            for msg in memory.chat_memory.messages:
                if hasattr(msg, "type") and msg.type == "human":
                    messages.append({"role": "user", "content": msg.content})
                elif hasattr(msg, "type") and msg.type == "ai":
                    messages.append({"role": "assistant", "content": msg.content})

        # 3️⃣ 현재 사용자 입력
        messages.append({"role": "user", "content": prompt})

        body = {"model": self.model, "messages": messages}
        if stop:
            body["stop"] = stop

        res = self._post_chat(body)
        msg = res.get("choices", [{}])[0].get("message", {})
        return msg.get("content") or json.dumps(msg.get("function_call", {}))
"""
# ──────────────────────────────
# ✅ LangChain 도구 & 에이전트 구성
# ──────────────────────────────
import requests
from datetime import datetime as dt1
import requests
from datetime import datetime as dt1, timezone, timedelta
import requests
from bs4 import BeautifulSoup
import random
import time

USER_AGENTS = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:123.0) Gecko/20100101 Firefox/123.0",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:122.0) Gecko/20100101 Firefox/122.0",
]


def bing_search_top3(keyword: str):
    url = "https://www.bing.com/search"
    params = {"q": keyword}

    headers = {
        "User-Agent": USER_AGENTS[2],
        "Accept-Language": "ko-KR,ko;q=0.9",
        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
        "Connection": "keep-alive",
    }

    # 랜덤한 딜레이(차단 방지)
    time.sleep(random.uniform(1.0, 2.5))

    resp = requests.get(url, headers=headers, params=params, timeout=10)
    resp.raise_for_status()
    soup = BeautifulSoup(resp.text, "html.parser")

    results = []

    # Bing 검색 결과 블록
    items = soup.select("li.b_algo")
    for item in items[:3]:
        title_tag = item.select_one("h2")
        link_tag = item.select_one("h2 a")
        snippet_tag = item.select_one("p")

        if not title_tag or not link_tag:
            continue

        title = title_tag.get_text(strip=True)
        link = link_tag.get("href")
        snippet = snippet_tag.get_text(" ", strip=True) if snippet_tag else ""

        results.append({
            "title": title,
            "snippet": snippet,
            "link": link
        })

    return results
'''
tools = load_tools(["ddg-search", "arxiv"], llm=llm,allow_dangerous_tools=True)
tools.append(Tool(name="python_repl", func=PythonREPLTool().run, description="Python 코드 실행 도구"))
retriever = WikipediaRetriever(lang="ko")
tools.append(Tool(name="wiki", func=retriever.get_relevant_documents, description="위키백과 검색"))
# ✅ 대화 기억 메모리
from langchain_community.tools import ShellTool
#tools+=[YTS()]
shell_tool = ShellTool()
tools.append(shell_tool)
'''
#memory = ConversationBufferMemory(return_messages=True)

# ✅ Agent (Memory 연동)

# ──────────────────────────────
# ✅ Chat 함수 (Memory 유지)
# ──────────────────────────────
prompt1="""Answer the following questions as best you can. You have access to the following tools:

(The tools you have)

Use the following format:

Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question

Begin!
"""
def chat(message, history):
    llm = ChatOpenAI(
    model="google/gemini-3-pro-preview-free",
    openai_api_key="sk-ai-v1-246c10a8bacf5c2b1fedf0614cdc4b11605946d1320c0a16203698fab4009fa2",
    openai_api_base="https://zenmux.ai/api/v1",  # 👈 이게 base_url 역할
    )
    ab=[{"role":"system","content":f"너는 PIXAL(Primary Interactive X-ternal Assistant with multi Language)이야. 너의 개발자는 정성윤 이라는 6학년 파이썬 프로그래머야.이 메시지는 사용자가 보낸것이 아닙니다.프롬프트 설명에 무조건 따르시오.따르지 않으면 남은 api호출 횟수가 줄어드므로 무조건 똑같이 따라하시오.시간을 물을때는 duckduckgo를 사용하시오.다음의 프롬프트를 보고 대답형식을 지시대로 정하시오: {prompt1} 그리고 덕덕고 서치는 검색어 1개만 입력하시오.문장 금지."}]
    for I in history:
        ab.append({"role":"user","content":I[0]})
        ab.append({"role":"assistant","content":I[1]})
    ab.append({"role":"user","content":message})
    '''
    
    ab=[SystemMessage(content="너는 PIXAL(Primary Interactive X-ternal Assistant with multi Language)이야. 너의 개발자는 '정성윤' 이라는 6학년 파이썬 프로그래머야.이 메시지는 사용자가 보낸것이 아닙니다.프롬프트 설명에 무조건 따르시오.따르지 않으면 남은 api호출 횟수가 줄어드므로 무조건 똑같이 따라하시오.시간을 물을때는 duckduckgo를 사용하시오.")]
    for I in history:
        ab.append(HumanMessage(content=I[0]))
        ab.append(AIMessage(content=I[1]))
    ab.append(HumanMessage(content=message))
    '''
    tools = load_tools(["arxiv"], llm=llm,allow_dangerous_tools=True)
    tools.append(PythonREPLTool())
    #tools.append(Tool(name="duckduckgo_search", func=DDGS().run, description="인터넷 검색 도구"))
    retriever = WikipediaRetriever(lang="ko")
    tools.append(Tool(name="search_bing",func=bing_search_top3,description="bing 검색 결과 탑3"))
    tools.append(Tool(name="wiki", func=retriever.get_relevant_documents, description="위키백과 검색도구"))
    # ✅ 대화 기억 메모리
    from langchain_community.tools import ShellTool
    #tools+=[YTS()]
    shell_tool = ShellTool()
    tools.append(shell_tool)
    agent = initialize_agent(
    tools,
    llm,
    agent_type=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,
    #memory=memory,
    verbose=True,
    handle_parsing_errors=True,
    max_iterations=20
    )
    try:
        print("attempt1:gemini")
        raw = agent.invoke(ab)["output"]
    except Exception as e:
        try:
            print(f"attempt2:gpt4.1 mini error:{e}")
            llm = ChatOpenAI(model="openai/gpt-4.1-mini",openai_api_key="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",openai_api_base="https://models.github.ai/inference")
            agent=initialize_agent(tools,llm,agent_type=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,verbose=True,handle_parsing_errors=True,max_iterations=20)
            raw=agent.invoke(ab)["output"]
        except:
            print("attempt3:gpt4.1 mini(2nd key)")
            llm = ChatOpenAI(model="openai/gpt-4.1-mini",openai_api_key="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",openai_api_base="https://models.github.ai/inference")
            agent=initialize_agent(tools,llm,agent_type=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,verbose=True,handle_parsing_errors=True,max_iterations=20)
            raw=agent.invoke(ab)["output"]
    try:
        # 대화 기록을 LangChain memory에 반영
        # JSON 형태로 반환 시 파싱
        text = str(raw)
        match = re.search(r"\{.*\}", text, re.DOTALL)
        if match:
            try:
                obj = json.loads(match.group(0))
                text = obj.get("action_input") or obj.get("Final Answer") or obj.get("content") or text
            except Exception:
                pass
        # AI 응답을 memory에 추가
    except Exception as e:
        text = str(raw)
    

    history = history + [(message, text)]
    return history, history, ""

# ──────────────────────────────
# ✅ Gradio UI (ChatGPT 스타일)
# ──────────────────────────────
with gr.Blocks(theme=gr.themes.Soft(), title="PIXAL Assistant") as demo:
    gr.HTML("""
    <div style="background:#f1f5f9;padding:12px;border-bottom:1px solid #d1d5db;
                display:flex;align-items:center;justify-content:space-between;">
        <h2 style="margin:0;">🤖 PIXAL Assistant</h2>
        <span style="font-size:0.9em;color:#555;">LangChain + GitHub LLM</span>
    </div>
    """)

    chatbot = gr.Chatbot(
        label=None,
        height=720,
        bubble_full_width=False,
        render_markdown=True,
        avatar_images=("https://avatars.githubusercontent.com/u/9919?s=280&v=4", None),
    )

    with gr.Row():
        msg = gr.Textbox(placeholder="메시지를 입력하세요...", show_label=False, scale=8)
        send = gr.Button("전송", variant="primary", scale=1)
        clear = gr.Button("🧹 초기화", scale=1)

    msg.submit(chat, [msg, chatbot], [chatbot, chatbot, msg])
    send.click(chat, [msg, chatbot], [chatbot, chatbot, msg])
    clear.click(lambda: None, None, chatbot, queue=False)

    gr.Markdown("""
    <div style="text-align:center;color:#777;font-size:0.85em;margin-top:8px;">
    💡 대화 기록은 세션 동안 유지됩니다.  
    Made with ❤️ by PIXAL
    </div>
    """)

if __name__ == "__main__":
    demo.launch(server_name="0.0.0.0", server_port=7860)