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.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. 커스텀 콜백 핸들러
class StreamingAgentCallbackHandler(BaseCallbackHandler):
    def __init__(self, q: queue.Queue):
        self.q = q

    def on_agent_action(self, action, **kwargs):
        log = f"🧠 Thought: {action.log.strip()}\n🔧 Action: {action.tool}({action.tool_input})"
        self.q.put(log)

    def on_tool_end(self, output, **kwargs):
        self.q.put(f"📦 Observation: {output}\n")

    def on_agent_finish(self, finish, **kwargs):
        self.q.put(f"\n✅ Final Answer: {finish.return_values['output']}")

    def on_llm_new_token(self, token: str, **kwargs):
        # 선택적으로 토큰 스트리밍 출력
        pass

    def on_llm_end(self, response, **kwargs):
        self.q.put("[END]")

    def on_llm_error(self, error, **kwargs):
        self.q.put(f"[ERROR] {str(error)}")

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 코드를 실행합니다."))

# ──────────────────────────────
# ✅ 파일 도구
# ──────────────────────────────


# ──────────────────────────────
# ✅ 정확한 한국 시간 함수 (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 초기화
# ──────────────────────────────
q_stream = queue.Queue()
handler = StreamingAgentCallbackHandler(q_stream)
agent = initialize_agent(
    tools,
    llm,
    memory=MEM(),
    agent_type=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True,
    handle_parsing_errors=True,
    callbacks=[handler]
)

# ──────────────────────────────
# ✅ Gradio UI
# ──────────────────────────────
def chat(message, history):
    try:
        response = agent.run(message)
    except Exception as e:
        response = f"⚠️ 오류: {e}"
    history = history + [(message, response)]
    while True:
        msg = q_stream.get()
        if msg == "[END]":
            break
        yield msg + "\n"

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
    """)
    ai1=gr.ChatInterface(chat)

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