| 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) |
