#### Конечный: import os from http import HTTPStatus import gradio as gr from dashscope import Generation, Role from typing import List, Optional, Tuple, Dict from urllib.error import HTTPError import unittest default_system = 'You are Qwen, created by Alibaba Cloud. You are a helpful assistant.' GROQ_API_KEY = os.environ.get('GROQ_API_KEY') client = OpenAI( api_key=GROQ_API_KEY, base_url="https://api.groq.com/openai/v1", ) History = List[Tuple[str, str]] Messages = List[Dict[str, str]] def clear_session() -> Tuple[str, History]: return '', [] def modify_system_session(system: str) -> Tuple[str, str, History]: if not system: system = default_system return system, system, [] def history_to_messages(history: History, system: str) -> Messages: messages = [{'role': Role.SYSTEM, 'content': system}] for h in history: messages.append({'role': Role.USER, 'content': h[0]}) messages.append({'role': Role.ASSISTANT, 'content': h[1]}) return messages def messages_to_history(messages: Messages) -> Tuple[str, History]: assert messages[0]['role'] == Role.SYSTEM system = messages[0]['content'] history = [] for q, r in zip(messages[1::2], messages[2::2]): history.append([q['content'], r['content']]) return system, history def model_chat(query: Optional[str], history: Optional[History], system: str, radio: str) -> Tuple[str, str, History]: if query is None: query = '' if history is None: history = [] messages = history_to_messages(history, system) messages.append({'role': Role.USER, 'content': query}) label_model = f"qwen2.5-coder-{radio.lower()}-instruct" try: gen = Generation.call( model=label_model, messages=messages, result_format='message', stream=True ) for response in gen: if response.status_code == HTTPStatus.OK: role = response.output.choices[0].message.role response = response.output.choices[0].message.content system, history = messages_to_history(messages + [{'role': role, 'content': response}]) yield '', history, system else: raise HTTPError(code=response.status_code, msg='Request failed with status code: %s' % response.status_code) except HTTPError as e: print(f"HTTP error occurred: {e}") except Exception as e: print(f"An error occurred: {e}") def choose_radio(radio, system): mark_ = gr.Markdown(value=f"
Qwen2.5-Coder-{radio}-instruct👾
") chatbot = gr.Chatbot(label=f'Qwen2.5-Coder-{radio.lower()}-instruct') if not system: system = default_system return mark_, chatbot, system, system, "" def update_other_radios(value, other_radio1, other_radio2): if not value: selected = other_radio1 or other_radio2 return selected, other_radio1, other_radio2 return value, "", "" def main(): # Создание интерфейса Gradio with gr.Blocks() as demo: with gr.Row(): options_coder = ["0.5B", "1.5B", "3B", "7B", "14B", "32B", ] with gr.Row(): radio = gr.Radio(choices=options_coder, label="Qwen2.5-Coder:", value="32B") with gr.Row(): with gr.Accordion(): mark_ = gr.Markdown("""
Qwen2.5-Coder-32B-Instruct Bot👾
""") with gr.Row(): with gr.Column(scale=3): system_input = gr.Textbox(value=default_system, lines=1, label='System') with gr.Column(scale=1): modify_system = gr.Button("🛠️ Set system prompt and clear history", scale=2) system_state = gr.Textbox(value=default_system, visible=False) chatbot = gr.Chatbot(label='Qwen2.5-Coder-32B-Instruct') textbox = gr.Textbox(lines=1, label='Input') with gr.Row(): clear_history = gr.Button("🧹 Clear History") sumbit = gr.Button("🚀 Send") textbox.submit(model_chat, inputs=[textbox, chatbot, system_state, radio], outputs=[textbox, chatbot, system_input]) sumbit.click(model_chat, inputs=[textbox, chatbot, system_state, radio], outputs=[textbox, chatbot, system_input], concurrency_limit=100) clear_history.click(fn=clear_session, inputs=[], outputs=[textbox, chatbot]) modify_system.click(fn=modify_system_session, inputs=[system_input], outputs=[system_state, system_input, chatbot]) radio.change(choose_radio, inputs=[radio, system_input], outputs=[mark_, chatbot, system_state, system_input, textbox]) demo.queue(api_open=False, default_concurrency_limit=40) demo.launch(max_threads=5) if __name__ == "__main__": main() # Тесты class TestApp(unittest.TestCase): def test_clear_session(self): self.assertEqual(clear_session(), ('', [])) def test_modify_system_session(self): self.assertEqual(modify_system_session(None), (default_system, default_system, [])) self.assertEqual(modify_system_session(""), (default_system, default_system, [])) self.assertEqual(modify_system_session("Custom System"), ("Custom System", "Custom System", [])) def test_update_other_radios(self): self.assertEqual(update_other_radios("", "32B", ""), ("32B", "32B", "")) self.assertEqual(update_other_radios("", "", "14B"), ("14B", "", "14B")) self.assertEqual(update_other_radios("7B", "", ""), ("7B", "", "")) if __name__ == "__main__": unittest.main()