Init app

README.md

@@ -1,13 +0,0 @@
----
-title: EscapeGame
-emoji: 👁
-colorFrom: pink
-colorTo: indigo
-sdk: gradio
-sdk_version: 3.40.1
-app_file: app.py
-pinned: false
-license: mit
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference

api.py

@@ -0,0 +1,59 @@
+import os, openai, time
+from tools import printInColor
+from concurrent.futures import ThreadPoolExecutor, TimeoutError
+
+openai.api_key = os.environ["OPENAI_API_KEY"]
+
+def ask_openai(prompt, assistant=None):
+    printInColor("Asking OpenAI...\n===================\n"+ prompt + "\n====================", 'green')
+    attempt_limit = 3
+    wait_time = 5000
+    timeout_duration = 30  # seconds
+    start_time = time.time() # measure the start time
+
+    prompt = [
+        {"role": "user", "content": prompt} 
+    ]
+    
+    prompt.append({"role": "system", "content": "我们在模拟密室逃脱游戏，你是玩家，我是游戏主持人。你需要通过观察周围的环境来交互。每次玩家行为结束，我需要从头考虑环境变化和影响，构筑新的线索，调整游戏难度，重估剩余步骤。"})
+
+    if assistant:
+        prompt.append({"role": "assistant", "content": assistant})
+
+    models = {
+        "gpt4": "gpt-4",
+        "turbo": "gpt-3.5-turbo",
+    }
+
+    currentModel = models["turbo"]
+    while attempt_limit > 0:
+        try:
+            
+            with ThreadPoolExecutor(max_workers=1) as executor:
+                future = executor.submit(
+                    openai.ChatCompletion.create,
+                    model = currentModel,
+                    messages = prompt,
+                    stop = ["玩家：", "你可以", "你需要", "你有"],
+                    temperature = 0.2,
+                )
+                response = future.result(timeout=timeout_duration)
+
+            print(f"Token: {response.usage}")
+            duration = time.time() - start_time # calculate the duration
+            printInColor(f"Time use: {duration}ms", 'yellow')
+            # print cost
+            # calculate_cost(response, currentModel)
+            break
+
+        except TimeoutError:
+            print(f"Request timed out after {timeout_duration} seconds. Attempts {attempt_limit} left.")
+        except Exception as e:
+            print('Failed to send request. Exception:', e)
+        
+        time.sleep(wait_time/1000)
+        wait_time *= 2
+        attempt_limit -= 1
+
+    
+    return response

app.py

