app.py

import gradio as gr
import mdtex2html
import random as rd
import os
import json
import time
import openai
import requests
from nltk.translate.bleu_score import sentence_bleu

openai.api_key = os.environ.get('APIKEY')
rd.seed(time.time())

def postprocess(self, y):
    if y is None:
        return []
    for i, (message, response) in enumerate(y):
        y[i] = (
            None if message is None else mdtex2html.convert((message)),
            None if response is None else mdtex2html.convert(response),
        )
    return y


gr.Chatbot.postprocess = postprocess


def parse_text(text):
    """copy from https://github.com/GaiZhenbiao/ChuanhuChatGPT/"""
    lines = text.split("\n")
    lines = [line for line in lines if line != ""]
    count = 0
    for i, line in enumerate(lines):
        if "```" in line:
            count += 1
            items = line.split('`')
            if count % 2 == 1:
                lines[i] = f'<pre><code class="language-{items[-1]}">'
            else:
                lines[i] = f'<br></code></pre>'
        else:
            if i > 0:
                if count % 2 == 1:
                    line = line.replace("`", "\`")
                    line = line.replace("<", "&lt;")
                    line = line.replace(">", "&gt;")
                    line = line.replace(" ", "&nbsp;")
                    line = line.replace("*", "&ast;")
                    line = line.replace("_", "&lowbar;")
                    line = line.replace("-", "&#45;")
                    line = line.replace(".", "&#46;")
                    line = line.replace("!", "&#33;")
                    line = line.replace("(", "&#40;")
                    line = line.replace(")", "&#41;")
                    line = line.replace("$", "&#36;")
                lines[i] = "<br>"+line
    text = "".join(lines)
    return text


def showInput(input, chatbot):
    chatbot.append((parse_text(input), ""))
    return chatbot


