se_app.py

import gradio as gr
from huggingface_hub import InferenceClient
import os
import numpy as np
from scipy.io.wavfile import write as write_wav
from PIL import Image
from tools import audio_to_str, image_to_str  # 导入tools.py中的方法

client = InferenceClient("HuggingFaceH4/zephyr-7b-beta")

# 指定保存文件的相对路径
SAVE_DIR = 'download'  # 相对路径
os.makedirs(SAVE_DIR, exist_ok=True)  # 确保目录存在

def get_client_ip(request: gr.Request, debug_mode=False):
    """获取客户端真实IP地址"""
    if request:
        # 从请求头中获取真实IP（考虑代理情况）
        x_forwarded_for = request.headers.get("x-forwarded-for", "")
        if x_forwarded_for:
            client_ip = x_forwarded_for.split(",")[0]
        else:
            client_ip = request.client.host
        if debug_mode:
            print(f"Debug: Client IP detected as {client_ip}")
        return client_ip
    return "unknown"

def save_audio(audio, filename):
    """保存音频为.wav文件"""
    sample_rate, audio_data = audio
    write_wav(filename, sample_rate, audio_data)

def save_image(image, filename):
    """保存图片为.jpg文件"""
    img = Image.fromarray(image.astype('uint8'))
    img.save(filename)

def process(audio, image, text, request: gr.Request):
    """处理语音、图片和文本的示例函数"""
    client_ip = get_client_ip(request, True)
    print(f"Processing request from IP: {client_ip}")

    audio_info = "未收到音频"
    image_info = "未收到图片"
    text_info = "未收到文本"
    audio_filename = None
    image_filename = None
    audio_text = ""
    image_text = ""

    if audio is not None:
        sample_rate, audio_data = audio
        audio_info = f"音频采样率: {sample_rate}Hz, 数据长度: {len(audio_data)}"
        # 保存音频为.wav文件
        audio_filename = os.path.join(SAVE_DIR, f"audio_{client_ip}.wav")
        save_audio(audio, audio_filename)
        print(f"Audio saved as {audio_filename}")
        # 调用tools.py中的audio_to_str方法处理音频
        audio_text = audio_to_str("33c1b63d", "40bf7cd82e31ace30a9cfb76309a43a3", "OTY1YzIyZWM3YTg0OWZiMGE2ZjA2ZmE4", audio_filename)
        if audio_text:
            print(f"Audio text: {audio_text}")
        else:
            print("Audio processing failed")

    if image is not None:
        image_info = f"图片尺寸: {image.shape}"
        # 保存图片为.jpg文件
        image_filename = os.path.join(SAVE_DIR, f"image_{client_ip}.jpg")
        save_image(image, image_filename)
        print(f"Image saved as {image_filename}")
        # 调用tools.py中的image_to_str方法处理图片
        image_text = image_to_str(endpoint="https://ai-siyuwang5414995ai361208251338.cognitiveservices.azure.com/", key="45PYY2Av9CdMCveAjVG43MGKrnHzSxdiFTK9mWBgrOsMAHavxKj0JQQJ99BDACHYHv6XJ3w3AAAAACOGeVpQ", unused_param=None, file_path=image_filename)
        if image_text:
            print(f"Image text: {image_text}")
        else:
            print("Image processing failed")

    if text:
        text_info = f"接收到文本: {text}"

    return audio_info, image_info, text_info, audio_text, image_text

# 创建自定义的聊天界面
with gr.Blocks() as app:
    gr.Markdown("# ToDoAgent Multi-Modal Interface")

    # 创建两个标签页
    with gr.Tab("Chat"):
        # 修复Chatbot类型警告
        chatbot = gr.Chatbot(height=500, type="messages")

        msg = gr.Textbox(label="输入消息", placeholder="输入您的问题...")

        # 上传区域
        with gr.Row():
            audio_input = gr.Audio(label="上传语音", type="numpy", sources=["upload", "microphone"])
            image_input = gr.Image(label="上传图片", type="numpy")

        # 设置区域
        with gr.Accordion("高级设置", open=False):
            system_msg = gr.Textbox(value="You are a friendly Chatbot.", label="系统提示")
            max_tokens = gr.Slider(minimum=1, maximum=2048, value=512, step=1, label="最大生成长度")
            temperature = gr.Slider(minimum=0.1, maximum=4.0, value=0.7, step=0.1, label="温度")
            top_p = gr.Slider(minimum=0.1, maximum=1.0, value=0.95, step=0.05, label="Top-p")

        # 提交按钮
        submit_btn = gr.Button("发送", variant="primary")

        # 清除按钮
        clear = gr.Button("清除聊天")

        # 事件处理
        def user(user_message, chat_history):
            return "", chat_history + [{"role": "user", "content": user_message}]