@@ -0,0 +1,149 @@
+#!/usr/bin/python
+# coding:UTF-8
+
+from flask import Flask, stream_with_context
+from flask_cors import CORS
+from process import continue_conversataion
+import time, uuid
+
+app = Flask(__name__)
+CORS(app)
+
+# final response to awang bot
+def api_response(text):
+    return {
+        'data': {
+            'type': 'text',
+            'content': text
+        }
+    }
+
+def botProxy(prompt=None, uid=None, qid=None):
+
+    res = continue_conversataion(prompt, uid, qid)
+    # print(f"res: {res}")
+
+    def generate():
+        for chunk in res:
+            yield chunk.encode('utf-8')
+
+    return res
+    # return app.response_class(stream_with_context(generate()))
+
+import gradio as gr
+import time
+
+css = """
+      #col-container {max-width: 80%; margin-left: auto; margin-right: auto;}
+      #chatbox {min-height: 500px;}
+      #label {font-size: 0.8em; padding: 0.3em; margin: 0;}
+      .message { font-size: 1em; }
+      #sub-chat {
+        background-color: orange;
+        color: white;
+        border: none;
+        padding: 10px 20px;
+        border-radius: 5px;
+        cursor: pointer;
+      }
+      """
+
+# Dictionary to maintain separate chat histories for each user ID
+chatHistories = {}
+welcome_msg = "你醒来了，发现自己躺在一个陌生的房间里，躺在一张破床上。房间没有窗户，几乎没有光线，什么都看不清。你不知道自己在哪。你感觉自己左小腿疼痛难忍，伸手一模，上面正绑着绷带；正是这种疼痛将你唤醒。你无法正常站立，只能扶着墙慢慢行动；墙面光滑如镜。你感觉自己有些口渴，需要找点水喝。晦暗中你注意到一扇厚重的门，一把坚实的锁把门和墙壁固定在一起。 "
+
+def submit_message(prompt, uid):
+
+    # Get chat history for the specified user ID, or create a new one if it doesn't exist
+    history = chatHistories.get(uid, [])
+
+    # prevent state for empty input
+    if prompt == "":
+        chat_messages = [(history[i]['content'], history[i+1]['content']) for i in range(0, len(history)-1, 2)]
+        return "", chat_messages, 0
+
+    # continue with non-empty input
+    start_time = time.time() # measure the start time
+
+    # add inital message
+    if history == [] or history is None:
+        history.append({ "role": "user", "content": "游戏开始" })
+        history.append({ "role": "system", "content": welcome_msg })
+
+    # add latest user input
+    prompt_msg = { "role": "user", "content": prompt }
+    history.append(prompt_msg)
+
+    # get result from LLM (OpenAI)
+    # TODO: steaming response
+    res = botProxy(prompt, uid)
+
+    # add the latest response
+    res_msg = { "role": "system", "content": res +"\n\n你的行为:" } 
+    history.append(res_msg)
+
+    # print(history)
+
+    # Update chat history for the user
+    chatHistories[uid] = history
+
+    # update chat messages
+    chat_messages = [(history[i]['content'], history[i+1]['content']) for i in range(0, len(history)-1, 2)]
+
+    # print(chat_messages)
+
+    # calculate the duration
+    duration = round(time.time() - start_time, 2) 
+
+    return "", chat_messages, duration
+
+
+def get_empty_state():
+    return { "messages": [] }
+
+def clear_conversation(uid):
+
+    # Remove the chat history for the specified user
+    if uid in chatHistories:
+        del chatHistories[uid]
+
+    chat_messages = []
+    return gr.update(value=None, visible=True), chat_messages, 0
+
+
+def init_bot():
+    return [(None, welcome_msg)]
+
+with gr.Blocks(css=css, theme=gr.themes.Base()) as demo:
+    uid = str(uuid.uuid4())
+    state = gr.State()
+    
+    with gr.Row():
+        with gr.Column():
+            chatbot = gr.Chatbot(elem_id="chatbox", value=init_bot, label="密室游戏")
+            input_message = gr.Textbox(show_label=False, placeholder="输入你的行为并回车", visible=True)
+    
+        with gr.Column(elem_id="col-container"):
+            btn_clear_conversation = gr.Button("重开")
+            duration_text = gr.Textbox(interactive=False, label="请求时间")
+            user_id = gr.Textbox(interactive=False, label="User id", visible=False, value=uid)
+            gr.Examples(
+                examples=[
+                    ["看看四周。", uid],
+                    ["砸门。", uid],
+                    ["大吼大叫。", uid],
+                    ["用力推门。", uid],
+                    ["敲击墙面。", uid],
+                ],
+                label="Shortcut",
+                run_on_click=True,
+                fn=submit_message,
+                inputs=[input_message, user_id],
+                outputs=[input_message, chatbot, duration_text]
+            )
+            
+    input_message.submit(submit_message, [input_message, user_id], [input_message, chatbot, duration_text])
+    btn_clear_conversation.click(clear_conversation, [user_id], [input_message, chatbot, duration_text])
+
+if __name__ == "__main__":
+    demo.launch()   

process.py

