负责定义基于LangChain的视觉AI代理使用RunnableWithMessageHistory import sys import os sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import time from datetime import datetime from typing import List, Dict, Any, Optional import base64 import io import logging # LangChain相关导入 from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_core.runnables.history import RunnableWithMessageHistory from langchain_community.chat_message_histories import ChatMessageHistory from langchain_core.output_parsers import StrOutputParser # 截图实现 import base64 import subprocess from datetime import datetime from PIL import Image import io import os def take_screenshot(): 获取屏幕截图返回base64数据 try: timestamp datetime.now().strftime(%Y%m%d_%H%M%S) filepath ftemp_screenshot_{timestamp}.png ps_script f Add-Type -AssemblyName System.Windows.Forms,System.Drawing $bounds [System.Windows.Forms.Screen]::PrimaryScreen.Bounds $bmp New-Object System.Drawing.Bitmap($bounds.width, $bounds.height) $graphics [System.Drawing.Graphics]::FromImage($bmp) $graphics.CopyFromScreen($bounds.X, $bounds.Y, 0, 0, $bounds.size) $bmp.Save({filepath}, [System.Drawing.Imaging.ImageFormat]::Png) $graphics.Dispose() $bmp.Dispose() subprocess.run([ powershell.exe, -NoProfile, -ExecutionPolicy, Bypass, -Command, ps_script ], capture_outputTrue) if os.path.exists(filepath): image Image.open(filepath) if image.width 2048 or image.height 2048: ratio min(2048/image.width, 2048/image.height) new_size (int(image.width*ratio), int(image.height*ratio)) image image.resize(new_size, Image.Resampling.LANCZOS) buffer io.BytesIO() image.save(buffer, formatPNG) base64_data base64.b64encode(buffer.getvalue()).decode() os.remove(filepath) return {success: True, data: base64_data} else: return {success: False, data: , error: 文件未生成} except Exception as e: return {success: False, data: , error: str(e)} import asyncio # TTS相关导入 import edge_tts # 配置日志 logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) logging.getLogger(mem0).setLevel(logging.CRITICAL) # 会话存储 store {} async def speak_text(text: str, voice: str zh-CN-XiaoxiaoNeural): 使用edge-tts合成并播放语音Windows原生API import tempfile import sys from ctypes import create_unicode_buffer, windll, wintypes if sys.platform ! win32: logger.warning(TTS仅支持Windows系统) return communicate edge_tts.Communicate(text, voice) # 创建临时MP3文件 with tempfile.NamedTemporaryFile(suffix.mp3, deleteFalse) as temp_file: temp_path temp_file.name try: # 保存音频到临时文件 await communicate.save(temp_path) # 使用Windows原生MCI API播放 _get_short_path_name_w windll.kernel32.GetShortPathNameW _get_short_path_name_w.argtypes [wintypes.LPCWSTR, wintypes.LPWSTR, wintypes.DWORD] _get_short_path_name_w.restype wintypes.DWORD def get_short_path_name(long_name: str) - str: output_buf_size 0 while True: output_buf create_unicode_buffer(output_buf_size) needed _get_short_path_name_w(long_name, output_buf, output_buf_size) if output_buf_size needed: return output_buf.value output_buf_size needed mci_send_string_w windll.winmm.mciSendStringW def mci_send(msg: str) - None: result mci_send_string_w(msg, 0, 0, 0) if result ! 0: logger.warning(fMCI错误 {result}: {msg}) # 获取短文件名并播放 mp3_shortname get_short_path_name(temp_path) mci_send(Close All) mci_send(fOpen {mp3_shortname} Type MPEGVideo Alias theMP3) mci_send(Play theMP3 Wait) mci_send(Close theMP3) except Exception as e: logger.error(fTTS播放失败: {e}) finally: # 立即清理临时文件 try: os.unlink(temp_path) except: pass def get_session_history(session_id: str): 获取会话历史 - 使用您提供的标准实现 if session_id not in store: store[session_id] ChatMessageHistory() return store[session_id] class LangChainVLMAgent: 基于LangChain的视觉语言模型代理 def __init__(self): # 使用OpenAI兼容协议连接Ollama self.llm ChatOpenAI( modelqwen3-vl:4b, base_urlhttp://localhost:11434/v1, api_keyollama, temperature0.7, timeout300, ) # 定义系统提示词 self.system_prompt 温顺的18岁AI少女,对话内容控制在30个字 # 创建提示模板 self.prompt ChatPromptTemplate.from_messages([ (system, self.system_prompt), MessagesPlaceholder(variable_namehistory), (human, {input}) ]) # 创建带历史记录的可运行对象 self.conversation RunnableWithMessageHistory( self.prompt | self.llm, get_session_history, input_messages_keyinput, history_messages_keyhistory ) async def process_message(self, user_input: str, session_id: str default) - str: 处理用户消息并返回AI回复 Args: user_input (str): 用户输入 session_id (str): 会话ID Returns: str: AI回复 try: # 获取截图 result take_screenshot() if result[success]: screenshot_base64 result[data] # 构造包含图像的消息 message_content [ {type: text, text: f用户{user_input}}, { type: image_url, image_url: { url: fdata:image/png;base64,{screenshot_base64} } } ] else: print(f截图失败: {result.get(error, 未知错误)}) message_content f用户{user_input} # 调用模型获取回复 response self.conversation.invoke( {input: message_content}, config{configurable: {session_id: session_id}} ) ai_response response.content if hasattr(response, content) else str(response) return ai_response except Exception as e: logger.error(f处理消息时发生错误: {e}) import traceback logger.error(f详细错误信息: {traceback.format_exc()}) return 抱歉我遇到了一些问题请稍后再试。 def save_to_memory_log(content: str): 将内容保存到memory文件夹下的日志文件中 current_date datetime.now().strftime(%Y-%m-%d) memory_dir os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), memory_data) if not os.path.exists(memory_dir): os.makedirs(memory_dir) log_file_path os.path.join(memory_dir, f{current_date}.txt) timestamp datetime.now().strftime(%Y-%m-%d %H:%M) try: with open(log_file_path, a, encodingutf-8) as f: f.write(f[{timestamp}] {content}\n) logger.info(f已保存到日志文件: {log_file_path}) except Exception as e: logger.error(f保存日志失败: {e}) # 创建全局代理实例 vlm_agent LangChainVLMAgent() async def chat_with_agent(user_input: str, session_id: str default) - str: 与视觉AI代理进行单次对话并播放TTS回复 ai_response await vlm_agent.process_message(user_input, session_id) save_to_memory_log(ai_response) print(fAI回复: {ai_response}) # 使用TTS播放回复 await speak_text(ai_response) return ai_response if __name__ __main__: import asyncio asyncio.run(chat_with_agent(屏幕有什么内容))