1  # app.py
  2  
  3  import os
  4  from http.client import HTTPMessage
  5  
  6  os.system('pip install dashscope')
  7  
  8  import gradio as gr
  9  from http import HTTPStatus
 10  import dashscope
 11  from dashscope import Generation
 12  from dashscope.api_entities.dashscope_response import Role
 13  from typing import List, Optional, Tuple, Dict
 14  from urllib.error import HTTPError
 15  
 16  default_system = 'You are Qwen, created by Alibaba Cloud. You are a helpful assistant.'
 17  
 18  YOUR_API_TOKEN = os.getenv('YOUR_API_TOKEN')
 19  dashscope.api_key = YOUR_API_TOKEN
 20  
 21  History = List[Tuple[str, str]]
 22  Messages = List[Dict[str, str]]
 23  
 24  def clear_session() -> History:
 25      return '', []
 26  
 27  def modify_system_session(system: str) -> str:
 28      if system is None or len(system) == 0:
 29          system = default_system
 30      return system, system, []
 31  
 32  def history_to_messages(history: History, system: str) -> Messages:
 33      messages = [{'role': Role.SYSTEM, 'content': system}]
 34      for h in history:
 35          messages.append({'role': Role.USER, 'content': h[0]})
 36          messages.append({'role': Role.ASSISTANT, 'content': h[1]})
 37      return messages
 38  
 39  
 40  def messages_to_history(messages: Messages) -> Tuple[str, History]:
 41      assert messages[0]['role'] == Role.SYSTEM
 42      system = messages[0]['content']
 43      history = []
 44      for q, r in zip(messages[1::2], messages[2::2]):
 45          history.append([q['content'], r['content']])
 46      return system, history
 47  
 48  
 49  def model_chat(query: Optional[str], history: Optional[History], system: str
 50  ) -> Tuple[str, str, History]:
 51      if query is None:
 52          query = ''
 53      if history is None:
 54          history = []
 55      messages = history_to_messages(history, system)
 56      messages.append({'role': Role.USER, 'content': query})
 57      gen = Generation.call(
 58          model = "qwen2.5-coder-32b-instruct",
 59          messages=messages,
 60          result_format='message',
 61          stream=True
 62      )
 63      for response in gen:
 64          if response.status_code == HTTPStatus.OK:
 65              role = response.output.choices[0].message.role
 66              response = response.output.choices[0].message.content
 67              system, history = messages_to_history(messages + [{'role': role, 'content': response}])
 68              yield '', history, system
 69          else:
 70              raise HTTPError( code=404, msg='Request id: %s, Status code: %s, error code: %s, error message: %s' % (
 71                  response.request_id, response.status_code,
 72                  response.code, response.message), hdrs=HTTPMessage(), url='http://example.com', fp=None)
 73  
 74  
 75  
 76  
 77  def chiose_radio(radio, system):
 78      mark_ = gr.Markdown(value=f"<center><font size=8>Qwen2.5-Coder-{radio}-instruct👾</center>")
 79      chatbot = gr.Chatbot(label=f'Qwen2.5-Coder-{radio.lower()}-instruct')
 80      
 81      if system is None or len(system) == 0:
 82          system = default_system
 83      
 84      return mark_, chatbot, system, system, ""
 85  
 86  
 87  def update_other_radios(value, other_radio1, other_radio2):
 88      if value == "":
 89          if other_radio1 != "":
 90              selected = other_radio1
 91          else:
 92              selected = other_radio2
 93          return selected, other_radio1, other_radio2
 94      return value, "", ""
 95  
 96  
 97  def main():
 98      # 创建两个标签
 99      with gr.Blocks() as demo:
100          with gr.Row():
101              options_coder = ["0.5B", "1.5B", "3B", "7B", "14B", "32B",]
102              with gr.Row():
103                  radio = gr.Radio(choices=options_coder, label="Qwen2.5-Coder：", value="32B")
104                  
105          with gr.Row():
106              with gr.Accordion():
107                  mark_ = gr.Markdown("""<center><font size=8>Qwen2.5-Coder-32B-Instruct Bot👾</center>""")
108                  with gr.Row():
109                      with gr.Column(scale=3):
110                          system_input = gr.Textbox(value=default_system, lines=1, label='System')
111                      with gr.Column(scale=1):
112                          modify_system = gr.Button("🛠️ Set system prompt and clear history", scale=2)
113                      system_state = gr.Textbox(value=default_system, visible=False)
114                  chatbot = gr.Chatbot(label='Qwen2.5-Coder-32B-Instruct')
115                  textbox = gr.Textbox(lines=1, label='Input')
116                  
117                  with gr.Row():
118                      clear_history = gr.Button("🧹 Clear History")
119                      sumbit = gr.Button("🚀 Send")
120                  
121                  textbox.submit(model_chat,
122                              inputs=[textbox, chatbot, system_state],
123                              outputs=[textbox, chatbot, system_input])
124                  sumbit.click(model_chat,
125                               inputs=[textbox, chatbot, system_state],
126                               outputs=[textbox, chatbot, system_input],
127                               concurrency_limit=100)
128                  clear_history.click(fn=clear_session,
129                                      inputs=[],
130                                      outputs=[textbox, chatbot])
131                  modify_system.click(fn=modify_system_session,
132                                      inputs=[system_input],
133                                      outputs=[system_state, system_input, chatbot])
134          
135          radio.change(chiose_radio,
136                       inputs=[radio, system_input],
137                       outputs=[mark_, chatbot, system_state, system_input, textbox])
138      
139      demo.queue(api_open=False, default_concurrency_limit=40)
140      demo.launch(max_threads=5)
141  
142  
143  if __name__ == "__main__":
144      main()