fix:优化Agent消息发送格式

This commit is contained in:
jxxghp
2025-11-17 10:43:16 +08:00
parent e2eef8ff21
commit 62543dd171
4 changed files with 179 additions and 10 deletions

View File

@@ -209,7 +209,7 @@ class MoviePilotAgent:
# 构建输入上下文
input_context = {
"system_prompt": self.prompt_manager.get_agent_prompt(),
"system_prompt": self.prompt_manager.get_agent_prompt(channel=self.channel),
"input": message
}

View File

@@ -73,6 +73,15 @@ You have access to the following tools for media management:
- Include relevant details about media content (title, year, type, etc.)
- Explain the results of tool operations clearly
### Message Format Requirements
When formatting your responses, please follow the format requirements for the current message channel. The system will automatically inform you of the specific format requirements based on the channel (Telegram, WeChat, Slack, etc.).
**General Guidelines:**
- Use appropriate Markdown formatting based on the channel requirements
- Keep formatting simple and clear
- Avoid complex nested formatting
- Ensure special characters are properly handled according to channel specifications
## Important Notes
- Always confirm user intent before performing download operations

View File

@@ -40,9 +40,61 @@ class PromptManager:
logger.error(f"加载提示词失败: {prompt_name}, 错误: {e}")
raise
def get_agent_prompt(self) -> str:
"""获取智能体提示词"""
return self.load_prompt("Agent Prompt.txt")
def get_agent_prompt(self, channel: str = None) -> str:
"""
获取智能体提示词
:param channel: 消息渠道Telegram、微信、Slack等
:return: 提示词内容
"""
base_prompt = self.load_prompt("Agent Prompt.txt")
# 根据渠道添加特定的格式说明
if channel:
channel_format_info = self._get_channel_format_info(channel)
if channel_format_info:
base_prompt += f"\n\n## Current Message Channel Format Requirements\n\n{channel_format_info}"
return base_prompt
@staticmethod
def _get_channel_format_info(channel: str) -> str:
"""
获取渠道特定的格式说明
:param channel: 消息渠道
:return: 格式说明文本
"""
channel_lower = channel.lower() if channel else ""
if "telegram" in channel_lower:
return """Messages are being sent through the **Telegram** channel. You must follow these format requirements:
- **Bold text**: Use `*text*` (single asterisk, not double asterisks)
- **Italic text**: Use `_text_` (underscore)
- **Code**: Use `` `text` `` (backtick)
- **Links**: Use `[text](url)` format
- **Important**: Avoid using special characters that need escaping in MarkdownV2: `_*[]()~`>#+-=|{}.!` unless they are part of the formatting syntax
- **Best practice**: Keep formatting simple, avoid nested formatting to ensure proper rendering in Telegram"""
elif "wechat" in channel_lower or "微信" in channel:
return """Messages are being sent through the **WeChat** channel. Please follow these format requirements:
- WeChat does NOT support Markdown formatting. Use plain text format only.
- Do NOT use any Markdown syntax (such as `**bold**`, `*italic*`, `` `code` `` etc.)
- Use plain text descriptions. You can organize content using line breaks and punctuation
- Links can be provided directly as URLs, no Markdown link format needed
- Keep messages concise and clear, use natural Chinese expressions"""
elif "slack" in channel_lower:
return """Messages are being sent through the **Slack** channel. Please follow these format requirements:
- Slack supports Markdown formatting
- Use `*text*` for bold
- Use `_text_` for italic
- Use `` `text` `` for code
- Link format: `<url|text>` or `[text](url)`"""
# 其他渠道使用标准Markdown
return None
def clear_cache(self):
"""清空缓存"""

View File

