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')[0]) 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')[0]] 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_llm/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_llm/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_llm/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="完成全部响应，请提交新问题。")