From 2291a67cf8779000f9532a991a8b5e73cf4b274f Mon Sep 17 00:00:00 2001 From: Rid7 Date: Mon, 15 May 2023 14:27:31 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=9E=E7=8E=B0Claude=E8=81=8A=E5=A4=A9?= =?UTF-8?q?=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- request_llm/bridge_claude.py | 296 ++++++++++++++++++++++++++++ request_llm/requirements_claude.txt | 1 + 2 files changed, 297 insertions(+) create mode 100644 request_llm/bridge_claude.py create mode 100644 request_llm/requirements_claude.txt diff --git a/request_llm/bridge_claude.py b/request_llm/bridge_claude.py new file mode 100644 index 0000000..f2511b0 --- /dev/null +++ b/request_llm/bridge_claude.py @@ -0,0 +1,296 @@ +from .bridge_newbing 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 +from slack_sdk.errors import SlackApiError +from slack_sdk.web.async_client import AsyncWebClient +import asyncio +import sys +sys.path.append('..') + + +""" +======================================================================== +第一部分：Slack API Client +https://github.com/yokonsan/claude-in-slack-api +======================================================================== +""" +load_message = "正在加载Claude组件，请稍候..." + + +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('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('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 + + +""" +======================================================================== +第二部分：子进程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() + 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_claude.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'] + # system_prompt=kwargs['system_prompt'] + + # 是否重置 + if len(self.local_history) > 0 and len(history) == 0: + await self.claude_model.reset() + self.local_history = [] + + # 开始问问题 + prompt = "" + # Slack API最好不要添加系统提示 + # if system_prompt not in self.local_history: + # self.local_history.append(system_prompt) + # prompt += system_prompt + '\n' + + # 追加历史 + for ab in history: + a, b = ab + if a not in self.local_history: + self.local_history.append(a) + prompt += a + '\n' + # if b not in self.local_history: + # self.local_history.append(b) + # prompt += b + '\n' + + # 问题 + prompt += question + self.local_history.append(question) + print('question:', prompt) + # 提交 + await self.claude_model.chat(prompt) + # 获取回复 + # async for final, response in self.claude_model.get_reply(): + # await self.handle_claude_response(final, response) + 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_USER_TOKEN, = get_conf('SLACK_USER_TOKEN') + self.claude_model = SlackClient(token=SLACK_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' + trimmed_format_exc() + '```' + 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: + import core_functional + importlib.reload(core_functional) # 热更新prompt + core_functional = core_functional.get_core_functions() + if "PreProcess" in core_functional[additional_fn]: + inputs = core_functional[additional_fn]["PreProcess"]( + inputs) # 获取预处理函数（如果有的话） + inputs = core_functional[additional_fn]["Prefix"] + \ + inputs + core_functional[additional_fn]["Suffix"] + + 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="完成全部响应，请提交新问题。") diff --git a/request_llm/requirements_claude.txt b/request_llm/requirements_claude.txt new file mode 100644 index 0000000..472d58c --- /dev/null +++ b/request_llm/requirements_claude.txt @@ -0,0 +1 @@ +slack-sdk==3.21.3 \ No newline at end of file