Spaces:
Running
Running
# Placeholder for visualization utilities | |
from openai import AsyncOpenAI | |
from openai.types.images_response import Image, ImagesResponse | |
from typing import Optional | |
import asyncio # For potential timeout, though not strictly required by tests yet | |
import httpx # Added for timeout configuration in generate_dalle_image | |
from langchain_openai import ChatOpenAI # For type hinting generate_mermaid_code | |
from langchain_core.messages import SystemMessage # For generate_mermaid_code | |
import re # For extracting mermaid code | |
# System prompt for Mermaid generation (copied from test file for consistency) | |
MERMAID_SYSTEM_PROMPT_TEMPLATE = """You are an expert in creating Mermaid diagrams. Based on the following text, generate a concise and accurate Mermaid diagram syntax. Only output the Mermaid code block (```mermaid\n... | |
```). Do not include any other explanatory text. If the text cannot be reasonably converted to a diagram, output '// No suitable diagram' as a comment. Text: {text_input} | |
""" | |
async def generate_dalle_image(prompt: str, client: AsyncOpenAI) -> Optional[str]: | |
"""Generates an image using DALL-E 3 and returns the URL.""" | |
try: | |
print(f"Generating DALL-E image for prompt: '{prompt[:100]}...'") | |
# Configure a timeout for the API call | |
# Note: httpx.AsyncClient allows request-specific timeouts. | |
# AsyncOpenAI uses httpx internally. Default timeout for AsyncOpenAI is 60s. | |
# We can either rely on its default or, if we needed finer control over this specific call, | |
# we might need to configure the client upon its instantiation or use a separate client. | |
# For now, let's assume the default client timeout is sufficient, or add a note about it. | |
# Default timeout is 1 minute. For DALL-E, this is usually enough. | |
# For more control, you can pass `timeout=httpx.Timeout(30.0, connect=5.0)` to AsyncOpenAI client init. | |
response = await client.images.generate( | |
prompt=prompt, | |
model="dall-e-3", | |
size="1024x1024", | |
quality="standard", # 'standard' or 'hd'. 'hd' is more detailed but might be slower/costlier. | |
n=1, | |
style="vivid" # 'vivid' (hyper-real and dramatic) or 'natural' (more natural, less hyper-real) | |
) | |
if response.data and response.data[0].url: | |
return response.data[0].url | |
else: | |
print("DALL-E API call succeeded but returned no data or URL.") | |
return None | |
except Exception as e: | |
print(f"An error occurred during DALL-E image generation: {e}") | |
return None | |
async def generate_mermaid_code(text_input: str, llm_client: ChatOpenAI) -> Optional[str]: | |
"""Generates Mermaid diagram code from text using an LLM.""" | |
if not text_input or not llm_client: | |
return None | |
prompt = MERMAID_SYSTEM_PROMPT_TEMPLATE.format(text_input=text_input) | |
messages = [SystemMessage(content=prompt)] | |
try: | |
print(f"Generating Mermaid code for text: '{text_input[:100]}...'") | |
response = await llm_client.ainvoke(messages) | |
content = response.content | |
if "// No suitable diagram" in content: | |
print("LLM indicated no suitable diagram for Mermaid generation.") | |
return None | |
# Extract content within ```mermaid ... ``` block | |
match = re.search(r"```mermaid\n(.*?)\n```", content, re.DOTALL) | |
if match: | |
return match.group(1).strip() | |
else: | |
# If no block found, but also no "// No suitable diagram" comment, | |
# it might be that the LLM failed to follow instructions or returned plain text. | |
# We could return the raw content, or None, or try to sanitize. | |
# For now, if it's not a proper block and not the explicit no-diagram comment, assume failure to follow format. | |
print(f"Mermaid LLM did not return a valid Mermaid block. Raw output: {content[:200]}...") | |
return None | |
except Exception as e: | |
print(f"An error occurred during Mermaid code generation: {e}") | |
return None |