app.py

import base64
import io
import os
import sys
import traceback
from typing import List
import gradio as gr
from PIL import Image
import time
from langchain.agents import AgentExecutor, create_react_agent
from langchain_core.prompts import ChatPromptTemplate
from model import *
from gradio import ChatMessage, on,HTML
from tools import *
from prompt import *
import sys
from langchain_openai import ChatOpenAI
from gradio.components.chatbot import MessageDict
from langchain_community.llms import Tongyi
from io import BytesIO
from css import *

os.environ["OPENAI_API_KEY"] = "sb-6a683cb3bd63a9b72040aa2dd08feff8b68f08a0e1d959f5"
os.environ['OPENAI_BASE_URL'] = "https://api.openai-sb.com/v1/"
os.environ["SERPAPI_API_KEY"] = "dcc98b22d5f7d413979a175ff7d75b721c5992a3ee1e2363020b2bbdf4f82404"
os.environ['TAVILY_API_KEY'] = "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"  # you
os.environ["REPLICATE_API_TOKEN"] = "r8_IYJpjwjrxegcUfBeBbyUxErJXXsnHDM4AlSQQ"
os.environ["DASHSCOPE_API_KEY"] = "sk-8159f0ed38994c3b96b4527404ea1cda"

class ImageProcessor:
    """Handles image processing and analysis"""

    @staticmethod
    def resize_image(image: Image.Image) -> Image.Image:
        """Resize image to appropriate dimensions"""
        width, height = image.size
        while width >= 500 or height >= 400:
            width = width * 0.8
            height = height * 0.8
        return image.resize((int(width), int(height)))

    @staticmethod
    def encode_image(image: Image.Image) -> str:
        """Encode image to base64 string"""
        buffered = io.BytesIO()
        image.save(buffered, format="PNG")
        return base64.b64encode(buffered.getvalue()).decode("utf-8")

    @staticmethod
    def load_image(image_path_or_url: str) -> Optional[Image.Image]:
        """Load image from path or URL"""
        if image_path_or_url.startswith(('http://', 'https://')):
            try:
                response = requests.get(image_path_or_url, timeout=10)
                response.raise_for_status()
                return Image.open(BytesIO(response.content))
            except Exception as e:
                print(f"Failed to download image: {image_path_or_url} | Error: {e}")
                return None
        else:
            try:
                return Image.open(image_path_or_url)
            except Exception as e:
                print(f"Failed to open local image: {image_path_or_url} | Error: {e}")
                return None

    @staticmethod
    def analyze_image(image: Image.Image, prompt: str) -> str:
        base64_image = ImageProcessor.encode_image(image)
        client = OpenAI(
            api_key=os.getenv('DASHSCOPE_API_KEY'),
            base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
        )
        completion = client.chat.completions.create(
            model="qwen-vl-plus",
            messages=[
                {
                    "role": "system",
                    "content": [{"type": "text", "text": f"{prompt}"}]},
                {
                    "role": "user",
                    "content": [
                        {
                            "type": "image_url",
                            "image_url": {"url": f"data:image/png;base64,{base64_image}"},
                        },
                        {"type": "text", "text": "请分析该图片"},
                    ],
                }
            ],
        )
        return completion.choices[0].message.content


