app.py

# pixal_agent_full.py
import os
import datetime
import gradio as gr
import requests
from typing import Optional, List
from langchain.llms.base import LLM
from langchain.agents import initialize_agent, AgentType,load_tools
from langchain.agents import AgentExecutor, create_structured_chat_agent
from langchain.tools import Tool
from langchain_experimental.tools.python.tool import PythonREPLTool
import queue
from typing import Any, Dict
import gradio as gr
from langchain.callbacks.base import BaseCallbackHandler


from langchain.tools import YouTubeSearchTool as YTS
# 2. 커스텀 콜백 핸들러


from langchain_community.retrievers import WikipediaRetriever
from langchain.tools.retriever import create_retriever_tool
retriever = WikipediaRetriever(lang="ko",top_k_results=10)
wiki=Tool(func=retriever.get_relevant_documents,name="WIKI SEARCH",description="위키백과에서 필요한 정보를 불러옵니다.결괴를 검증하여 사용하시오.")
# ──────────────────────────────
# ✅ GitHub Models LLM
# ──────────────────────────────
class GitHubModelLLM(LLM):
    model: str = "openai/gpt-4.1"
    endpoint: str = "https://models.github.ai/inference"
    token: Optional[str] = None

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

    def _call(self, prompt: str, stop: Optional[List[str]] = None) -> str:
        if not self.token:
            raise ValueError("GitHub API token이 필요합니다.")

        headers = {
            "Authorization": "Bearer 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",
            "Content-Type": "application/json",
        }
        body = {"model": self.model, "messages": [{"role": "user", "content": prompt}]}

        resp = requests.post(f"{self.endpoint}/chat/completions", json=body, headers=headers)
        if resp.status_code != 200:
            raise ValueError(f"API 오류: {resp.status_code} - {resp.text}")
        return resp.json()["choices"][0]["message"]["content"]

# ──────────────────────────────
# ✅ LLM 설정
# ──────────────────────────────
token = os.getenv("GITHUB_TOKEN") or os.getenv("token")
if not token:
    print("⚠️ GitHub Token이 필요합니다. 예: setx GITHUB_TOKEN your_token")

llm = GitHubModelLLM(model="openai/gpt-4.1", token=token)

# ──────────────────────────────
# ✅ LangChain 기본 도구 불러오기
# ──────────────────────────────
tools = load_tools(
    ["ddg-search", "requests_all", "llm-math"],
    llm=llm,allow_dangerous_tools=True
)+[YTS()]+[wiki]
# ──────────────────────────────
# ✅ Python 실행 도구 (LangChain 내장)
# ──────────────────────────────
python_tool = PythonREPLTool()
tools.append(Tool(name="python_repl", func=python_tool.run, description="Python 코드를 실행합니다."))
from langchain import hub
prompt=hub.pull("hwchase17/structured-chat-agent")
# ──────────────────────────────
# ✅ 파일 도구
# ──────────────────────────────


# ──────────────────────────────
# ✅ 정확한 한국 시간 함수 (Asia/Seoul)
# ──────────────────────────────
import requests
from datetime import datetime
from zoneinfo import ZoneInfo

def time_now(_=""):
    try:
        # 정확한 UTC 시각을 외부 API에서 가져옴
        resp = requests.get("https://timeapi.io/api/Time/current/zone?timeZone=Asia/Seoul", timeout=5)
        if resp.status_code == 200:
            data = resp.json()
            dt = data["dateTime"].split(".")[0].replace("T", " ")
            return f"현재 시각: {dt} (Asia/Seoul, 서버 기준 NTP 동기화)"
        else:
            # API 실패 시 로컬 시스템 시각으로 대체
            tz = ZoneInfo("Asia/Seoul")
            now = datetime.now(tz)
            return f"현재 시각(로컬): {now.strftime('%Y-%m-%d %H:%M:%S')} (Asia/Seoul)"
    except Exception as e:
        tz = ZoneInfo("Asia/Seoul")
        now = datetime.now(tz)
        return f"현재 시각(백업): {now.strftime('%Y-%m-%d %H:%M:%S')} (Asia/Seoul, 오류: {e})"
# ──────────────────────────────
# ✅ 도구 등록
# ──────────────────────────────
tools.extend([Tool(name="time_now", func=time_now, description="현재 시간을 반환합니다.")])
from langchain.memory import ConversationBufferMemory as MEM
from langchain.agents.agent_toolkits import FileManagementToolkit as FMT
tools.extend(FMT(root_dir=str(os.getcwd())).get_tools())
# ──────────────────────────────
# ✅ Agent 초기화
# ──────────────────────────────
mem=MEM()
agent = create_structured_chat_agent(llm, tools, prompt)
agent= AgentExecutor(agent=agent, tools=tools,memory=mem)


# ──────────────────────────────
# ✅ Gradio UI
# ──────────────────────────────
def chat(message, history):
    try:
        response = agent.run(message)
    except Exception as e:
        response = f"⚠️ 오류: {e}"
    history = history + [(message, response)]
    return history,history

with gr.Blocks(theme=gr.themes.Soft(), title="PIXAL Assistant (LangChain + GitHub LLM)") as demo:
    gr.Markdown("""
    ## 🤖 PIXAL Assistant  
    **LangChain 기반 멀티툴 에이전트**  
    🧰 DuckDuckGo · Wikipedia · Math · Requests · Python REPL · File · Time
    """)
    chatbot = gr.Chatbot(label="PIXAL 대화", height=600)
    msg = gr.Textbox(label="메시지", placeholder="명령 또는 질문을 입력하세요...")
    clear = gr.Button("초기화")

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


if __name__ == "__main__":
   demo.launch()