用户: {user_name}
分析时间: {current_date_time_str}
星期: {current_weekday_zh}
"""
class Action:
class Valves(BaseModel):
show_status: bool = Field(
default=True, description="是否在聊天界面显示操作状态更新。"
)
LLM_MODEL_ID: str = Field(
default="gemini-2.5-flash",
description="用于文本分析的内置LLM模型ID。",
)
MIN_TEXT_LENGTH: int = Field(
default=100, description="进行思维导图分析所需的最小文本长度(字符数)。"
)
def __init__(self):
self.valves = self.Valves()
self.weekday_map = {
"Monday": "星期一",
"Tuesday": "星期二",
"Wednesday": "星期三",
"Thursday": "星期四",
"Friday": "星期五",
"Saturday": "星期六",
"Sunday": "星期日",
}
def _extract_markdown_syntax(self, llm_output: str) -> str:
match = re.search(r"```markdown\s*(.*?)\s*```", llm_output, re.DOTALL)
if match:
extracted_content = match.group(1).strip()
else:
logger.warning(
"LLM输出未严格遵循预期Markdown格式，将整个输出作为摘要处理。"
)
extracted_content = llm_output.strip()
return extracted_content.replace("", "<\\/script>")
async def action(
self,
body: dict,
__user__: Optional[Dict[str, Any]] = None,
__event_emitter__: Optional[Any] = None,
__request__: Optional[Request] = None,
) -> Optional[dict]:
logger.info("Action: 智绘心图 (v12 - Final Feedback Fix) started")
if isinstance(__user__, (list, tuple)):
user_language = (
__user__[0].get("language", "zh-CN") if __user__ else "zh-CN"
)
user_name = __user__[0].get("name", "用户") if __user__[0] else "用户"
user_id = (
__user__[0]["id"]
if __user__ and "id" in __user__[0]
else "unknown_user"
)
elif isinstance(__user__, dict):
user_language = __user__.get("language", "zh-CN")
user_name = __user__.get("name", "用户")
user_id = __user__.get("id", "unknown_user")
try:
shanghai_tz = pytz.timezone("Asia/Shanghai")
current_datetime_shanghai = datetime.now(shanghai_tz)
current_date_time_str = current_datetime_shanghai.strftime(
"%Y-%m-%d %H:%M:%S"
)
current_weekday_en = current_datetime_shanghai.strftime("%A")
current_weekday_zh = self.weekday_map.get(current_weekday_en, "未知星期")
current_year = current_datetime_shanghai.strftime("%Y")
current_timezone_str = "Asia/Shanghai"
except Exception as e:
logger.warning(f"获取时区信息失败: {e}，使用默认值。")
now = datetime.now()
current_date_time_str = now.strftime("%Y-%m-%d %H:%M:%S")
current_weekday_zh = "未知星期"
current_year = now.strftime("%Y")
current_timezone_str = "未知时区"
if __event_emitter__:
await __event_emitter__(
{
"type": "notification",
"data": {
"type": "info",
"content": "智绘心图已启动，正在为您生成思维导图...",
},
}
)
messages = body.get("messages")
if (
not messages
or not isinstance(messages, list)
or not messages[-1].get("content")
):
error_message = "无法获取有效的用户消息内容。"
if __event_emitter__:
await __event_emitter__(
{
"type": "notification",
"data": {"type": "error", "content": error_message},
}
)
return {
"messages": [{"role": "assistant", "content": f"❌ {error_message}"}]
}
parts = re.split(r"```html.*?```", messages[-1]["content"], flags=re.DOTALL)
long_text_content = ""
if parts:
for part in reversed(parts):
if part.strip():
long_text_content = part.strip()
break
if not long_text_content:
long_text_content = messages[-1]["content"].strip()
if len(long_text_content) < self.valves.MIN_TEXT_LENGTH:
short_text_message = f"文本内容过短({len(long_text_content)}字符)，无法进行有效分析。请提供至少{self.valves.MIN_TEXT_LENGTH}字符的文本。"
if __event_emitter__:
await __event_emitter__(
{
"type": "notification",
"data": {"type": "warning", "content": short_text_message},
}
)
return {
"messages": [
{"role": "assistant", "content": f"⚠️ {short_text_message}"}
]
}
if self.valves.show_status and __event_emitter__:
await __event_emitter__(
{
"type": "status",
"data": {
"description": "智绘心图: 深入分析文本结构...",
"done": False,
"hidden": False,
},
}
)
try:
unique_id = f"id_{int(time.time() * 1000)}"
formatted_user_prompt = USER_PROMPT_GENERATE_MINDMAP.format(
user_name=user_name,
current_date_time_str=current_date_time_str,
current_weekday=current_weekday_zh,
current_timezone_str=current_timezone_str,
user_language=user_language,
long_text_content=long_text_content,
)
llm_payload = {
"model": self.valves.LLM_MODEL_ID,
"messages": [
{"role": "system", "content": SYSTEM_PROMPT_MINDMAP_ASSISTANT},
{"role": "user", "content": formatted_user_prompt},
],
"temperature": 0.5,
"stream": False,
}
user_obj = Users.get_user_by_id(user_id)
if not user_obj:
raise ValueError(f"无法获取用户对象，用户ID: {user_id}")
llm_response = await generate_chat_completion(
__request__, llm_payload, user_obj
)
if (
not llm_response
or "choices" not in llm_response
or not llm_response["choices"]
):
raise ValueError("LLM响应格式不正确或为空。")
assistant_response_content = llm_response["choices"][0]["message"][
"content"
]
markdown_syntax = self._extract_markdown_syntax(assistant_response_content)
final_html_content = (
HTML_TEMPLATE_MINDMAP.replace("{unique_id}", unique_id)
.replace("{user_language}", user_language)
.replace("{user_name}", user_name)
.replace("{current_date_time_str}", current_date_time_str)
.replace("{current_weekday_zh}", current_weekday_zh)
.replace("{current_year}", current_year)
.replace("{markdown_syntax}", markdown_syntax)
)
html_embed_tag = f"```html\n{final_html_content}\n```"
body["messages"][-1]["content"] = f"{long_text_content}\n\n{html_embed_tag}"
if self.valves.show_status and __event_emitter__:
await __event_emitter__(
{
"type": "status",
"data": {
"description": "智绘心图: 绘制完成！",
"done": True,
"hidden": False,
},
}
)
await __event_emitter__(
{
"type": "notification",
"data": {
"type": "success",
"content": f"思维导图已生成，{user_name}！",
},
}
)
logger.info("Action: 智绘心图 (v12) completed successfully")
except Exception as e:
error_message = f"智绘心图处理失败: {str(e)}"
logger.error(f"智绘心图错误: {error_message}", exc_info=True)
user_facing_error = f"抱歉，智绘心图在处理时遇到错误: {str(e)}。\n请检查Open WebUI后端日志获取更多详情。"
body["messages"][-1][
"content"
] = f"{long_text_content}\n\n❌ **错误:** {user_facing_error}"
if __event_emitter__:
if self.valves.show_status:
await __event_emitter__(
{
"type": "status",
"data": {
"description": "智绘心图: 处理失败。",
"done": True,
"hidden": False,
},
}
)
await __event_emitter__(
{
"type": "notification",
"data": {
"type": "error",
"content": f"智绘心图生成失败, {user_name}！",
},
}
)
return body