Due to concurrency limits, complex network environments, and other unforeseen circumstances, our connections may sometimes be interrupted. Typically, these intermittent disruptions donβt last long. We want our services to remain stable even in such cases. Implementing a simple reconnection feature can be achieved with just a few lines of code.
from openai import OpenAIimport timeclient = OpenAI( api_key = "$MOONSHOT_API_KEY", base_url = "https://api.moonshot.ai/v1",)def chat_once(msgs): response = client.chat.completions.create( model = "kimi-k2.5", messages = msgs ) return response.choices[0].message.contentdef chat(input: str, max_attempts: int = 100) -> str: messages = [ {"role": "system", "content": "You are Kimi, an AI assistant provided by Moonshot AI. You are proficient in Chinese and English conversations. You aim to provide users with safe, helpful, and accurate answers. You will refuse to answer any questions related to terrorism, racism, or explicit content. Moonshot AI is a proper noun and should not be translated into other languages."}, ] # We construct the user's latest question as a message (role=user) and append it to the end of the messages list messages.append({ "role": "user", "content": input, }) st_time = time.time() for i in range(max_attempts): print(f"Attempts: {i+1}/{max_attempts}") try: response = chat_once(messages) ed_time = time.time() print("Query Successful!") print(f"Query Time: {ed_time-st_time}") return response except Exception as e: print(e) time.sleep(1) continue print("Query Failed.") returnprint(chat("Hello, please tell me a fairy tale."))
The code above implements a simple reconnection feature, allowing up to 100 retries with a 1-second wait between each attempt. You can adjust these values and the conditions for retries based on your specific needs.