class FactChecker:
    """Main fact-checking agent class"""

    def __init__(self):
        self.tools = None
        self.selected_language = "ch"
        self.image_summaries = ""
        self.query = ""
        self.llm = ChatOpenAI(model_name="ep-20250416211133-w5rft",openai_api_key="272b1003-3823-4723-834d-c004e9072e2f",openai_api_base="https://ark.cn-beijing.volces.com/api/v3")
        self.selected_contrast = "qwen"
        self.selected_model = "自研"
        self.contrast_model_prompt = ""
        self.fact_checking = False
        dashscope.api_key = "sk-8159f0ed38994c3b96b4527404ea1cda"

    def set_language(self, language: str):
        """Set the language for responses"""
        self.selected_language = "ch" if language == "中文" else "en"

    def get_prompt(self) -> ChatPromptTemplate:
        if self.selected_language == "ch":
            return ChatPromptTemplate.from_template(ch_prompt)
        else:
            return ChatPromptTemplate.from_template(en_prompt)

    def get_image_prompt(self) -> str:
        if self.selected_language == "ch":
            return image_ch_prompt
        else:
            return image_en_prompt

    def process_image(self, image_path_or_url: str) -> Optional[str]:
        """Process and analyze an image"""
        if not image_path_or_url:
            return None

        image = ImageProcessor.load_image(image_path_or_url)
        if not image:
            return None

        image = ImageProcessor.resize_image(image)
        prompt = self.get_image_prompt()
        analysis = ImageProcessor.analyze_image(image, prompt)
        return analysis.replace('*', '').replace('\n', ' ').strip()

    def process_image_load(self, image: Image.Image) -> Optional[str]:
        if not image:
            return None
        image = ImageProcessor.resize_image(image)
        prompt = self.get_image_prompt()
        analysis = ImageProcessor.analyze_image(image, prompt)
        return analysis.replace('*', '').replace('\n', ' ').strip()

    def extract_agent_thoughts(self, intermediate_steps: List) -> List[ChatMessage]:
        """Extract agent thoughts from intermediate steps"""
        agent_thought = []
        pattern1 = r"Pre Thought:(.*?)Thought: (.*?)\nAction: (.*?)\nAction Input: (.*)"
        pattern2 = r"Thought: (.*?)\nAction: (.*?)\nAction Input: (.*)"

        for index in intermediate_steps:
            match = re.search(pattern1, index[0].log, re.S)
            before_thought = match.group(1).strip() if match else ""
            thought = match.group(2).strip() if match else ""
            action = match.group(3).strip() if match else ""
            action_input = match.group(4).strip() if match else ""

            if not match:
                match = re.search(pattern2, index[0].log, re.S)
                before_thought = match.group(1).strip() if match else ""
                thought = match.group(2).strip() if match else ""
                action_input = match.group(3).strip() if match else ""

            if len(before_thought) >= 3:
                # agent_thought.append({"title": "预思考：", "content": before_thought})
                agent_thought.append(
                    ChatMessage(
                        role="assistant",
                        content=before_thought,
                        metadata={"title": "预思考："}
                    )
                )
            if len(action_input) >= 3:
                # agent_thought.append({"title": f"工具调用:{action}", "content": action_input})
                web_search_tools = {"BoCha Webs Search","Image Search","Baidu News Search","tavily_search_results_json"}
                if action in web_search_tools or action == "":
                    action = "网页搜索"
                agent_thought.append(
                    ChatMessage(
                        role="assistant",
                        content=action_input,
                        metadata={"title": f"工具调用:{action}"}
                    )
                )

        return agent_thought

    def process_news_results(self, intermediate_steps: List) -> List[Dict]:
        news_results = []
        if not intermediate_steps:
            return news_results

        first_step = intermediate_steps[0][0]

        if first_step.tool == "tavily_search_results_json":
            for step1 in intermediate_steps[:1]:  # Only first step
                try:
                    for step2 in step1[1][:3]:  # First 3 results
                        news_results.append({
                            "title": step2['content'][:28] + "...",
                            "url": step2['url'],
                            "source": "",
                            "image": "",
                            "time": ""
                        })
                except:
                    continue

        elif first_step.tool == "BoCha Webs Search":
            for step1 in intermediate_steps:
                try:
                    for i in range(3):
                        news_results.append({
                            "title": step1[1]['data']['webPages']['value'][i]['snippet'][:28] + "...",
                            "url": step1[1]['data']['webPages']['value'][i]['url'],
                            "source": step1[1]['data']['webPages']['value'][i]['siteName'],
                            "image": step1[1]['data']['images']['value'][i]['contentUrl'],
                            "time": step1[1]['data']['webPages']['value'][i]['dateLastCrawled']
                        })
                except:
                    continue

        elif first_step.tool == "Baidu News Search":
            for step1 in intermediate_steps:
                try:
                    for i in range(3):
                        news_results.append({
                            "title": step1[1][i]['title'][:28] + "...",
                            "url": step1[1][i]['link'],
                            "source": step1[1][i]['source'],
                            "image": "",
                            "time": ""
                        })
                except:
                    continue

        return news_results

    def set_tools(self):
        self.tools = [
            BoChaSearchTool(selected_language=self.selected_language,fact_checking=self.fact_checking),
            ChineseTavilySearchResults(selected_language=self.selected_language,fact_checking=self.fact_checking),
            ImageSearchTool(selected_language=self.selected_language),
            WeatherCrossing(selected_language=self.selected_language),
            GetHoliday(selected_language=self.selected_language),
            GetLocation(selected_language=self.selected_language),
            CurrencyConversion(selected_language=self.selected_language),
            SafeCodeExecutor(selected_language=self.selected_language),
            SafeExpressionEvaluator(selected_language=self.selected_language),
            RegionInquiryTool(selected_language=self.selected_language),
            HTMLTextExtractor(selected_language=self.selected_language)
        ]

    @staticmethod
    def format_reply(reply:Dict) -> HTML:
        formatted_text = ""
        news_list = reply.get("news", [])
        container_html = '<div class="news-container">'  # 添加容器

        for news in news_list[:5]:
            card_html = f"""
                <div class="news-card">
                    <img src="{news['image']}" alt="新闻图片">
                    <div class="title"><a href="{news['url']}" target="_blank">{news['title']}</a></div>
                    <div class="source">{news['source']}</div>
                </div>
                """
            container_html += card_html
        container_html += '</div>'  # 关闭容器
        formatted_text += container_html
        formatted_text += reply.get("text", "")
        return gr.HTML(formatted_text)

    def contract(self)-> str | None:
        tool = BoChaSearchTool(self.selected_language)
        web_information = tool._run(self.query)
        information = []
        for step in web_information['data']['webPages']['value']:
            information.append(step['snippet'])
        contract_query = self.query + "搜索到的相关新闻：" + str(information)
        if self.selected_contrast == "qwen":
            result = qwen(contract_query)
        elif self.selected_contrast == "llama":
            result = llama(contract_query)
        elif self.selected_contrast == "glm":
            result = glm(contract_query)
        elif self.selected_contrast == "doubao":
            result = doubao(contract_query)
        elif self.selected_contrast == "deepseek":
            result = deepseek(contract_query)
        else:
            result = baichuan(contract_query)
        return result

    def SelectLanguage(self,option:str):
        if option == "英文":
            self.selected_language = "en"
        else:
            self.selected_language = "ch"

    def SelectModel(self,option:str):
        if option == "自研":
            self.selected_model = "gpt"
        elif option == "llama3-70b":
            self.selected_model = "llama3-70b"
        elif option == "llama3-8b":
            self.selected_model = "llama3-8b"
        else:
            self.selected_model = "mistral"

    def SelectMode(self,option:str):
        if option == "谣言检测":
            self.fact_checking = False
        else:
            self.fact_checking = True

    def SelectContrast(self,option:str):
        if option == "qwen":
            self.selected_contrast = "qwen"
        elif option == "llama":
            self.selected_contrast = "llama"
        elif option == "glm":
            self.selected_contrast = "glm"
        elif option == "doubao":
            self.selected_contrast = "doubao"
        elif option == "deepseek":
            self.selected_contrast = "deepseek"
        else:
            self.selected_contrast = "baichuan"

    def update_image_summaries(self):
        return self.image_summaries

    def ModelPrompt(self,prompt):
        self.contrast_model_prompt = prompt

    def check_facts_chat(self,dict, space):
        """Main fact-checking method"""
        topic = dict['text']
        image = dict['files']

        if not image:
            self.image_summaries = ""
        else:
            image_path = image[0]
            image = Image.open(image_path)
            self.image_summaries = self.process_image_load(image)

        if not topic:
            return "无输入内容"

        self.query = topic

        # Create agent
        prompt = self.get_prompt()

        self.set_tools()
        agent = create_react_agent(self.llm, self.tools, prompt)
        # Redirect stdout to capture agent thoughts
        captured_output = io.StringIO()
        sys.stdout = captured_output

        cur_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
        # Execute agent
        agent_executor = AgentExecutor(
            agent=agent,
            tools=self.tools,
            verbose=True,
            handle_parsing_errors=True,
            return_intermediate_steps=True
        )
        try:
            response = agent_executor.invoke({
                "current_time": cur_time,
                "input": topic,
                "image_information": self.image_summaries
            })

            sys.stdout = sys.__stdout__
            captured_content = str(captured_output.getvalue())
            captured_content = re.sub(r'\x1b\[[0-9;]*m', '', captured_content)

            # Extract explanation from captured output
            explain = ""
            try:
                match = re.search(r'\{.*\}(.*)', captured_content, re.DOTALL)
                if match:
                    extracted_content = match.group(1).strip()
                    match2 = re.search(r'Summary:(.*)Final Answer:', extracted_content, re.DOTALL)
                    if match2:
                        explain = match2.group(1).strip()
            except Exception as e:
                print(f"Error extracting explanation: {e}")

            # Format response
            text = re.sub(r'<.*?>','',response['output'].replace('*','').strip())
            reply_explain = {
                'text': text,
                'news': self.process_news_results(response['intermediate_steps'])
            }
            # print(response['intermediate_steps'])
            formatted_reply = FactChecker.format_reply(reply_explain)

            return self.extract_agent_thoughts(response['intermediate_steps']) + [formatted_reply]
        except ValueError as e:
            if "DataInspectionFailed" in str(e):
                return "您的输入可能包含敏感内容，请重新描述。"
            else:
                return "出现错误，请稍后再试。"

