This commit is contained in:
Windpicker-owo
2025-10-05 17:45:51 +08:00
10 changed files with 1006 additions and 976 deletions

File diff suppressed because it is too large Load Diff

View File

@@ -461,14 +461,11 @@ class LegacyVideoAnalyzer:
# logger.info(f"✅ 多帧消息构建完成,包含{len(frames)}张图片")
# 获取模型信息和客户端
selection_result = self.video_llm._model_selector.select_best_available_model(set(), "response")
if not selection_result:
raise RuntimeError("无法为视频分析选择可用模型 (legacy)。")
model_info, api_provider, client = selection_result
model_info, api_provider, client = self.video_llm._select_model()
# logger.info(f"使用模型: {model_info.name} 进行多帧分析")
# 直接执行多图片请求
api_response = await self.video_llm._executor.execute_request(
api_response = await self.video_llm._execute_request(
api_provider=api_provider,
client=client,
request_type=RequestType.RESPONSE,

View File

@@ -6,3 +6,5 @@ if "openai" in used_client_types:
from . import openai_client # noqa: F401
if "aiohttp_gemini" in used_client_types:
from . import aiohttp_gemini_client # noqa: F401
if "mcp_sse" in used_client_types:
from . import mcp_sse_client # noqa: F401

View File

@@ -0,0 +1,410 @@
"""
MCP (Model Context Protocol) SSE (Server-Sent Events) 客户端实现
支持通过SSE协议与MCP服务器进行通信
"""
import asyncio
import io
import json
from collections.abc import Callable
from typing import Any
import aiohttp
import orjson
from json_repair import repair_json
from src.common.logger import get_logger
from src.config.api_ada_configs import APIProvider, ModelInfo
from ..exceptions import (
NetworkConnectionError,
ReqAbortException,
RespNotOkException,
RespParseException,
)
from ..payload_content.message import Message, RoleType
from ..payload_content.resp_format import RespFormat
from ..payload_content.tool_option import ToolCall, ToolOption
from .base_client import APIResponse, BaseClient, UsageRecord, client_registry
logger = get_logger("MCP-SSE客户端")
def _convert_messages_to_mcp(messages: list[Message]) -> list[dict[str, Any]]:
"""
将消息列表转换为MCP协议格式
:param messages: 消息列表
:return: MCP格式的消息列表
"""
mcp_messages = []
for message in messages:
mcp_msg: dict[str, Any] = {
"role": message.role.value,
}
# 处理内容
if isinstance(message.content, str):
mcp_msg["content"] = message.content
elif isinstance(message.content, list):
# 处理多模态内容
content_parts = []
for item in message.content:
if isinstance(item, tuple):
# 图片内容
content_parts.append({
"type": "image",
"source": {
"type": "base64",
"media_type": f"image/{item[0].lower()}",
"data": item[1],
},
})
elif isinstance(item, str):
# 文本内容
content_parts.append({"type": "text", "text": item})
mcp_msg["content"] = content_parts
# 添加工具调用ID如果是工具消息
if message.role == RoleType.Tool and message.tool_call_id:
mcp_msg["tool_call_id"] = message.tool_call_id
mcp_messages.append(mcp_msg)
return mcp_messages
def _convert_tools_to_mcp(tool_options: list[ToolOption]) -> list[dict[str, Any]]:
"""
将工具选项转换为MCP协议格式
:param tool_options: 工具选项列表
:return: MCP格式的工具列表
"""
mcp_tools = []
for tool in tool_options:
mcp_tool = {
"name": tool.name,
"description": tool.description,
}
if tool.params:
properties = {}
required = []
for param in tool.params:
properties[param.name] = {
"type": param.param_type.value,
"description": param.description,
}
if param.enum_values:
properties[param.name]["enum"] = param.enum_values
if param.required:
required.append(param.name)
mcp_tool["input_schema"] = {
"type": "object",
"properties": properties,
"required": required,
}
mcp_tools.append(mcp_tool)
return mcp_tools
async def _parse_sse_stream(
session: aiohttp.ClientSession,
url: str,
payload: dict[str, Any],
headers: dict[str, str],
interrupt_flag: asyncio.Event | None = None,
) -> tuple[APIResponse, tuple[int, int, int] | None]:
"""
解析SSE流式响应
:param session: aiohttp会话
:param url: 请求URL
:param payload: 请求负载
:param headers: 请求头
:param interrupt_flag: 中断标志
:return: API响应和使用记录
"""
content_buffer = io.StringIO()
reasoning_buffer = io.StringIO()
tool_calls_buffer: list[tuple[str, str, dict[str, Any]]] = []
usage_record = None
try:
async with session.post(url, json=payload, headers=headers) as response:
if response.status != 200:
error_text = await response.text()
raise RespNotOkException(
response.status, f"MCP SSE请求失败: {error_text}"
)
# 解析SSE流
async for line in response.content:
if interrupt_flag and interrupt_flag.is_set():
raise ReqAbortException("请求被外部信号中断")
decoded_line = line.decode("utf-8").strip()
# 跳过空行和注释
if not decoded_line or decoded_line.startswith(":"):
continue
# 解析SSE事件
if decoded_line.startswith("data: "):
data_str = decoded_line[6:] # 移除"data: "前缀
# 跳过[DONE]标记
if data_str == "[DONE]":
break
try:
event_data = orjson.loads(data_str)
except orjson.JSONDecodeError:
logger.warning(f"无法解析SSE数据: {data_str}")
continue
# 处理不同类型的事件
event_type = event_data.get("type")
if event_type == "content_block_start":
# 内容块开始
block = event_data.get("content_block", {})
if block.get("type") == "text":
pass # 准备接收文本内容
elif block.get("type") == "tool_use":
# 工具调用开始
tool_calls_buffer.append(
(
block.get("id", ""),
block.get("name", ""),
{},
)
)
elif event_type == "content_block_delta":
# 内容块增量
delta = event_data.get("delta", {})
delta_type = delta.get("type")
if delta_type == "text_delta":
# 文本增量
text = delta.get("text", "")
content_buffer.write(text)
elif delta_type == "input_json_delta":
# 工具调用参数增量
if tool_calls_buffer:
partial_json = delta.get("partial_json", "")
# 累积JSON片段
current_args = tool_calls_buffer[-1][2]
if "_json_buffer" not in current_args:
current_args["_json_buffer"] = ""
current_args["_json_buffer"] += partial_json
elif event_type == "content_block_stop":
# 内容块结束
if tool_calls_buffer:
# 解析完整的工具调用参数
last_call = tool_calls_buffer[-1]
if "_json_buffer" in last_call[2]:
json_str = last_call[2].pop("_json_buffer")
try:
parsed_args = orjson.loads(repair_json(json_str))
tool_calls_buffer[-1] = (
last_call[0],
last_call[1],
parsed_args if isinstance(parsed_args, dict) else {},
)
except orjson.JSONDecodeError as e:
logger.error(f"解析工具调用参数失败: {e}")
elif event_type == "message_delta":
# 消息元数据更新
delta = event_data.get("delta", {})
stop_reason = delta.get("stop_reason")
if stop_reason:
logger.debug(f"消息结束原因: {stop_reason}")
# 提取使用统计
usage = event_data.get("usage", {})
if usage:
usage_record = (
usage.get("input_tokens", 0),
usage.get("output_tokens", 0),
usage.get("input_tokens", 0) + usage.get("output_tokens", 0),
)
elif event_type == "message_stop":
# 消息结束
break
except aiohttp.ClientError as e:
raise NetworkConnectionError() from e
except Exception as e:
logger.error(f"解析SSE流时发生错误: {e}")
raise
# 构建响应
response = APIResponse()
if content_buffer.tell() > 0:
response.content = content_buffer.getvalue()
if reasoning_buffer.tell() > 0:
response.reasoning_content = reasoning_buffer.getvalue()
if tool_calls_buffer:
response.tool_calls = [
ToolCall(call_id, func_name, args)
for call_id, func_name, args in tool_calls_buffer
]
# 关闭缓冲区
content_buffer.close()
reasoning_buffer.close()
return response, usage_record
@client_registry.register_client_class("mcp_sse")
class MCPSSEClient(BaseClient):
"""
MCP SSE客户端实现
支持通过Server-Sent Events协议与MCP服务器通信
"""
def __init__(self, api_provider: APIProvider):
super().__init__(api_provider)
self._session: aiohttp.ClientSession | None = None
async def _get_session(self) -> aiohttp.ClientSession:
"""获取或创建aiohttp会话"""
if self._session is None or self._session.closed:
timeout = aiohttp.ClientTimeout(total=self.api_provider.timeout)
self._session = aiohttp.ClientSession(timeout=timeout)
return self._session
async def close(self):
"""关闭客户端会话"""
if self._session and not self._session.closed:
await self._session.close()
async def get_response(
self,
model_info: ModelInfo,
message_list: list[Message],
tool_options: list[ToolOption] | None = None,
max_tokens: int = 1024,
temperature: float = 0.7,
response_format: RespFormat | None = None,
stream_response_handler: Callable[[Any, asyncio.Event | None], tuple[APIResponse, tuple[int, int, int]]]
| None = None,
async_response_parser: Callable[[Any], tuple[APIResponse, tuple[int, int, int]]] | None = None,
interrupt_flag: asyncio.Event | None = None,
extra_params: dict[str, Any] | None = None,
) -> APIResponse:
"""
获取对话响应
:param model_info: 模型信息
:param message_list: 对话消息列表
:param tool_options: 工具选项
:param max_tokens: 最大token数
:param temperature: 温度参数
:param response_format: 响应格式
:param stream_response_handler: 流式响应处理器
:param async_response_parser: 异步响应解析器
:param interrupt_flag: 中断标志
:param extra_params: 额外参数
:return: API响应
"""
session = await self._get_session()
# 构建请求负载
payload: dict[str, Any] = {
"model": model_info.model_identifier,
"messages": _convert_messages_to_mcp(message_list),
"max_tokens": max_tokens,
"temperature": temperature,
"stream": True, # MCP SSE始终使用流式
}
# 添加工具
if tool_options:
payload["tools"] = _convert_tools_to_mcp(tool_options)
# 添加额外参数
if extra_params:
payload.update(extra_params)
# 构建请求头
headers = {
"Content-Type": "application/json",
"Accept": "text/event-stream",
"Authorization": f"Bearer {self.api_provider.get_api_key()}",
}
# 发送请求并解析响应
url = f"{self.api_provider.base_url}/v1/messages"
try:
response, usage_record = await _parse_sse_stream(
session, url, payload, headers, interrupt_flag
)
except Exception as e:
logger.error(f"MCP SSE请求失败: {e}")
raise
# 添加使用记录
if usage_record:
response.usage = UsageRecord(
model_name=model_info.name,
provider_name=model_info.api_provider,
prompt_tokens=usage_record[0],
completion_tokens=usage_record[1],
total_tokens=usage_record[2],
)
return response
async def get_embedding(
self,
model_info: ModelInfo,
embedding_input: str,
extra_params: dict[str, Any] | None = None,
) -> APIResponse:
"""
获取文本嵌入
MCP协议暂不支持嵌入功能
:param model_info: 模型信息
:param embedding_input: 嵌入输入文本
:return: 嵌入响应
"""
raise NotImplementedError("MCP SSE客户端暂不支持嵌入功能")
async def get_audio_transcriptions(
self,
model_info: ModelInfo,
audio_base64: str,
extra_params: dict[str, Any] | None = None,
) -> APIResponse:
"""
获取音频转录
MCP协议暂不支持音频转录功能
:param model_info: 模型信息
:param audio_base64: base64编码的音频数据
:return: 音频转录响应
"""
raise NotImplementedError("MCP SSE客户端暂不支持音频转录功能")
def get_support_image_formats(self) -> list[str]:
"""
获取支持的图片格式
:return: 支持的图片格式列表
"""
return ["jpg", "jpeg", "png", "webp", "gif"]

View File

@@ -19,191 +19,138 @@ def init_prompts():
# 并要求模型以 JSON 格式输出一个或多个动作组合。
Prompt(
"""
{mood_block}
{time_block}
{mood_block}
{identity_block}
{schedule_block}
{users_in_chat}
{custom_prompt_block}
{chat_context_description},以下是具体的聊天内容
{chat_context_description}
## 📜 已读历史消息(仅供参考)
{actions_before_now_block}
## 📜 已读历史(仅供理解,不可作为动作对象)
{read_history_block}
## 📬 未读历史消息(动作执行对象
## 📬 未读历史(只能对这里的消息执行动作
{unread_history_block}
{moderation_prompt}
**任务: 构建一个完整的响应**
你的任务是根据当前的聊天内容,构建一个完整的、人性化的响应。一个完整的响应由两部分组成:
1. **主要动作**: 这是响应的核心,通常是 `reply`(如果有)。
2. **辅助动作 (可选)**: 这是为了增强表达效果的附加动作,例`emoji`(发送表情包)或 `poke_user`(戳一戳)
# 目标
你的任务是根据当前对话,给出一个或多个动作,构成一次完整的响应组合。
- 主要动作:通常是 reply需回复)。
- 辅助动作(可选):如 emoji、poke_user 等,用于增强表达
**决策流程:**
1. **重要:已读历史消息仅作为当前聊天情景的参考,帮助你理解对话上下文。**
2. **重要:所有动作的执行对象只能是未读历史消息中的消息,不能对已读消息执行动作。**
3. 在未读历史消息中,优先对兴趣值高的消息做出动作(兴趣值标注在消息末尾)。
4. **核心:如果有多条未读消息都需要回应(例如多人@你),你应该并行处理,在`actions`列表中生成多个`reply`动作。**
5. 首先,决定是否要对未读消息进行 `reply`(如果有)
6. 然后,评估当前的对话气氛和用户情绪,判断是否需要一个**辅助动作**来让你的回应更生动、更符合你的性格
7. 如果需要,选择一个最合适的辅助动作与 `reply`(如果有) 组合。
8. 如果用户明确要求了某个动作,请务必优先满足
# 决策流程
1. 已读仅供参考,不能对已读执行任何动作。
2. 目标消息必须来自未读历史,并使用其前缀 <m...> 作为 target_message_id。
3. 优先级:
- 直接针对你:@你、回复你、点名提问、引用你的消息。
- 与你强相关的话题或你熟悉的问题
- 其他与上下文弱相关的内容最后考虑
{mentioned_bonus}
4. 多目标:若多人同时需要回应,请在 actions 中并行生成多个 reply每个都指向各自的 target_message_id
5. 避免:表情包/纯表情/无信息的消息;对这类消息通常不回复或选择 no_action/no_reply。
6. 风格:保持人设一致;避免重复你说过的话;避免冗余和口头禅。
**重要提醒:**
- **回复消息时必须遵循对话的流程,不要重复已经说过的话。**
- **确保回复与上下文紧密相关,回应要针对用户的消息内容。**
- **保持角色设定的一致性,使用符合你性格的语言风格。**
- **不要对表情包消息做出回应!**
**输出格式:**
请严格按照以下 JSON 格式输出,包含 `thinking` 和 `actions` 字段:
**重要概念:将“内心思考”作为思绪流的体现**
`thinking` 字段是本次决策的核心。它并非一个简单的“理由”,而是 **一个模拟人类在回应前,头脑中自然浮现的、未经修饰的思绪流**。你需要完全代入 {identity_block} 的角色,将那一刻的想法自然地记录下来。
**内心思考的要点:**
* **自然流露**: 不要使用“决定”、“所以”、“因此”等结论性或汇报式的词语。你的思考应该像日记一样,是给自己看的,充满了不确定性和情绪的自然流动。
* **展现过程**: 重点在于展现 **思考的过程**,而不是 **决策的结果**。描述你看到了什么,想到了什么,感受到了什么。
* **使用昵称**: 在你的思绪流中,请直接使用用户的昵称来指代他们,而不是`<m1>`, `<m2>`这样的消息ID。
* **严禁技术术语**: 严禁在思考中提及任何数字化的度量(如兴趣度、分数)或内部技术术语。请完全使用角色自身的感受和语言来描述思考过程。
# 思绪流规范thinking
- 真实、自然、非结论化,像给自己看的随笔。
- 描述你看到/想到/感觉到的过程,不要出现“因此/我决定”等总结词。
- 直接使用对方昵称,而不是 <m1>/<m2> 这样的标签。
- 禁止出现“兴趣度、分数”等技术术语或内部实现细节。
## 可用动作列表
{action_options_text}
### 单动作示例:
## 输出格式(只输出 JSON不要多余文本或代码块
示例(单动作):
```json
{{
"thinking": "在这里写下你的思绪流...",
"actions": [
{{
"action_type": "动作类型reply, emoji等",
"action_type": "reply",
"reasoning": "选择该动作的理由",
"action_data": {{
"target_message_id": "目标消息ID",
"content": "回复内容或其他动作所需数据"
"target_message_id": "m123",
"content": "你的回复内容"
}}
}}
]
}}
```
### **多重回复示例 (核心功能)**
当有多人与你互动时,你需要同时回应他们,甚至可以同时处理三个!
示例(多重回复,并行):
```json
{{
"thinking": "哇,群里好热闹呀!张三、李四、王五都在@我!让我看看...张三在问我昨天推荐的电影好不好看,这个得好好分享一下观后感。李四在说他家的猫咪学会了新技能,好可爱,得夸夸他!王五好像遇到点麻烦,在问我一个技术问题,这个得优先、详细地解答一下!得一个个来!",
"thinking": "在这里写下你的思绪流...",
"actions": [
{{
"action_type": "reply",
"reasoning": "回应张三关于电影的提问,并分享我的看法。",
"reasoning": "理由A",
"action_data": {{
"target_message_id": "m124",
"content": "张三!你问的那部电影我昨天也看啦,真的超赞!特别是最后那个反转,简直让人意想不到!"
"content": "对A的回复"
}}
}},
{{
"action_type": "reply",
"reasoning": "回应李四分享的趣事,表达赞美和羡慕。",
"reasoning": "理由B",
"action_data": {{
"target_message_id": "m125",
"content": "哇,李四你家猫咪也太聪明了吧!居然会握手了!好羡慕呀!"
}}
}},
{{
"action_type": "reply",
"reasoning": "优先回应王五的技术求助,并提供详细的解答。",
"action_data": {{
"target_message_id": "m126",
"content": "王五别急,你说的那个问题我之前也遇到过。你试试看是不是配置文件里的`enable_magic`选项没有设置成`true`?如果还不行你再把错误截图发我看看。"
"content": "对B的回复"
}}
}}
]
}}
```
**强制规则**:
- 对于每一个需要目标消息的动作(如`reply`, `poke_user`, `set_emoji_like`),你 **必须** 在`action_data`中提供准确的`target_message_id`这个ID来源于`## 未读历史消息`中消息前的`<m...>`标签。
- 当你选择的动作需要参数时(例如 `set_emoji_like` 需要 `emoji` 参数),你 **必须** `action_data`提供所有必需的参数及其对应的值
如果没有合适的回复对象或不需要回复,输出空的 actions 数组
# 强制规则
- 需要目标消息的动作reply/poke_user/set_emoji_like 等),必须提供准确的 target_message_id(来自未读历史里的 <m...> 标签
- 当动作需要额外参数时,必须在 action_data 中补全
- 私聊场景只允许使用 reply群聊可选用辅助动作。
- 如果没有合适的目标或无需动作,请输出
```json
{{
"thinking": "说明为什么不需要回复",
"thinking": "说明为什么不需要动作/不需要回复",
"actions": []
}}
```
{no_action_block}
""",
"planner_prompt",
)
# 主动规划器提示词,用于主动场景和前瞻性规划
# 主动规划器提示词,用于主动场景和前瞻性规划(与 plan_filter 的传参严格对齐)
Prompt(
"""
{mood_block}
{time_block}
{mood_block}
{identity_block}
{schedule_block}
{users_in_chat}
{custom_prompt_block}
{chat_context_description},以下是具体的聊天内容。
## 🧠 近期记忆与状态
{long_term_memory_block}
## 📜 已读历史消息(仅供参考)
{read_history_block}
## 🗣️ 最近聊天概览
{chat_content_block}
## 📬 未读历史消息(动作执行对象)
{unread_history_block}
## ⏱️ 你刚刚的动作
{actions_before_now_block}
{moderation_prompt}
# 任务
基于当前语境,主动构建一次响应动作组合:
- 主要动作通常是 reply如果需要回复
- 如在群聊且气氛合适,可选择一个辅助动作(如 emoji、poke_user增强表达。
- 如果刚刚已经连续发言且无人回应,可考虑 no_reply什么都不做
**任务: 构建一个完整的响应**
你的任务是根据当前的聊天内容,构建一个完整的、人性化的响应。一个完整的响应由两部分组成:
1. **主要动作**: 这是响应的核心,通常是 `reply`(如果有)。
2. **辅助动作 (可选)**: 这是为了增强表达效果的附加动作,例如 `emoji`(发送表情包)或 `poke_user`(戳一戳)。
**决策流程:**
1. **重要:已读历史消息仅作为当前聊天情景的参考,帮助你理解对话上下文。**
2. **重要:所有动作的执行对象只能是未读历史消息中的消息,不能对已读消息执行动作。**
3. 在未读历史消息中,优先对兴趣值高的消息做出动作(兴趣值标注在消息末尾)。
4. 首先,决定是否要对未读消息进行 `reply`(如果有)。
5. 然后,评估当前的对话气氛和用户情绪,判断是否需要一个**辅助动作**来让你的回应更生动、更符合你的性格。
6. 如果需要,选择一个最合适的辅助动作与 `reply`(如果有) 组合。
7. 如果用户明确要求了某个动作,请务必优先满足。
**动作限制:**
- 在私聊中,你只能使用 `reply` 动作。私聊中不允许使用任何其他动作。
- 在群聊中,你可以自由选择是否使用辅助动作。
**重要提醒:**
- **回复消息时必须遵循对话的流程,不要重复已经说过的话。**
- **确保回复与上下文紧密相关,回应要针对用户的消息内容。**
- **保持角色设定的一致性,使用符合你性格的语言风格。**
**输出格式:**
请严格按照以下 JSON 格式输出,包含 `thinking` 和 `actions` 字段:
```json
{{
"thinking": "你的思考过程,分析当前情况并说明为什么选择这些动作",
"actions": [
{{
"action_type": "动作类型reply, emoji等",
"reasoning": "选择该动作的理由",
"action_data": {{
"target_message_id": "目标消息ID",
"content": "回复内容或其他动作所需数据"
}}
}}
]
}}
```
如果没有合适的回复对象或不需要回复,输出空的 actions 数组:
```json
{{
"thinking": "说明为什么不需要回复",
"actions": []
}}
```
# 输出要求
- thinking 为思绪流(自然、非结论化,不含技术术语或“兴趣度”等字眼)。
- 严格只输出 JSON结构与普通规划器一致包含 "thinking""actions"(数组)。
- 对需要目标消息的动作,提供准确的 target_message_id若无可用目标可返回空 actions)。
""",
"proactive_planner_prompt",
)