from .bridge_newbingfree import preprocess_newbing_out, preprocess_newbing_out_simple from multiprocessing import Process, Pipe from toolbox import update_ui, get_conf, trimmed_format_exc import threading import importlib import logging import time from toolbox import get_conf import asyncio load_message = "正在加载Claude组件，请稍候..." try: """ =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=- 第一部分：Slack API Client https://github.com/yokonsan/claude-in-slack-api =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=- """ from slack_sdk.errors import SlackApiError from slack_sdk.web.async_client import AsyncWebClient class SlackClient(AsyncWebClient): """SlackClient类用于与Slack API进行交互，实现消息发送、接收等功能。 属性： - CHANNEL_ID：str类型，表示频道ID。 方法： - open_channel()：异步方法。通过调用conversations_open方法打开一个频道，并将返回的频道ID保存在属性CHANNEL_ID中。 - chat(text: str)：异步方法。向已打开的频道发送一条文本消息。 - get_slack_messages()：异步方法。获取已打开频道的最新消息并返回消息列表，目前不支持历史消息查询。 - get_reply()：异步方法。循环监听已打开频道的消息，如果收到"Typing…_"结尾的消息说明Claude还在继续输出，否则结束循环。 """ CHANNEL_ID = None async def open_channel(self): response = await self.conversations_open( users=get_conf("SLACK_CLAUDE_BOT_ID") ) self.CHANNEL_ID = response["channel"]["id"] async def chat(self, text): if not self.CHANNEL_ID: raise Exception("Channel not found.") resp = await self.chat_postMessage(channel=self.CHANNEL_ID, text=text) self.LAST_TS = resp["ts"] async def get_slack_messages(self): try: # TODO：暂时不支持历史消息，因为在同一个频道里存在多人使用时历史消息渗透问题 resp = await self.conversations_history( channel=self.CHANNEL_ID, oldest=self.LAST_TS, limit=1 ) msg = [ msg for msg in resp["messages"] if msg.get("user") == get_conf("SLACK_CLAUDE_BOT_ID") ] return msg except (SlackApiError, KeyError) as e: raise RuntimeError(f"获取Slack消息失败。") async def get_reply(self): while True: slack_msgs = await self.get_slack_messages() if len(slack_msgs) == 0: await asyncio.sleep(0.5) continue msg = slack_msgs[-1] if msg["text"].endswith("Typing…_"): yield False, msg["text"] else: yield True, msg["text"] break except: pass """ =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=- 第二部分：子进程Worker（调用主体） =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=- """ class ClaudeHandle(Process): def __init__(self): super().__init__(daemon=True) self.parent, self.child = Pipe() self.claude_model = None self.info = "" self.success = True self.local_history = [] self.check_dependency() if self.success: self.start() self.threadLock = threading.Lock() def check_dependency(self): try: self.success = False import slack_sdk self.info = "依赖检测通过，等待Claude响应。注意目前不能多人同时调用Claude接口（有线程锁），否则将导致每个人的Claude问询历史互相渗透。调用Claude时，会自动使用已配置的代理。" self.success = True except: self.info = "缺少的依赖，如果要使用Claude，除了基础的pip依赖以外，您还需要运行`pip install -r request_llms/requirements_slackclaude.txt`安装Claude的依赖，然后重启程序。" self.success = False def ready(self): return self.claude_model is not None async def async_run(self): await self.claude_model.open_channel() while True: # 等待 kwargs = self.child.recv() question = kwargs["query"] history = kwargs["history"] # 开始问问题 prompt = "" # 问题 prompt += question print("question:", prompt) # 提交 await self.claude_model.chat(prompt) # 获取回复 async for final, response in self.claude_model.get_reply(): if not final: print(response) self.child.send(str(response)) else: # 防止丢失最后一条消息 slack_msgs = await self.claude_model.get_slack_messages() last_msg = ( slack_msgs[-1]["text"] if slack_msgs and len(slack_msgs) > 0 else "" ) if last_msg: self.child.send(last_msg) print("-------- receive final ---------") self.child.send("[Finish]") def run(self): """ 这个函数运行在子进程 """ # 第一次运行，加载参数 self.success = False self.local_history = [] if (self.claude_model is None) or (not self.success): # 代理设置 proxies = get_conf("proxies") if proxies is None: self.proxies_https = None else: self.proxies_https = proxies["https"] try: SLACK_CLAUDE_USER_TOKEN = get_conf("SLACK_CLAUDE_USER_TOKEN") self.claude_model = SlackClient( token=SLACK_CLAUDE_USER_TOKEN, proxy=self.proxies_https ) print("Claude组件初始化成功。") except: self.success = False tb_str = "\n```\n" + trimmed_format_exc() + "\n```\n" self.child.send(f"[Local Message] 不能加载Claude组件。{tb_str}") self.child.send("[Fail]") self.child.send("[Finish]") raise RuntimeError(f"不能加载Claude组件。") self.success = True try: # 进入任务等待状态 asyncio.run(self.async_run()) except Exception: tb_str = "\n```\n" + trimmed_format_exc() + "\n```\n" self.child.send(f"[Local Message] Claude失败 {tb_str}.") self.child.send("[Fail]") self.child.send("[Finish]") def stream_chat(self, **kwargs): """ 这个函数运行在主进程 """ self.threadLock.acquire() self.parent.send(kwargs) # 发送请求到子进程 while True: res = self.parent.recv() # 等待Claude回复的片段 if res == "[Finish]": break # 结束 elif res == "[Fail]": self.success = False break else: yield res # Claude回复的片段 self.threadLock.release() """ =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=- 第三部分：主进程统一调用函数接口 =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=- """ global claude_handle claude_handle = None def predict_no_ui_long_connection( inputs, llm_kwargs, history=[], sys_prompt="", observe_window=None, console_slience=False, ): """ 多线程方法 函数的说明请见 request_llms/bridge_all.py """ global claude_handle if (claude_handle is None) or (not claude_handle.success): claude_handle = ClaudeHandle() observe_window[0] = load_message + "\n\n" + claude_handle.info if not claude_handle.success: error = claude_handle.info claude_handle = None raise RuntimeError(error) # 没有 sys_prompt 接口，因此把prompt加入 history history_feedin = [] for i in range(len(history) // 2): history_feedin.append([history[2 * i], history[2 * i + 1]]) watch_dog_patience = 5 # 看门狗 (watchdog) 的耐心, 设置5秒即可 response = "" observe_window[0] = "[Local Message] 等待Claude响应中 ..." for response in claude_handle.stream_chat( query=inputs, history=history_feedin, system_prompt=sys_prompt, max_length=llm_kwargs["max_length"], top_p=llm_kwargs["top_p"], temperature=llm_kwargs["temperature"], ): observe_window[0] = preprocess_newbing_out_simple(response) if len(observe_window) >= 2: if (time.time() - observe_window[1]) > watch_dog_patience: raise RuntimeError("程序终止。") return preprocess_newbing_out_simple(response) def predict( inputs, llm_kwargs, plugin_kwargs, chatbot, history=[], system_prompt="", stream=True, additional_fn=None, ): """ 单线程方法 函数的说明请见 request_llms/bridge_all.py """ chatbot.append((inputs, "[Local Message] 等待Claude响应中 ...")) global claude_handle if (claude_handle is None) or (not claude_handle.success): claude_handle = ClaudeHandle() chatbot[-1] = (inputs, load_message + "\n\n" + claude_handle.info) yield from update_ui(chatbot=chatbot, history=[]) if not claude_handle.success: claude_handle = None return if additional_fn is not None: from core_functional import handle_core_functionality inputs, history = handle_core_functionality( additional_fn, inputs, history, chatbot ) history_feedin = [] for i in range(len(history) // 2): history_feedin.append([history[2 * i], history[2 * i + 1]]) chatbot[-1] = (inputs, "[Local Message] 等待Claude响应中 ...") response = "[Local Message] 等待Claude响应中 ..." yield from update_ui( chatbot=chatbot, history=history, msg="Claude响应缓慢，尚未完成全部响应，请耐心完成后再提交新问题。" ) for response in claude_handle.stream_chat( query=inputs, history=history_feedin, system_prompt=system_prompt ): chatbot[-1] = (inputs, preprocess_newbing_out(response)) yield from update_ui( chatbot=chatbot, history=history, msg="Claude响应缓慢，尚未完成全部响应，请耐心完成后再提交新问题。" ) if response == "[Local Message] 等待Claude响应中 ...": response = "[Local Message] Claude响应异常，请刷新界面重试 ..." history.extend([inputs, response]) logging.info(f"[raw_input] {inputs}") logging.info(f"[response] {response}") yield from update_ui(chatbot=chatbot, history=history, msg="完成全部响应，请提交新问题。")