Spaces:
Runtime error
Runtime error
File size: 10,179 Bytes
2756582 532a759 2756582 1543ec3 2756582 1543ec3 2756582 1543ec3 2756582 1543ec3 2756582 1543ec3 2756582 1543ec3 532a759 1543ec3 2756582 1543ec3 2756582 1543ec3 2756582 1543ec3 2756582 1543ec3 2756582 1543ec3 2756582 1543ec3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 | import json
import gradio as gr
from fastapi.encoders import jsonable_encoder
from langchain.callbacks import get_openai_callback
from edu_assistant.learning_tasks.coding_problem import (
DEFAULT_FIRST_QUESTION,
DEFAULT_INSTRUCTION,
CodingProblem,
CodingProblemAnalysis,
)
from edu_assistant.utils.langchain_utils import load_vectorstore, shrink_docs
CodingProblem.enable_redis_orm()
class CodingProblemUI:
def __init__(
self,
*,
instruction: str = DEFAULT_INSTRUCTION,
first_question: str = DEFAULT_FIRST_QUESTION,
knowledge_name: str = "example",
enable_gpt4: bool = False,
):
self._init_task(instruction, first_question, knowledge_name, enable_gpt4)
self._init_ui()
def ui_render(self):
self.ui.render()
def ui_reload(
self,
*,
instruction: str = DEFAULT_INSTRUCTION,
first_question: str = DEFAULT_FIRST_QUESTION,
knowledge_name: str = "example",
enable_gpt4: bool = False,
refresh: bool = True,
):
self._init_task(instruction, first_question, knowledge_name, enable_gpt4)
if refresh:
self.ui_render()
def get_instruction(self):
return self.instruction
def get_first_question(self):
return self.first_question
def _init_task(self, instruction: str, first_question: str, knowledge_name: str, enable_gpt4: bool):
self.instruction = instruction
self.first_question = first_question
self.knowledge = knowledge_name
self.enable_gpt4 = enable_gpt4
self.task = CodingProblemAnalysis(
instruction=instruction,
first_question=first_question,
knowledge=load_vectorstore(knowledge_name).as_retriever(),
enable_gpt4=enable_gpt4,
)
def _init_ui(self):
self.ui = gr.Blocks()
with self.ui:
with gr.Row():
with gr.Column(scale=6):
problem_selector = gr.Dropdown(choices=self._get_problems(), show_label=False, interactive=True)
with gr.Column(scale=1):
refresh_btn = gr.Button(value="刷新")
with gr.Row():
with gr.Column(scale=6):
with gr.Tab(label="错误代码分析"):
with gr.Row():
with gr.Column(scale=3):
with gr.Row():
problem_view = gr.Markdown(label="题目")
with gr.Row():
code_view = gr.Textbox(label="代码", lines=10, interactive=True)
with gr.Column(scale=3):
with gr.Row():
chat_box = gr.Chatbot(height=500, label="聊天记录")
with gr.Row():
chat_input = gr.Textbox(show_label=False)
with gr.Column():
with gr.Row():
analysis_btn = gr.Button(value="分析")
with gr.Row():
clear = gr.ClearButton()
with gr.Row():
session_id = gr.Textbox(label="Session", interactive=False, value="")
with gr.Row():
status = gr.JSON(value={"tokens": 0}, label="Status")
with gr.Row():
docs = gr.JSON(value=["docs"], label="Docs")
with gr.Tab(label="题库管理"):
with gr.Row():
with gr.Column(scale=6):
with gr.Row():
title_edit = gr.Textbox(label="标题", interactive=True)
with gr.Row():
language_edit = gr.Dropdown(
choices=["python", "cpp", "java"],
label="语言",
interactive=True,
allow_custom_value=True,
)
with gr.Column(scale=1):
manage_update = gr.Button(value="更新")
manage_delete = gr.Button(value="删除", variant="stop")
with gr.Row():
with gr.Column():
problem_edit = gr.Textbox(label="题目", lines=10, max_lines=100, interactive=True)
with gr.Column():
analysis_edit = gr.Textbox(label="解析", lines=10, max_lines=100, interactive=True)
with gr.Row():
answer_edit = gr.Textbox(label="标准答案", lines=10, max_lines=100, interactive=True)
with gr.Row():
extra_edit = gr.Textbox(label="额外信息", lines=10, max_lines=100, interactive=True)
refresh_btn.click(self._update_problems, [], [problem_selector])
problem_selector.select(
self._select_problem,
[
problem_selector,
],
[problem_view, title_edit, language_edit, problem_edit, analysis_edit, answer_edit, extra_edit],
)
analysis_btn.click(
self._analysis_problem,
[problem_selector, code_view],
[chat_box, session_id, status, docs],
)
chat_input.submit(self._chat, [chat_input, chat_box, session_id], [chat_input, chat_box, status, docs])
manage_update.click(
self._update_problem,
[title_edit, language_edit, problem_edit, analysis_edit, answer_edit, extra_edit],
[],
)
manage_delete.click(
self._delete_problem,
[problem_selector],
[
problem_selector,
problem_view,
title_edit,
language_edit,
problem_edit,
analysis_edit,
answer_edit,
extra_edit,
],
)
clear.click(
self._clear,
[],
[
problem_selector,
problem_view,
code_view,
chat_box,
session_id,
status,
docs,
problem_view,
title_edit,
language_edit,
problem_edit,
analysis_edit,
answer_edit,
extra_edit,
],
)
def _get_problems(self) -> list[str]:
data = CodingProblem.select(columns=["title"])
if not data:
return []
titles = [problem_data["title"] for problem_data in data]
return titles
def _update_problems(self):
titles = self._get_problems()
gr.Info("更新题目列表成功")
return gr.Dropdown.update(choices=titles)
def _select_problem(self, title: str):
problem: CodingProblem = CodingProblem.select(ids=[title])[0]
return (
problem.expr(),
problem.title,
problem.language,
problem.question,
problem.analysis,
problem.standard_answer,
json.dumps(problem.extra, ensure_ascii=False, indent=4),
)
def _update_problem(self, title, language, problem, analysis, answer, extra):
# TODO: add language
try:
extra_data = json.loads(extra)
except json.JSONDecodeError:
extra_data = [extra]
CodingProblem.update(
title,
data={
"title": title,
"language": language,
"question": problem,
"analysis": analysis,
"standard_answer": answer,
"extra": extra_data,
},
)
gr.Info("更新题目成功")
def _delete_problem(self, title):
CodingProblem.delete(ids=[title])
gr.Info("删除题目成功")
return "", "", "", "", "", "", "", ""
def _analysis_problem(self, title, code, extra: str = ""):
problem = CodingProblem.select(ids=[title])[0]
answer = CodingProblemAnalysis.build_coding_answer(answer=code, extra=[extra])
with get_openai_callback() as cb:
result = self.task.start_analysis(problem, answer)
status = {"tokens": cb.total_tokens, "cost": f"${cb.total_cost:.4f}"}
answer = result["answer"]
session_id = result["session_id"]
docs = jsonable_encoder(shrink_docs(result.get("source_documents", [])))
return [(self.first_question, answer)], session_id, status, docs
def _chat(self, message, chat_history, session_id):
if not session_id:
return "", "", {"tokens": 0}, []
with get_openai_callback() as cb:
result = self.task.ask(message, session_id=session_id)
if not result:
raise gr.Error("Session expired. Please recreate a new problem analysis session.")
session_id = result["session_id"]
docs = jsonable_encoder(result.get("source_documents", []))
bot_message = result["answer"]
chat_history.append((message, bot_message))
status = {"tokens": cb.total_tokens, "cost": f"${cb.total_cost:.4f}"}
return "", chat_history, status, docs
def _clear(self):
return "", "", "", [], "", {"tokens": 0}, ["docs"], "", "", "", "", "", "", ""
|