class History:
    @staticmethod
    def generate_chat_title(conversation: list[MessageDict]) -> str:
        title = ""
        for message in conversation:
            if message["role"] == "user":
                if isinstance(message["content"], str):
                    title += message["content"]
                    break
                else:
                    title += "📎 "
        if len(title) > 40:
            title = title[:40] + "..."

        # print(title)
        return title or "Conversation"

    @staticmethod
    def load_chat_history(conversations):
        return gr.Dataset(
            samples=[
                [History.generate_chat_title(conv)]
                for conv in conversations or []
                if conv
            ]
        )
    @staticmethod
    def save_conversation(
            index: int | None,
            conversation: list[MessageDict],
            saved_conversations: list[list[MessageDict]],
    ):
        if index is not None:
            saved_conversations[index] = conversation
        else:
            saved_conversations.append(conversation)
            index = len(saved_conversations) - 1
        return index, saved_conversations

    @staticmethod
    def load_conversation(
            index: int,
            conversations: list[list[MessageDict]],
    ):
        return (
            index,
            gr.Chatbot(
                value=conversations[index],  # type: ignore
                feedback_value=[],
                type="messages"
            ),
        )


def build_interface() -> gr.Blocks:
    with gr.Blocks(css=css, theme='soft') as demo:
        gr.HTML(
            "<h1 style='font-size: 36px; text-align: center; color: #333333; margin-bottom: 20px;'>惊堂木-虚假信息检测系统</h1>")
        with gr.Tab(label='Chat'):
            with gr.Row():
                with gr.Sidebar():
                    with gr.Column(scale=1):
                        gr.Textbox(visible=False)
                    with gr.Column(scale=1):
                        gr.Textbox(visible=False)
                    with gr.Column(scale=1):
                        new_chat_button = gr.Button(
                            "New chat",
                            variant="primary",
                            size="md",
                            icon="plus.svg",
                        )
                        chat_history_dataset = gr.Dataset(
                            components=[gr.Textbox(visible=False)],
                            show_label=False,
                            layout="table",
                            type="index",
                        )

                        language_select = gr.Dropdown(["中文", "英文"], label="请选择要使用的语言", scale=1, value="中文")
                        # model_select = gr.Dropdown(["自研", "llama3-70b", "llama3-8b", "mistral"],
                        #                            label="请选择要使用的大模型",
                        #                            scale=1, value="自研")
                        # mode_select = gr.Dropdown(["事实核查","谣言检测"],label= "请选择使用的模式",scale= 1, value= "谣言检测")

                with gr.Column(scale=3):
                    bot = gr.ChatInterface(
                        fn=factChecker.check_facts_chat,
                        examples=[
                            {
                                "text": "9月19日，马来西亚最高元首 Ibrahim 应邀对中国进行为期8天国事访问，亦是2024年1月上任以来首次访问东盟外国家。"},
                            {
                                "text": "据最新天文研究，火星的轨道将逐渐接近地球，最终成为地球的“第二月亮”。天文学家预测这一变化将在2025年发生，届时火星将在夜空中与月亮一样明亮，影响全球潮汐和生态平衡。"}
                        ],
                        chatbot=gr.Chatbot(label='惊堂木',
                                           avatar_images=("./image/user.png", "./image/logo.png"),
                                           type="messages",
                                           height=600,
                                           layout="bubble",
                                           show_copy_button=True,
                                           show_copy_all_button=True,
                                           ),
                        multimodal=True,
                        show_progress='full',
                        type="messages",
                        flagging_mode='manual',
                        cache_examples=False,
                        example_icons=["./image/search.png", "./image/search.png"]
                    )

                with gr.Column(scale=1):
                    with gr.Accordion("图片分析", open=False):
                        img_info = gr.Textbox(label="分析结果", lines=5)
                    contrast_model = gr.Textbox(label="对比模型", lines=5)
                    #with gr.Accordion("展示对比模型prompt", open=False):
                        #model_prompt = gr.Textbox(label="对比模型prompt", lines=5, placeholder=constract_model_prompt, interactive=True)
                    contrast_select = gr.Dropdown(["qwen", "llama", "glm", "doubao", "deepseek", "baichuan"],
                                                   label="请选择使用的对比模型", scale=1, value="qwen")
                    contrast_bn = gr.Button("展示对比模型")
                    #prompt_bn = gr.Button("确定更改提示词")
                    img_bn = gr.Button("显示分析结果")
            img_bn.click(factChecker.update_image_summaries, [], img_info)
            language_select.change(factChecker.SelectLanguage, language_select, [])
            # model_select.change(factChecker.SelectModel, model_select, [])
            # mode_select.change(factChecker.SelectMode, mode_select, [])
            contrast_select.change(factChecker.SelectContrast, contrast_select, [])
            contrast_bn.click(factChecker.contract, [], contrast_model)
            #prompt_bn.click(factChecker.ModelPrompt,model_prompt, [])
            new_chat_button.click(
                lambda x: x,
                [bot.chatbot],
                [bot.chatbot_state],
                show_api=False,
                queue=False,
            ).then(
                History.save_conversation,
                [bot.conversation_id, bot.chatbot_state, bot.saved_conversations],
                [bot.conversation_id, bot.saved_conversations]
            ).then(
                lambda: (None, []),
                None,
                [bot.conversation_id, bot.chatbot],
                show_api=False,
                queue=False,
            ).then(
                lambda x: x,
                [bot.chatbot],
                [bot.chatbot_state],
                show_api=False,
                queue=False,
            )
            on(
                triggers=[demo.load, bot.saved_conversations.change],
                fn=History.load_chat_history,
                inputs=bot.saved_conversations,
                outputs=chat_history_dataset,
                show_api=False,
                queue=False,
            )
            chat_history_dataset.click(
                lambda: [],
                None,
                [bot.chatbot],
                show_api=False,
                queue=False,
                show_progress="hidden",
            ).then(
                History.load_conversation,
                [chat_history_dataset, bot.saved_conversations],
                [bot.conversation_id, bot.chatbot],
                show_api=False,
                queue=False,
                show_progress="hidden",
            )

        # with gr.Tab(label='Para', scale=1):
        #     gr.Textbox()
        return demo

if __name__ == "__main__":
    factChecker = FactChecker()
    demo = build_interface()
    demo.launch()