""" 聊天视图组件 处理普通对话的 UI 展示 - 支持流式消息、加载动画和 Markdown 渲染 """ import tkinter as tk from tkinter import scrolledtext from typing import Callable, Optional, List, Tuple import re import webbrowser class MarkdownRenderer: """Markdown 渲染器 - 将 Markdown 文本渲染到 Text 组件""" # URL 正则表达式 URL_PATTERN = re.compile( r'https?://[^\s<>\[\]()()\u4e00-\u9fff]+' ) # Markdown 链接模式 [text](url) MD_LINK_PATTERN = re.compile(r'\[([^\]]+)\]\(([^)]+)\)') def __init__(self, text_widget: tk.Text): self.text_widget = text_widget self._link_count = 0 self._configure_tags() def _configure_tags(self): """配置 Markdown 样式标签""" # 标题样式 self.text_widget.tag_configure('md_h1', font=('Microsoft YaHei UI', 16, 'bold'), foreground='#4fc3f7') self.text_widget.tag_configure('md_h2', font=('Microsoft YaHei UI', 14, 'bold'), foreground='#4fc3f7') self.text_widget.tag_configure('md_h3', font=('Microsoft YaHei UI', 12, 'bold'), foreground='#4fc3f7') # 粗体和斜体 self.text_widget.tag_configure('md_bold', font=('Microsoft YaHei UI', 11, 'bold')) self.text_widget.tag_configure('md_italic', font=('Microsoft YaHei UI', 11, 'italic')) # 代码样式 self.text_widget.tag_configure('md_code', font=('Consolas', 10), background='#3c3c3c', foreground='#ce9178') self.text_widget.tag_configure('md_code_block', font=('Consolas', 10), background='#1e1e1e', foreground='#d4d4d4') # 列表样式 self.text_widget.tag_configure('md_list', foreground='#d4d4d4', lmargin1=20, lmargin2=35) self.text_widget.tag_configure('md_list_bullet', foreground='#ffd54f') # 链接样式 self.text_widget.tag_configure('md_link', foreground='#64b5f6', underline=True) # 引用样式 self.text_widget.tag_configure('md_quote', foreground='#9e9e9e', lmargin1=20, lmargin2=20, font=('Microsoft YaHei UI', 11, 'italic')) def render(self, text: str, base_tag: str = 'assistant') -> None: """ 渲染 Markdown 文本 Args: text: Markdown 文本 base_tag: 基础样式标签 """ lines = text.split('\n') in_code_block = False code_block_content = [] for i, line in enumerate(lines): # 代码块处理 if line.strip().startswith('```'): if in_code_block: # 结束代码块 self._insert_code_block('\n'.join(code_block_content)) code_block_content = [] in_code_block = False else: # 开始代码块 in_code_block = True continue if in_code_block: code_block_content.append(line) continue # 普通行处理 self._render_line(line, base_tag) # 添加换行(除了最后一行) if i < len(lines) - 1: self.text_widget.insert(tk.END, '\n') def _render_line(self, line: str, base_tag: str) -> None: """渲染单行""" stripped = line.strip() # 空行 if not stripped: return # 标题 if stripped.startswith('### '): self.text_widget.insert(tk.END, stripped[4:], 'md_h3') return elif stripped.startswith('## '): self.text_widget.insert(tk.END, stripped[3:], 'md_h2') return elif stripped.startswith('# '): self.text_widget.insert(tk.END, stripped[2:], 'md_h1') return # 引用 if stripped.startswith('> '): self.text_widget.insert(tk.END, stripped[2:], 'md_quote') return # 无序列表 if stripped.startswith('- ') or stripped.startswith('* '): self.text_widget.insert(tk.END, ' • ', 'md_list_bullet') self._render_inline(stripped[2:], base_tag, 'md_list') return # 有序列表 list_match = re.match(r'^(\d+)\.\s+(.+)$', stripped) if list_match: num = list_match.group(1) content = list_match.group(2) self.text_widget.insert(tk.END, f' {num}. ', 'md_list_bullet') self._render_inline(content, base_tag, 'md_list') return # 普通段落 self._render_inline(line, base_tag) def _render_inline(self, text: str, base_tag: str, extra_tag: str = None) -> None: """渲染行内元素(粗体、斜体、代码、链接)""" tags = (base_tag, extra_tag) if extra_tag else (base_tag,) # 先处理 Markdown 链接 [text](url) last_end = 0 for match in self.MD_LINK_PATTERN.finditer(text): # 插入链接前的文本 if match.start() > last_end: self._render_inline_formatting(text[last_end:match.start()], tags) # 插入链接 link_text = match.group(1) link_url = match.group(2) self._insert_link(link_text, link_url) last_end = match.end() # 处理剩余文本 if last_end < len(text): remaining = text[last_end:] self._render_inline_formatting(remaining, tags) def _render_inline_formatting(self, text: str, tags: tuple) -> None: """处理行内格式(粗体、斜体、代码、纯URL)""" # 处理粗体 **text** parts = re.split(r'(\*\*[^*]+\*\*)', text) for part in parts: if part.startswith('**') and part.endswith('**'): self.text_widget.insert(tk.END, part[2:-2], tags + ('md_bold',)) else: # 处理斜体 *text* sub_parts = re.split(r'(\*[^*]+\*)', part) for sub_part in sub_parts: if sub_part.startswith('*') and sub_part.endswith('*') and len(sub_part) > 2: self.text_widget.insert(tk.END, sub_part[1:-1], tags + ('md_italic',)) else: # 处理行内代码 `code` code_parts = re.split(r'(`[^`]+`)', sub_part) for code_part in code_parts: if code_part.startswith('`') and code_part.endswith('`'): self.text_widget.insert(tk.END, code_part[1:-1], ('md_code',)) else: # 处理纯 URL self._render_urls(code_part, tags) def _render_urls(self, text: str, tags: tuple) -> None: """渲染纯 URL 链接""" last_end = 0 for match in self.URL_PATTERN.finditer(text): # 插入 URL 前的文本 if match.start() > last_end: self.text_widget.insert(tk.END, text[last_end:match.start()], tags) # 插入 URL 链接 url = match.group(0) # 清理 URL 末尾的标点 while url and url[-1] in '.,;:!?。,;:!?': url = url[:-1] self._insert_link(url, url) # 如果清理了标点,插入标点 original_url = match.group(0) if len(original_url) > len(url): self.text_widget.insert(tk.END, original_url[len(url):], tags) last_end = match.end() # 插入剩余文本 if last_end < len(text): self.text_widget.insert(tk.END, text[last_end:], tags) def _insert_link(self, text: str, url: str) -> None: """插入可点击的链接""" tag_name = f'link_{self._link_count}' self._link_count += 1 self.text_widget.tag_configure(tag_name, foreground='#64b5f6', underline=True) # 绑定点击事件 - 使用 ButtonRelease 而不是 Button-1,更可靠 def on_click(event, u=url): self._open_url(u) return "break" # 阻止事件继续传播 self.text_widget.tag_bind(tag_name, '', on_click) self.text_widget.tag_bind(tag_name, '', lambda e: self._set_cursor('hand2')) self.text_widget.tag_bind(tag_name, '', lambda e: self._set_cursor('')) self.text_widget.insert(tk.END, text, (tag_name, 'md_link')) def _set_cursor(self, cursor: str) -> None: """设置鼠标光标""" try: self.text_widget.config(cursor=cursor) except: pass def _insert_code_block(self, code: str) -> None: """插入代码块""" self.text_widget.insert(tk.END, '\n') self.text_widget.insert(tk.END, code, 'md_code_block') self.text_widget.insert(tk.END, '\n') def _open_url(self, url: str) -> None: """打开 URL""" try: webbrowser.open(url) except Exception as e: print(f"Failed to open URL: {url}, error: {e}") class LoadingIndicator: """加载动画指示器""" FRAMES = ["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"] def __init__(self, parent: tk.Widget, text: str = "处理中"): self.parent = parent self.text = text self.frame_index = 0 self.running = False self.after_id = None # 创建标签 self.label = tk.Label( parent, text="", font=('Microsoft YaHei UI', 10), fg='#ffd54f', bg='#1e1e1e' ) def start(self, text: str = None): """开始动画""" if text: self.text = text self.running = True self.label.pack(pady=5) self._animate() def stop(self): """停止动画""" self.running = False if self.after_id: self.parent.after_cancel(self.after_id) self.after_id = None self.label.pack_forget() def update_text(self, text: str): """更新提示文字""" self.text = text def _animate(self): """动画帧更新""" if not self.running: return frame = self.FRAMES[self.frame_index] self.label.config(text=f"{frame} {self.text}...") self.frame_index = (self.frame_index + 1) % len(self.FRAMES) self.after_id = self.parent.after(100, self._animate) class ChatView: """ 聊天视图 包含: - 消息显示区域(支持 Markdown 渲染) - 输入框 - 发送按钮 - 流式消息支持 """ def __init__( self, parent: tk.Widget, on_send: Callable[[str], None], on_show_history: Optional[Callable[[], None]] = None, on_show_settings: Optional[Callable[[], None]] = None ): """ 初始化聊天视图 Args: parent: 父容器 on_send: 发送消息回调函数 on_show_history: 显示历史记录回调函数 on_show_settings: 显示设置页面回调函数 """ self.parent = parent self.on_send = on_send self.on_show_history = on_show_history self.on_show_settings = on_show_settings # 流式消息状态 self._stream_active = False self._stream_tag = None self._stream_buffer = [] # 用于缓存流式内容,最后渲染 Markdown # 加载指示器 self.loading: Optional[LoadingIndicator] = None # Markdown 渲染器 self.md_renderer: Optional[MarkdownRenderer] = None self._create_widgets() def _create_widgets(self): """创建 UI 组件""" # 主框架 self.frame = tk.Frame(self.parent, bg='#1e1e1e') self.frame.pack(fill=tk.BOTH, expand=True, padx=10, pady=10) # 标题栏(包含标题和历史按钮) title_frame = tk.Frame(self.frame, bg='#1e1e1e') title_frame.pack(fill=tk.X, pady=(0, 10)) # 标题 title_label = tk.Label( title_frame, text="LocalAgent - 本地 AI 助手", font=('Microsoft YaHei UI', 16, 'bold'), fg='#61dafb', bg='#1e1e1e' ) title_label.pack(side=tk.LEFT, expand=True) # 按钮容器(右侧) btn_container = tk.Frame(title_frame, bg='#1e1e1e') btn_container.pack(side=tk.RIGHT) # 清空对话按钮 self.clear_btn = tk.Button( btn_container, text="🗑️ 清空", font=('Microsoft YaHei UI', 10), bg='#424242', fg='#ef9a9a', activebackground='#616161', activeforeground='#ef9a9a', relief=tk.FLAT, padx=10, pady=3, cursor='hand2', command=self._on_clear_chat ) self.clear_btn.pack(side=tk.RIGHT, padx=(5, 0)) # 设置按钮 if self.on_show_settings: self.settings_btn = tk.Button( btn_container, text="⚙️ 设置", font=('Microsoft YaHei UI', 10), bg='#424242', fg='#90caf9', activebackground='#616161', activeforeground='#90caf9', relief=tk.FLAT, padx=10, pady=3, cursor='hand2', command=self.on_show_settings ) self.settings_btn.pack(side=tk.RIGHT, padx=(5, 0)) # 历史记录按钮 if self.on_show_history: self.history_btn = tk.Button( btn_container, text="📜 历史", font=('Microsoft YaHei UI', 10), bg='#424242', fg='#ce93d8', activebackground='#616161', activeforeground='#ce93d8', relief=tk.FLAT, padx=10, pady=3, cursor='hand2', command=self.on_show_history ) self.history_btn.pack(side=tk.RIGHT) # 消息显示区域 self.message_area = scrolledtext.ScrolledText( self.frame, wrap=tk.WORD, font=('Microsoft YaHei UI', 11), bg='#2d2d2d', fg='#d4d4d4', insertbackground='white', relief=tk.FLAT, padx=10, pady=10, cursor='arrow' ) self.message_area.pack(fill=tk.BOTH, expand=True, pady=(0, 10)) # 禁止编辑但允许选择和点击链接 self.message_area.bind('', lambda e: 'break') # 禁止键盘输入 # 允许鼠标操作(选择文本、点击链接) # 配置消息标签样式 self.message_area.tag_configure('user', foreground='#4fc3f7', font=('Microsoft YaHei UI', 11, 'bold')) self.message_area.tag_configure('assistant', foreground='#81c784', font=('Microsoft YaHei UI', 11)) self.message_area.tag_configure('system', foreground='#ffb74d', font=('Microsoft YaHei UI', 10, 'italic')) self.message_area.tag_configure('error', foreground='#ef5350', font=('Microsoft YaHei UI', 10)) self.message_area.tag_configure('streaming', foreground='#81c784', font=('Microsoft YaHei UI', 11)) # 初始化 Markdown 渲染器 self.md_renderer = MarkdownRenderer(self.message_area) # 输入区域框架 input_frame = tk.Frame(self.frame, bg='#1e1e1e') input_frame.pack(fill=tk.X) # 输入框 self.input_entry = tk.Entry( input_frame, font=('Microsoft YaHei UI', 12), bg='#3c3c3c', fg='#ffffff', insertbackground='white', relief=tk.FLAT ) self.input_entry.pack(side=tk.LEFT, fill=tk.X, expand=True, ipady=8, padx=(0, 10)) self.input_entry.bind('', self._on_enter_pressed) # 发送按钮 self.send_button = tk.Button( input_frame, text="发送", font=('Microsoft YaHei UI', 11, 'bold'), bg='#0078d4', fg='white', activebackground='#106ebe', activeforeground='white', relief=tk.FLAT, padx=20, pady=5, cursor='hand2', command=self._on_send_clicked ) self.send_button.pack(side=tk.RIGHT) # 显示欢迎消息 welcome_msg = ( "欢迎使用 LocalAgent!\n" "- 输入问题进行对话\n" "- 输入文件处理需求(如\"复制文件\"、\"整理图片\")将触发执行模式" ) self.add_message(welcome_msg, 'system') # 创建加载指示器(放在消息区域下方) self.loading = LoadingIndicator(self.frame) def _on_enter_pressed(self, event): """回车键处理""" self._on_send_clicked() def _on_send_clicked(self): """发送按钮点击处理""" text = self.input_entry.get().strip() if text: self.input_entry.delete(0, tk.END) self.on_send(text) def _on_clear_chat(self): """清空对话""" from tkinter import messagebox if messagebox.askyesno("确认", "确定要清空当前对话吗?\n(这将同时清空对话上下文)"): self.clear_messages() # 通知 agent 清空上下文(通过回调) if hasattr(self, 'on_clear_context') and self.on_clear_context: self.on_clear_context() # 重新显示欢迎消息 welcome_msg = ( "欢迎使用 LocalAgent!\n" "- 输入问题进行对话\n" "- 输入文件处理需求(如\"复制文件\"、\"整理图片\")将触发执行模式" ) self.add_message(welcome_msg, 'system') def set_clear_context_callback(self, callback: Callable[[], None]): """设置清空上下文的回调""" self.on_clear_context = callback def add_message(self, message: str, tag: str = 'assistant', use_markdown: bool = True): """ 添加消息到显示区域 Args: message: 消息内容 tag: 消息类型 (user/assistant/system/error) use_markdown: 是否使用 Markdown 渲染(assistant 消息默认启用) """ # 添加前缀 prefix_map = { 'user': '\n[你] ', 'assistant': '\n[助手] ', 'system': '\n[系统] ', 'error': '\n[错误] ' } prefix = prefix_map.get(tag, '\n') self.message_area.insert(tk.END, prefix, tag) # 根据消息类型决定是否使用 Markdown 渲染 if use_markdown and tag == 'assistant' and self.md_renderer: self.md_renderer.render(message, tag) else: self.message_area.insert(tk.END, message, tag) self.message_area.insert(tk.END, '\n') self.message_area.see(tk.END) def start_stream_message(self, tag: str = 'assistant'): """ 开始流式消息 Args: tag: 消息类型 """ self._stream_active = True self._stream_tag = tag self._stream_buffer = [] # 添加前缀 prefix_map = { 'user': '\n[你] ', 'assistant': '\n[助手] ', 'system': '\n[系统] ', 'error': '\n[错误] ' } prefix = prefix_map.get(tag, '\n') self.message_area.insert(tk.END, prefix, tag) # 使用 mark 来标记内容开始位置,比索引更可靠 self.message_area.mark_set("stream_start", tk.END + "-1c") self.message_area.mark_gravity("stream_start", tk.LEFT) self.message_area.see(tk.END) def append_stream_chunk(self, chunk: str): """ 追加流式消息片段 Args: chunk: 消息片段 """ if not self._stream_active: return self._stream_buffer.append(chunk) self.message_area.insert(tk.END, chunk, self._stream_tag) self.message_area.see(tk.END) # 强制更新 UI self.message_area.update_idletasks() def end_stream_message(self): """结束流式消息,重新渲染为 Markdown""" if self._stream_active: # 获取完整的流式内容 full_content = ''.join(self._stream_buffer) # 如果是 assistant 消息且有内容,重新渲染为 Markdown if self._stream_tag == 'assistant' and self.md_renderer and full_content.strip(): # 删除原来的纯文本内容(从 mark 位置到末尾) try: self.message_area.delete("stream_start", tk.END) except tk.TclError: pass # 重新渲染为 Markdown self.md_renderer.render(full_content, self._stream_tag) self.message_area.insert(tk.END, '\n') self.message_area.see(tk.END) # 重置状态 self._stream_active = False self._stream_tag = None self._stream_buffer = [] def clear_messages(self): """清空消息区域""" self.message_area.delete(1.0, tk.END) def set_input_enabled(self, enabled: bool): """设置输入区域是否可用""" state = tk.NORMAL if enabled else tk.DISABLED self.input_entry.config(state=state) self.send_button.config(state=state) def show_loading(self, text: str = "处理中"): """显示加载动画""" if self.loading: self.loading.start(text) def hide_loading(self): """隐藏加载动画""" if self.loading: self.loading.stop() def update_loading_text(self, text: str): """更新加载提示文字""" if self.loading: self.loading.update_text(text) def get_frame(self) -> tk.Frame: """获取主框架""" return self.frame