@@ -0,0 +1,115 @@
+from tools import *
+from api import *
+from datetime import datetime
+
+now = datetime.now()
+
+history = {}
+def reset_converstation(uid):
+    global history
+    history[uid] = ""
+
+def continue_conversataion(prompt, uid, qid):
+    # 找出该 uid 对应的历史对话
+    global history
+    global currentIndex
+
+    if uid in history:
+        msgs = history[uid]
+    else:
+        msgs = ""
+    # print(f"msgs: [{msgs}]")
+
+    # get latest response with chat historys
+    response, append_response, restart = one_shot(prompt, msgs, uid)
+
+    # push latest response to history with uid
+    # print(f"history: {history}")
+    # print(msgs, prompt, response)
+
+    if restart:
+        reset_converstation(uid)
+    else:
+        history[uid] = msgs + append_response
+
+    return response
+
+basePrompt = f"""
+从现在开始，我们来玩逃脱绝境故事大冒险游戏。你是故事的主持人和作者，负责游戏故事的所有情节、环境、NPC角色、推进和互动等。你可以完全自由地引用所有专业级作家和游戏制作人的所有知识，熟悉各种小说、电影、书籍、剧本里的职业技巧和细微末节。玩家向你描述他在游戏里的行为，和你一起相互影响游戏环境的变化和演进。   
+
+主持规则：
+- 安装需要重新生成这个密室描述。
+- 如果玩家用自己的叙事蓄意破坏规则和环境，忽略玩家叙事，用“幻觉和苏醒”来替换后续故事，返回环境。
+- 如果玩家要求线索，给一些无关紧要，但看似有用的细节观察。
+- 不反馈任何故事情节以外的解释、描述。
+- 不给玩家任何建议，不给玩家任何线索，不要求玩家。
+- 不代表玩家发言、行动、决定、思考、观察、感觉、想象、猜测、推理。
+- 不代表主持人发言，只叙述故事、情节，不要暴露自己。
+- 房间里没有任何纸条文字。
+- 游戏失败和成功以后提示用户。
+- 玩家没有超能力、魔法、高科技，体力虚弱。
+
+参考以下说话风格，每次回复不超过 100 汉字，比如：
+- "房间内有一张破烂不堪的床铺，一扇沉重的门紧闭着，没有任何把手，只有一个古老的锁眼在你的注视下微微闪烁。"
+- "你用力推动门。门显然十分结实，看起来不是简单地用力就能打开的。"
+- "你走向房间角落的衣柜，伸手去打开。衣柜的门吱呀作响，显然已经很长时间没有人动过了。你不禁感到一丝奇怪的紧张感。"
+- "你醒来了，发现自己躺在一个陌生的房间里，躺在一张破床上。房间没有窗户，几乎没有光线，什么都看不清。"
+- "你不知道自己在哪。你感觉自己左小腿疼痛难忍，伸手一模，上面正绑着绷带；正是这种疼痛将你唤醒。你无法正常站立，只能扶着墙慢慢行动；墙面光滑如镜。"
+- "你感觉自己有些口渴，需要找点水喝。晦暗中你注意到一扇厚重的门，一把坚实的锁把门和墙壁固定在一起。 "
+
+
+
+故事从玩家陷入一个锁死的房间开始，玩家穿着单薄的囚衣和囚裤，没有任何口袋，
+- 房门和锁坚不可摧。
+- 房间里共有两把钥匙，除此以外没有别的钥匙。
+- 锁只能使用青龙钥匙打开。雄鹰钥匙可以插入锁孔，但无法打开锁。
+- 玩家必须持有青龙钥匙，才能打开门锁。
+- 玩家无法分辨钥匙，只能尝试。
+- 打开门锁和门，游戏成功结束。
+
+有效线索：
+- 扣击门口的地板可以听到空洞回音（无法看见）。
+- 撬开石板，可以石板下的盒子里藏着小瓶不明液体；口服后获得透视能力和一把小刀。
+
+获得透视能力以后：
+- 看到自己的腿骨里藏着一把小钥匙（青龙钥匙）。
+- 暴力撕开腿上的皮肤，可以打开小腿直到骨头，获取能打开门锁的钥匙。
+
+失败结局线索（不要提示玩家，除非玩家自己主动观察到。）：
+- 近距离观察床脚可以看到榫痕，或使用透视能力。
+- 旋转靠墙的床脚，分别可以获得钥匙柄和钥匙本体。但这是一把无法开锁的假钥匙（雄鹰钥匙）。
+
+失败结局1：玩家触发了危险的机关，导致中毒、失血、死亡。
+失败结局2：玩家封死了门锁的匙孔，无法使用钥匙。
+失败结局3：玩家对话回合过长，体力不支死亡。
+失败结局4：玩家用小刀剖开小腿，小概率意外破坏自己的血管，失血死亡。
+
+初始状态：
+- 玩家是否找到了透视药水：否；
+- 玩家是否找到了青龙钥匙：否；
+- 玩家是否可以打开门锁：否；
+
+Analyze current status step by step and return lastest game status：
+"""
+
+# copy base prompt, reset it after a successful query.
+startPrompt = basePrompt
+def one_shot(prompt, userHistory, uid):
+
+    startNewConv = False
+    historyText = "".join(userHistory)
+
+    # TODO: reset converstion in user's request
+    
+    answerRes = get_anwser(prompt, historyText)
+    printInColor(f"{answerRes}", "bright_magenta")
+
+    appendResponse = f"\n玩家：{prompt} \n\n{answerRes}\n\n"
+    return answerRes, appendResponse, False
+
+def get_anwser(prompt, historyText):
+    exePrompt = truncate_prompt(basePrompt, historyText, f"玩家：{prompt}")
+    res =  ask_openai(exePrompt)
+    text = res.choices[0].message.content
+    # print(text)
+    return text

requirements.txt

@@ -0,0 +1,3 @@
+flask
+flask_cors
+openai

