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 = '
' # 添加容器 for news in news_list[:5]: card_html = f"""
新闻图片
{news['title']}
{news['source']}
""" container_html += card_html container_html += '
' # 关闭容器 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( "

惊堂木-虚假信息检测系统

") 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()