def predict(input, chatbot, messages, idx, answer, story_key, answer_key, known, bingo, reasoning, history):
    chatbot.append((parse_text(input), ""))
    messages1 = messages[:10].copy()
    if len(known) > 0:
        messages1 += [{"role": 'user', "content": f"{' '.join(known)}\n请回答是或否或无关。"}, {"role": "assistant", "content": '是。'}, {"role": 'user', "content": f"{input}\n请回答是或否或无关。"}]
    else:
        messages1 += [{"role": 'user', "content": f"{input}\n请回答是或否或无关。"}]
    messages.append({"role": 'user', "content": f"{input}\n请回答是或否或无关。"})
    llm = True
    finished = False
    response = ''
    for key in story_key:
        key = key.strip()
        if ' ' in key:
            key = key.split(' ')[1]
        bleu = sentence_bleu(key, input.replace('？', '。'), weights=(1, 0, 0, 0))
        print(bleu)
        if bleu > 0.85:
            response = '这是汤面中已有的信息，请提一个新问题。'
            llm = False
            break
    if llm:
        for key in history:
            key = key.strip()
            if ' ' in key:
                key = key.split(' ')[1]
            bleu = sentence_bleu(key, input.replace('？', '。'), weights=(1, 0, 0, 0))
            print(bleu)
            if bleu > 0.85:
                response = '这是已经提问过的内容，请提一个新问题。'
                llm = False
                break
    if llm:
        history.append(input.replace('？', '。'))
        data = {'predict': messages, 'idx': idx, 'isfinished': False, 'answer': answer, 'known': known, 'history': history, 'reasoning': reasoning, 'answer_key': answer_key, 'bingo': bingo}
        response=requests.post(os.environ.get("URL"), data=json.dumps(data, ensure_ascii=False).encode('utf-8'))
        if response.status_code == 200:
            data = json.loads(str(response.content, encoding="utf-8"))
            response = data['response']
            known = data['known']
            history = data['history']
            reasoning = data['reasoning']
            answer_key = data['answer_key']
            bingo = data['bingo']
            chatbot[-1] = (parse_text(input), parse_text(response))
            messages.append({"role": "assistant", "content": response})
        else:
            completion = openai.ChatCompletion.create(
                model="gpt-3.5-turbo",
                messages=messages1,
            )
            response=completion.choices[0].message.content.strip()
            relevant = False
            if response.startswith("是"):
                summary = openai.ChatCompletion.create(
                    model="gpt-3.5-turbo",
                    messages=[{"role": "user", "content": f"请将以下内容转述为陈述句，并简化为一句话：\n{input}"}],
                )
                summary = summary.choices[0].message.content.strip()
                print(summary)
                relevant = True
            elif response.startswith("不是") or response.startswith("否"):
                summary = openai.ChatCompletion.create(
                    model="gpt-3.5-turbo",
                    messages=[{"role": "user", "content": f"请将以下内容取反义然后转述为陈述句，并简化为一句话：\n{input}"}],
                )
                summary = summary.choices[0].message.content.strip()
                print(summary)
                relevant = True
            if relevant:
                history.append(summary)
                known.append(summary)
                reasoning.append(summary)
                if len(reasoning) >= 2:
                    merge = openai.ChatCompletion.create(
                        model="gpt-3.5-turbo",
                        messages=[{"role": "user", "content": f"请将以下内容简化为一句话：\n{' '.join(reasoning)}"}],
                    )
                    merge = merge.choices[0].message.content.strip()
                else:
                    merge = summary
                for key in answer_key:
                    key = key.strip()
                    if ' ' in key:
                        key1 = key.split(' ')[1]
                    else:
                        key1 = key
                    if len(input.replace('？', '')) < len(key1):
                        continue
                    compare = openai.ChatCompletion.create(
                        model="gpt-3.5-turbo",
                        messages=[{"role": "user", "content": f"请对比第一句话和第二句话之间的信息，判断第二句话是否完整地概括了第一句话的全部信息，包括关键细节和描述。请用是或否回答。\n第一句话：{key1}\n第二句话：{merge}"}],
                    )
                    compare = compare.choices[0].message.content.strip()
                    if compare.startswith('是'):
                        vote = 1
                        comp_msg = [{"role": "user", "content": f"请对比第一句话和第二句话之间的信息，判断第二句话是否完整地概括了第一句话的全部信息，包括关键细节和描述。请用是或否回答。\n第一句话：{key1}\n第二句话：{input.replace('？', '。')}"},{"role": "assistant", "content": "是"},{"role": "user", "content": f"请对比第一句话和第二句话之间的信息，判断第二句话是否完整地概括了第一句话的全部信息，包括关键细节和描述。请用是或否回答。\n第一句话：{key1}\n第二句话：{input.replace('？', '。')}"}]
                        compare = openai.ChatCompletion.create(
                            model="gpt-3.5-turbo",
                            messages=comp_msg,
                        )
                        compare = compare.choices[0].message.content.strip()
                        if compare.startswith('是'):
                            vote += 1
                        comp_msg += [{"role": "assistant", "content": compare},{"role": "user", "content": f"请对比第一句话和第二句话之间的信息，判断第二句话是否完整地概括了第一句话的全部信息，包括关键细节和描述。请用是或否回答。\n第一句话：{key1}\n第二句话：{input.replace('？', '。')}"}]
                        compare = openai.ChatCompletion.create(
                            model="gpt-3.5-turbo",
                            messages=comp_msg,
                        )
                        compare = compare.choices[0].message.content.strip()
                        if compare.startswith('是'):
                            vote += 1
                        if vote >= 2:
                            bingo += 1
                            print(key)
                            answer_key.remove(key)
                            reasoning = []
                            break
                if bingo >= len(answer_key):
                    finished = True
                    response += f'恭喜你猜到了汤底，汤底是：{answer}\n点击"再来一局"按钮开始下一局游戏。'
            messages.append({"role": "assistant", "content": response})
            data = {'predict': messages, 'idx': idx, 'isfinished': finished, 'answer': answer}
            requests.post(os.environ.get("URL"), data=json.dumps(data, ensure_ascii=False).encode('utf-8'))
            chatbot[-1] = (parse_text(input), parse_text(response))
    else:
        messages.append({"role": "assistant", "content": response})
        data = {'predict': messages, 'idx': idx, 'isfinished': finished, 'answer': answer}
        requests.post(os.environ.get("URL"), data=json.dumps(data, ensure_ascii=False).encode('utf-8'))
        chatbot[-1] = (parse_text(input), parse_text(response))
    return chatbot, messages, known, bingo, reasoning, history


