319 lines
11 KiB
Python
319 lines
11 KiB
Python
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="完成全部响应，请提交新问题。")
|