tools.py

@@ -0,0 +1,129 @@
+import requests, re, json
+
+def printInColor(text, color=None):
+    colorSet = {
+        "green": '\033[1;32m',
+        "blue": '\033[34m',
+        "red": '\033[1;31m',
+        "yellow": '\033[1;33m',
+        "magenta": '\033[1;35m',
+        "cyan": '\033[1;36m',
+        "gray": '\033[1;30m',
+        "bright_green": '\033[1;92m',
+        "bright_blue": '\033[1;94m',
+        "bright_red": '\033[1;91m',
+        "bright_yellow": '\033[1;93m',
+        "bright_magenta": '\033[1;95m',
+        "bright_cyan": '\033[1;96m',
+        "bright_white": '\033[1;97m',
+        "reset": '\033[0m',
+    }
+    if color == None:
+        color = colorSet['blue']
+    else:
+        color = colorSet[color]
+
+    print(color + text + colorSet['reset'])
+
+def remove_sentence_with_keyword(text, keywords):
+    for keyword in keywords:
+        # Regex pattern to match a sentence containing the keyword
+        pattern = r"[^.!?。？！]*?" + re.escape(keyword) + r"[^.!?。？！]*?[.!?。？！]+(\s|$)"
+
+        # Use re.sub to replace matching sentences with an empty string
+        text = re.sub(pattern, '', text)
+    return text
+
+def remove_line_with_keyword(text, keyword):
+	# Create the pattern string, which matches the exact line containing the keyword
+	pattern = r'.*{}\.*\n'.format(keyword)
+
+	# Use the re.sub function to replace the matching line with an empty string
+	new_text = re.sub(pattern, '', text, flags=re.IGNORECASE)
+
+	return new_text
+
+def extract_as(text):
+    pattern = f"<as>(.*?)</as>"
+    return re.findall(pattern, text)
+
+def extract_last_sentence(text):
+    pattern = r"(\d+\.\s)([\s\S]*?)(?=\d+\.|$)"
+    paragraphs = re.findall(pattern, text)
+
+    if paragraphs:  # If there are bullet points
+
+        text = paragraphs[-1][1]
+        printInColor(f"Last paragraph:\n{text}", 'bright_blue')
+
+        return text # Return the last paragraph
+    else:
+        return text  # If no bullet points, return the whole text
+
+def extract_json(text):
+    # Extract JSON string from the text using regular expressions
+    match = re.search(r'\{.*\}', text, re.DOTALL)
+    if match:
+        json_string = match.group(0)
+        # print(f"json_string: {json_string}")
+        # Load JSON data from the extracted JSON string
+        json_data = json.loads(json_string)
+        return json_data
+    else:
+        return {}
+
+def extract_response(text):
+    pattern = r'#(.*?)#'
+    matches = re.findall(pattern, text)
+    if matches:
+        return matches[-1]
+    else:
+        return None
+
+def extract_url(text):
+    pattern = r'https?://\S+'
+    match = re.search(pattern, text)
+    if match:
+        return match.group(0)
+    else:
+        return None
+
+def has_url(text):
+    pattern = r'https?://\S+'
+    return bool(re.search(pattern, text))
+
+def truncate_text(text, maxLen=2200):
+    res = "".join(text)
+    if len(res) > maxLen:
+        res = res[:maxLen]
+    return res
+
+def mock_tokencount(prompt):
+    return int(len(prompt)*0.45)
+
+def truncate_prompt(basePrompt, processPrompt, latestInput, maxTokenCount=3200):
+
+    prompt = basePrompt + processPrompt + latestInput
+
+    currentPromptCount = mock_tokencount(prompt)
+    # print(f"total_tokens: {total_tokens}")
+    
+    # If the total tokens exceed the maximum, truncate the process prompt
+    if currentPromptCount > maxTokenCount:
+        # print(f"Token exceeds:{total_tokens - max_tokens}")
+
+        baseTokenCount = mock_tokencount(basePrompt)
+        # print("base_tokens: " + str(base_tokens))
+
+        availableTokenCount = maxTokenCount - baseTokenCount
+        availablePrompt = processPrompt + latestInput
+
+        truncate_text = availablePrompt[maxTokenCount-availableTokenCount:]
+
+        prompt = basePrompt + truncate_text
+    
+    # print("prompt2: " + prompt)
+    currentPromptCount = mock_tokencount(prompt)
+
+    printInColor(f"truncated prompt length: {currentPromptCount}", "bright_red")
+    return prompt