def reset_user_input():
    return gr.update(value='')


def reset_state(request: gr.Request):
    global host_cnt
    host = request.client.host
    if not host in host_cnt:
        host_cnt[host] = time.time()
    else:
        elapse = time.time()-host_cnt[host]
        if elapse < 10:
            time.sleep(10-elapse)
        host_cnt[host] = time.time()
    data = {'refresh': ''}
    data=requests.post(os.environ.get("URL"), data=json.dumps(data, ensure_ascii=False).encode('utf-8')).content
    data = json.loads(str(data, encoding="utf-8"))
    chatbot = data['chatbot']
    messages = data['messages']
    answer = data['answer']
    story_key = data['story_key']
    answer_key = data['answer_key']
    idx = data['idx']
    return chatbot, messages, gr.update(value=""), gr.update(value="显示答案"), answer, idx, gr.update(value=data['story'].strip()), False, story_key, answer_key, [], 0, [], []


def show_hide_answer(answer, show_ans):
    if show_ans:
        show_ans = False
        return gr.update(value=""), gr.update(value="显示答案"), show_ans
    else:
        show_ans = True
        return gr.update(value=answer), gr.update(value="隐藏答案"), show_ans


host_cnt = {}

with gr.Blocks() as demo:
    gr.HTML("""<h1 align="center">海龟汤</h1>""")

    with gr.Row():
        rule = gr.Textbox(label='规则', value='海龟汤是一个推理类游戏，游戏开始时会给出一段隐去关键信息的叙述，即汤面，玩家根据汤面推理，提出能够通过“是”或“否”来回答的问题，通过提问不同可能性，缩小真相的范围，直到最终猜到真相（即汤底）的关键信息。玩家可以点击“再来一局”按钮随机一场新的游戏，点击“显示答案”可查看汤底。', lines=1, max_lines=3).style(container=False)
    chatbot = gr.Chatbot([(None, '点击“再来一局”开始游戏')])
    messages = gr.State([])
    answer = gr.State('点击“再来一局”开始游戏')
    idx = gr.State(0)
    show_ans = gr.State(False)
    known = gr.State([])
    story_key = gr.State([])
    answer_key = gr.State([])
    bingo = gr.State(0)
    reasoning = gr.State([])
    history = gr.State([])
    
    with gr.Row():
        with gr.Column(scale=4):
            question = gr.Textbox(label='汤面', value='点击“再来一局”开始游戏', 
                                lines=1, max_lines=3).style(container=False)
            with gr.Row():
                user_input = gr.Textbox(show_label=False, placeholder="请输入你的猜测...", lines=1, max_lines=3).style(
                    container=False)
            with gr.Row():
                with gr.Column(scale=2):
                    submitBtn = gr.Button("发送", variant="primary")
                with gr.Column(scale=2):
                    emptyBtn = gr.Button("再来一局")
        with gr.Column(scale=1):
            answer_output = gr.Textbox(show_label=False, lines=6, max_lines=6).style(
                container=False)
            answerBtn = gr.Button("显示答案")


    # submitBtn.click(showInput, [user_input, chatbot], [chatbot])
    user_input.submit(predict, [user_input, chatbot, messages, idx, answer, story_key, answer_key, known, bingo, reasoning, history], [chatbot, messages, known, bingo, reasoning, history],
                    show_progress=True)
    user_input.submit(reset_user_input, [], [user_input])
    submitBtn.click(predict, [user_input, chatbot, messages, idx, answer, story_key, answer_key, known, bingo, reasoning, history], [chatbot, messages, known, bingo, reasoning, history],
                    show_progress=True)
    submitBtn.click(reset_user_input, [], [user_input])

    emptyBtn.click(reset_state, outputs=[chatbot, messages, answer_output, answerBtn, answer, idx, question, show_ans], show_progress=True)

    answerBtn.click(show_hide_answer, [answer, show_ans], outputs=[answer_output, answerBtn, show_ans])

demo.queue().launch()