#新增多模态处理--1
        def respond(message, chat_history, system_message, max_tokens, temperature, top_p, audio=None, image=None, text=None, request=None):
            """生成响应的函数"""
            # 处理多模态输入
            multimodal_content = ""
            if audio is not None:
                try:
                    audio_filename = os.path.join(SAVE_DIR, "temp_audio.wav")
                    save_audio(audio, audio_filename)
                    audio_text = audio_to_str("33c1b63d", "40bf7cd82e31ace30a9cfb76309a43a3", "OTY1YzIyZWM3YTg0OWZiMGE2ZjA2ZmE4", audio_filename)
                    if audio_text:
                        multimodal_content += f"音频内容: {audio_text}\n"
                except Exception as e:
                    print(f"Audio processing error: {e}")
            
            if image is not None:
                try:
                    image_filename = os.path.join(SAVE_DIR, "temp_image.jpg")
                    save_image(image, image_filename)
                    image_text = image_to_str(endpoint="https://ai-siyuwang5414995ai361208251338.cognitiveservices.azure.com/", key="45PYY2Av9CdMCveAjVG43MGKrnHzSxdiFTK9mWBgrOsMAHavxKj0JQQJ99BDACHYHv6XJ3w3AAAAACOGeVpQ", unused_param=None, file_path=image_filename)
                    if image_text:
                        multimodal_content += f"图片内容: {image_text}\n"
                except Exception as e:
                    print(f"Image processing error: {e}")
            
            # 组合最终消息
            final_message = message
            if multimodal_content:
                final_message = f"{message}\n\n{multimodal_content}"
            
            # 构建消息历史
            messages = [{"role": "system", "content": system_message}]
            for chat in chat_history:
                if isinstance(chat, dict) and "role" in chat and "content" in chat:
                    messages.append(chat)
            
            messages.append({"role": "user", "content": final_message})
            
            # 调用HuggingFace API
            try:
                response = client.chat_completion(
                    messages,
                    max_tokens=max_tokens,
                    stream=True,
                    temperature=temperature,
                    top_p=top_p,
                )
                
                partial_message = ""
                for token in response:
                    if token.choices[0].delta.content is not None:
                        partial_message += token.choices[0].delta.content
                        yield partial_message
            except Exception as e:
                yield f"抱歉，生成响应时出现错误: {str(e)}"

        def bot(chat_history, system_message, max_tokens, temperature, top_p, audio, image, text):
            # 检查chat_history是否为空
            if not chat_history or len(chat_history) == 0:
                return
            
            # 获取最后一条用户消息
            last_message = chat_history[-1]
            if not last_message or not isinstance(last_message, dict) or "content" not in last_message:
                return
            
            user_message = last_message["content"]

            # 生成响应
            bot_response = ""
            for response in respond(
                    user_message,
                    chat_history[:-1],
                    system_message,
                    max_tokens,
                    temperature,
                    top_p,
                    audio,
                    image,
                    text
            ):
                bot_response = response
                # 添加助手回复到聊天历史
                updated_history = chat_history + [{"role": "assistant", "content": bot_response}]
                yield updated_history

        msg.submit(user, [msg, chatbot], [msg, chatbot], queue=False).then(
            bot, [chatbot, system_msg, max_tokens, temperature, top_p, audio_input, image_input, msg], chatbot
        )

        submit_btn.click(user, [msg, chatbot], [msg, chatbot], queue=False).then(
            bot, [chatbot, system_msg, max_tokens, temperature, top_p, audio_input, image_input, msg], chatbot
        )

        clear.click(lambda: None, None, chatbot, queue=False)

    with gr.Tab("Audio/Image Processing"):
        gr.Markdown("## 处理音频和图片")
        audio_processor = gr.Audio(label="上传音频", type="numpy")
        image_processor = gr.Image(label="上传图片", type="numpy")
        text_input = gr.Textbox(label="输入文本")
        process_btn = gr.Button("处理", variant="primary")
        audio_output = gr.Textbox(label="音频信息")
        image_output = gr.Textbox(label="图片信息")
        text_output = gr.Textbox(label="文本信息")
        audio_text_output = gr.Textbox(label="音频转文字结果")
        image_text_output = gr.Textbox(label="图片转文字结果")

        # 修改后的处理函数调用
        process_btn.click(
            process,
            inputs=[audio_processor, image_processor, text_input],
            outputs=[audio_output, image_output, text_output, audio_text_output, image_text_output]
        )

if __name__ == "__main__":
    app.launch()