Spaces:
Sleeping
Sleeping
| import gradio as gr | |
| import os | |
| import json | |
| from typing import TypedDict, List, Dict, Any, Optional | |
| from langgraph.graph import StateGraph, START, END | |
| # from langchain_openai import ChatOpenAI | |
| from langchain_core.messages import HumanMessage, AIMessage | |
| # from llama_index.llms.huggingface_api import HuggingFaceInferenceAPI | |
| from huggingface_hub import InferenceClient | |
| class EmailState(TypedDict): | |
| # 正在处理的电子邮件 | |
| email: Dict[str, Any] # 包含主题、发件人、正文等。 | |
| # 分析与决策 | |
| is_spam: Optional[bool] | |
| spam_reason: Optional[str] | |
| email_category: Optional[str] | |
| # 响应生成 | |
| draft_response: Optional[str] | |
| # 处理元数据 | |
| messages: List[Dict[str, Any]] # 跟踪与 LLM 的对话以进行分析 | |
| # Initialize our LLM | |
| # model = ChatOpenAI(temperature=0) | |
| # model = HuggingFaceInferenceAPI(model_name="Qwen/Qwen2.5-Coder-32B-Instruct") | |
| hf_token=os.getenv("hf_token") | |
| model = InferenceClient( | |
| model="Qwen/Qwen2.5-Coder-32B-Instruct", | |
| token=hf_token, | |
| timeout=30 | |
| ) | |
| def query_llm(prompt: str): | |
| response = model.chat.completions.create(messages=[ | |
| { | |
| "role": "user", | |
| "content": prompt | |
| }],temperature=0) | |
| return response.choices[0].message | |
| def read_email(state: EmailState): | |
| """Alfred reads and logs the incoming email""" | |
| email = state["email"] | |
| # 在这里我们可能会做一些初步的预处理 | |
| print(f"Alfred is processing an email from {email['sender']} with subject: {email['subject']}") | |
| # 这里不需要更改状态 | |
| return {} | |
| def classify_email(state: EmailState): | |
| """Alfred uses an LLM to determine if the email is spam or legitimate""" | |
| email = state["email"] | |
| # 为 LLM 准备提示 | |
| prompt = f""" | |
| As Alfred the butler, analyze this email and determine if it is spam or legitimate. | |
| Email: | |
| From: {email['sender']} | |
| Subject: {email['subject']} | |
| Body: {email['body']} | |
| First, determine if this email is spam. If it is spam, explain why. | |
| If it is legitimate, categorize it (inquiry, complaint, thank you, etc.). | |
| Please answer strictly in the following format: | |
| Email category: inquiry, complaint, thank you, spam, etc. | |
| Is spam: | |
| Reason for spam: | |
| """ | |
| # Call the LLM | |
| # messages = [HumanMessage(content=prompt)] | |
| response = query_llm(prompt) | |
| # 解析响应的简单逻辑（在实际应用中，您需要更强大的解析） | |
| response_text = response.content.lower() | |
| # print(f"test response_text:{response_text}") | |
| is_spam = "is spam: yes" in response_text | |
| # print(f"test is_spam:{is_spam and "reasoning:" in response_text}") | |
| # print(f"test reasoning:{response_text.split("reasoning:")[1].strip()}") | |
| # 如果是垃圾邮件，请提取原因,输出的格式可能不同，需要根据提示词固定格式。 | |
| spam_reason = None | |
| spam_reason_keyword="reason for spam:" | |
| if is_spam and spam_reason_keyword in response_text: | |
| spam_reason = response_text.split(spam_reason_keyword)[1].strip() | |
| # print(f"test spam_reason:{spam_reason}") | |
| # 确定类别是否合法 | |
| email_category = None | |
| if not is_spam: | |
| categories = ["inquiry", "complaint", "thank you", "request", "information"] | |
| for category in categories: | |
| if category in response_text: | |
| email_category = category | |
| break | |
| # 更新消息以进行追踪 | |
| new_messages = state.get("messages", []) + [ | |
| {"role": "user", "content": prompt}, | |
| {"role": "assistant", "content": response.content} | |
| ] | |
| # 返回状态更新 | |
| return { | |
| "is_spam": is_spam, | |
| "spam_reason": spam_reason, | |
| "email_category": email_category, | |
| "messages": new_messages | |
| } | |
| def route_email(state: EmailState) -> str: | |
| """Determine the next step based on spam classification""" | |
| # print(f"route_email,state={state}") | |
| if state["is_spam"]: | |
| return "spam" | |
| else: | |
| return "legitimate" | |
| def handle_spam(state: EmailState): | |
| """Alfred discards spam email with a note""" | |
| # print(f"handle_spam,state={state}") | |
| # print(f"is_spam:{state.get("is_spam","none")}") | |
| print(f"Alfred has marked the email as spam. Reason: {state.get("spam_reason","none")}") | |
| print("The email has been moved to the spam folder.") | |
| # 我们已处理完这封电子邮件 | |
| return {} | |
| def draft_response(state: EmailState): | |
| """Alfred drafts a preliminary response for legitimate emails""" | |
| email = state["email"] | |
| category = state["email_category"] or "general" | |
| # 为 LLM 准备提示词 | |
| prompt = f""" | |
| As Alfred the butler, draft a polite preliminary response to this email. | |
| Email: | |
| From: {email['sender']} | |
| Subject: {email['subject']} | |
| Body: {email['body']} | |
| This email has been categorized as: {category} | |
| Draft a brief, professional response that Mr. Hugg can review and personalize before sending. | |
| """ | |
| # Call the LLM | |
| # messages = [HumanMessage(content=prompt)] | |
| response = query_llm(prompt) | |
| # 更新消息以进行追踪 | |
| new_messages = state.get("messages", []) + [ | |
| {"role": "user", "content": prompt}, | |
| {"role": "assistant", "content": response.content} | |
| ] | |
| # 返回状态更新 | |
| return { | |
| "draft_response": response.content, | |
| "messages": new_messages | |
| } | |
| def notify_mr_hugg(state: EmailState): | |
| """Alfred notifies Mr. Hugg about the email and presents the draft response""" | |
| email = state["email"] | |
| print("\n" + "="*50) | |
| print(f"Sir, you've received an email from {email['sender']}.") | |
| print(f"Subject: {email['subject']}") | |
| print(f"Category: {state['email_category']}") | |
| print("\nI've prepared a draft response for your review:") | |
| print("-"*50) | |
| print(state["draft_response"]) | |
| print("="*50 + "\n") | |
| # 我们已处理完这封电子邮件 | |
| return {} | |
| # 创建 graph | |
| email_graph = StateGraph(EmailState) | |
| # 添加 nodes | |
| email_graph.add_node("read_email", read_email) | |
| email_graph.add_node("classify_email", classify_email) | |
| email_graph.add_node("handle_spam", handle_spam) | |
| email_graph.add_node("draft_response", draft_response) | |
| email_graph.add_node("notify_mr_hugg", notify_mr_hugg) | |
| # 添加 edges - 定义流程 | |
| email_graph.add_edge(START, "read_email") | |
| email_graph.add_edge("read_email", "classify_email") | |
| # 从 classify_email 添加条件分支 | |
| email_graph.add_conditional_edges( | |
| "classify_email", | |
| route_email, | |
| { | |
| "spam": "handle_spam", | |
| "legitimate": "draft_response" | |
| } | |
| ) | |
| # 添加最后的 edges | |
| email_graph.add_edge("handle_spam", END) | |
| email_graph.add_edge("draft_response", "notify_mr_hugg") | |
| email_graph.add_edge("notify_mr_hugg", END) | |
| # 编译 graph | |
| compiled_graph = email_graph.compile() | |
| # 合法电子邮件示例 | |
| legitimate_email = { | |
| "sender": "john.smith@example.com", | |
| "subject": "Question about your services", | |
| "body": "Dear Mr. Hugg, I was referred to you by a colleague and I'm interested in learning more about your consulting services. Could we schedule a call next week? Best regards, John Smith" | |
| } | |
| # 垃圾邮件示例 | |
| spam_email = { | |
| "sender": "winner@lottery-intl.com", | |
| "subject": "YOU HAVE WON $5,000,000!!!", | |
| "body": "CONGRATULATIONS! You have been selected as the winner of our international lottery! To claim your $5,000,000 prize, please send us your bank details and a processing fee of $100." | |
| } | |
| def classify_email(json_input): | |
| email = json.loads(json_input) | |
| result = compiled_graph.invoke({ | |
| "email": email, | |
| "is_spam": None, | |
| "spam_reason": None, | |
| "email_category": None, | |
| "draft_response": None, | |
| "messages": [] | |
| }) | |
| return f"is_spam:{result.get("is_spam")}\nspam_reason:{result.get("spam_reason")}" | |
| # ===================== 核心处理函数（模拟邮件分析） ===================== | |
| def classify_email(sender: str, subject: str, body: str) -> tuple[str, str]: | |
| if sender.strip() == "" or subject.strip() == "" or body.strip() == "" : | |
| return "true", "sender、subject、subject为空" | |
| email = { | |
| "sender": sender, | |
| "subject": subject, | |
| "body": body | |
| } | |
| result = compiled_graph.invoke({ | |
| "email": email, | |
| "is_spam": None, | |
| "spam_reason": None, | |
| "email_category": None, | |
| "draft_response": None, | |
| "messages": [] | |
| }) | |
| return result.get("is_spam"), result.get("spam_reason") | |
| # ===================== Gradio 表单界面构建 ===================== | |
| with gr.Blocks(title="邮件分析表单", theme=gr.themes.Soft()) as demo: | |
| # 标题说明 | |
| gr.Markdown("### 📧 邮件垃圾信息分析表单") | |
| gr.Markdown("请输入以下3项信息，提交后自动分析邮件是否为垃圾邮件") | |
| # 表单区域：3个文本输入框 | |
| with gr.Row(): # 行布局，可选：改为 gr.Column() 垂直布局 | |
| with gr.Column(scale=1): | |
| sender_input = gr.Textbox( | |
| label="1. 发件人邮箱地址", | |
| placeholder="john.smith@gmail.com", | |
| lines=1, | |
| max_lines=1 | |
| ) | |
| with gr.Column(scale=2): | |
| subject_input = gr.Textbox( | |
| label="2. 邮件主题", | |
| placeholder="预约会议", | |
| lines=1, | |
| max_lines=1 | |
| ) | |
| # 第三个文本框：多行输入（邮件内容） | |
| content_input = gr.Textbox( | |
| label="3. 邮件内容", | |
| placeholder="我想预约下周二的咨询会议", | |
| lines=5, | |
| max_lines=10 | |
| ) | |
| # 提交按钮 | |
| submit_btn = gr.Button("📤 提交分析", variant="primary", size="lg") | |
| # 分割线 | |
| # gr.Divider() | |
| # 结果展示区域（结构化输出） | |
| gr.Markdown("### 📝 分析结果") | |
| with gr.Row(): | |
| is_spam_output = gr.Textbox(label="是否垃圾邮件", interactive=False) | |
| reason_output = gr.Textbox(label="判定原因", lines=3, interactive=False) | |
| # 绑定按钮与处理函数 | |
| submit_btn.click( | |
| fn=classify_email, | |
| inputs=[sender_input, subject_input, content_input], # 3个输入框 | |
| outputs=[is_spam_output, reason_output] # 2个输出框 | |
| ) | |
| # 重置按钮（可选） | |
| reset_btn = gr.Button("🔄 清空输入") | |
| reset_btn.click( | |
| fn=lambda: ("", "", "", "", "", ""), # 重置所有输入输出 | |
| inputs=[], | |
| outputs=[sender_input, subject_input, content_input, is_spam_output, reason_output] | |
| ) | |
| # ===================== 启动界面 ===================== | |
| if __name__ == "__main__": | |
| # server_name="0.0.0.0" 允许局域网访问，server_port 自定义端口 | |
| demo.launch() | |
| # demo = gr.Interface(fn=classify_email, inputs="text", outputs="text") | |
| # demo.launch() |