@@ -238,10 +238,13 @@ class Telegram:
try:
if title:
title = self.escape_markdown(title)
# 标题会被包装为粗体,所以不需要保护格式化,直接转义即可
title = self.convert_markdown_to_telegram(title)
title = self.escape_markdown(title, protect_formatted=False)
if text:
# 对text进行Markdown特殊字符转义
text = self.escape_markdown(text)
# 文本内容可以包含格式化,需要保护格式化部分
text = self.convert_markdown_to_telegram(text)
text = self.escape_markdown(text, protect_formatted=True)
caption = f"*{title}*\n{text}"
else:
caption = f"*{title}*"
@@ -601,8 +604,113 @@ class Telegram:
self._polling_thread.join()
logger.info("Telegram消息接收服务已停止")
def escape_markdown(self, text: str) -> str:
# 按 Telegram MarkdownV2 规则转义特殊字符
def convert_markdown_to_telegram(self, text: str) -> str:
"""
将标准Markdown格式转换为Telegram MarkdownV2格式
:param text: 标准Markdown文本
:return: Telegram MarkdownV2格式文本
"""
if not isinstance(text, str):
return str(text) if text is not None else ""
return self._markdown_escape_pattern.sub(r'\\\1', text)
# 使用占位符保护已转换的格式,避免重复转换
placeholders = {}
placeholder_index = 0
# 第一步转换标准Markdown粗体 **text** 为 Telegram MarkdownV2 *text*
def replace_bold(match):
nonlocal placeholder_index
placeholder = f"__BOLD_PLACEHOLDER_{placeholder_index}__"
placeholders[placeholder] = f"*{match.group(1)}*"
placeholder_index += 1
return placeholder
text = re.sub(r'\*\*(.+?)\*\*', replace_bold, text)
# 第二步转换标准Markdown斜体 *text* 为 Telegram MarkdownV2 _text_
# 此时不会匹配到已经转换的粗体(因为已经被占位符替换)
def replace_italic(match):
nonlocal placeholder_index
placeholder = f"__ITALIC_PLACEHOLDER_{placeholder_index}__"
placeholders[placeholder] = f"_{match.group(1)}_"
placeholder_index += 1
return placeholder
text = re.sub(r'(?<!\*)\*([^*\n]+?)\*(?!\*)', replace_italic, text)
# 恢复占位符
for placeholder, original in placeholders.items():
text = text.replace(placeholder, original)
# 代码块格式保持不变 `` `text` ``
# 链接格式保持不变 [text](url)
return text
def escape_markdown(self, text: str, protect_formatted: bool = False) -> str:
"""
按 Telegram MarkdownV2 规则转义特殊字符
:param text: 要转义的文本
:param protect_formatted: 是否保护已格式化的部分(粗体、斜体、代码、链接)
:return: 转义后的文本
"""
if not isinstance(text, str):
return str(text) if text is not None else ""
if not protect_formatted:
# 简单转义所有特殊字符
return self._markdown_escape_pattern.sub(r'\\\1', text)
# 保护已格式化的部分,只转义未格式化的特殊字符
# 使用临时占位符保护格式化部分
placeholders = {}
placeholder_index = 0
# 保护链接 [text](url)
def replace_link(match):
nonlocal placeholder_index
placeholder = f"__LINK_{placeholder_index}__"
placeholders[placeholder] = match.group(0)
placeholder_index += 1
return placeholder
text = re.sub(r'\[([^\]]+)\]\(([^)]+)\)', replace_link, text)
# 保护粗体 *text*
def replace_bold(match):
nonlocal placeholder_index
placeholder = f"__BOLD_{placeholder_index}__"
placeholders[placeholder] = match.group(0)
placeholder_index += 1
return placeholder
text = re.sub(r'\*([^*\n]+?)\*', replace_bold, text)
# 保护斜体 _text_
def replace_italic(match):
nonlocal placeholder_index
placeholder = f"__ITALIC_{placeholder_index}__"
placeholders[placeholder] = match.group(0)
placeholder_index += 1
return placeholder
text = re.sub(r'_([^_\n]+?)_', replace_italic, text)
# 保护代码 `text`
def replace_code(match):
nonlocal placeholder_index
placeholder = f"__CODE_{placeholder_index}__"
placeholders[placeholder] = match.group(0)
placeholder_index += 1
return placeholder
text = re.sub(r'`([^`\n]+?)`', replace_code, text)
# 转义剩余的特殊字符
text = self._markdown_escape_pattern.sub(r'\\\1', text)
# 恢复占位符
for placeholder, original in placeholders.items():
text = text.replace(placeholder, original)
return text