# app_pixal_chat.py import os, re, json, gradio as gr, requests from typing import Optional, List, Dict, Any from requests.adapters import HTTPAdapter, Retry from langchain.llms.base import LLM from langchain.memory import ConversationBufferMemory from langchain.agents import initialize_agent, AgentType, load_tools from langchain.tools import Tool from langchain_experimental.tools.python.tool import PythonREPLTool from langchain_community.retrievers import WikipediaRetriever import datetime # ────────────────────────────── # ✅ GitHubModelLLM (그대로 유지) # ────────────────────────────── from langchain.chat_models.base import BaseChatModel 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(BaseChatModel): """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 _generate(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 _agenerate(self, messages: List[Any], stop: Optional[List[str]] = None, **kwargs): return self._generate(messages, stop, **kwargs) from typing import Optional, List, Dict, Any from langchain.llms.base import LLM 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 도구 & 에이전트 구성 # ────────────────────────────── llm = GitHubModelLLM() tools = load_tools(["ddg-search", "requests_all", "llm-math"], 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="위키백과 검색")) tools.append(Tool(name="time_now", func=lambda _: f"현재 시각: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')} (Asia/Seoul)", description="현재 시간을 반환합니다.")) # ✅ 대화 기억 메모리 memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True) # ✅ Agent (Memory 연동) agent = initialize_agent( tools, llm, agent_type=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION, memory=memory, verbose=True ) # ────────────────────────────── # ✅ Chat 함수 (Memory 유지) # ────────────────────────────── def chat(message, history): raw = agent.run(message) try: # 대화 기록을 LangChain memory에 반영 memory.chat_memory.add_user_message(message) # 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에 추가 memory.chat_memory.add_ai_message(text) 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("""

🤖 PIXAL Assistant

LangChain + GitHub LLM
""") 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("""
💡 대화 기록은 세션 동안 유지됩니다. Made with ❤️ by PIXAL
""") if __name__ == "__main__": demo.launch(server_name="0.0.0.0", server_port=7860)