| from typing import List | |
| import openai | |
| prompt_template = "Tell me a short joke about {topic}" | |
| client = openai.OpenAI() | |
| async_client = openai.AsyncOpenAI() | |
| async def acall_chat_model( | |
| messages: List[dict]) -> str: | |
| response = await ( | |
| async_client.chat.completions.create( | |
| model="gpt-3.5-turbo", | |
| messages=messages, | |
| ) | |
| ) | |
| return response.choices[0].message.content | |
| async def ainvoke_chain(topic: str) -> str: | |
| prompt_value = prompt_template.format( | |
| topic=topic | |
| ) | |
| messages = [{"role": "user", | |
| "content": prompt_value}] | |
| return await acall_chat_model(messages) | |
| if __name__ == "__main__": | |
| import asyncio | |
| loop = asyncio.get_event_loop() | |
| result = loop.run_until_complete( | |
| ainvoke_chain("ice cream") | |
| ) | |
| print(result) |