update web app cookbook

2025-05-26 19:16:17 -04:00
parent aaf69731ee
commit 92ccbea299
17 changed files with 660 additions and 297 deletions
@@ -0,0 +1,57 @@
+# PocketFlow Streamlit Image Generation HITL
+
+Human-in-the-Loop (HITL) image generation application using PocketFlow and Streamlit. Enter text prompts, generate images with OpenAI, and approve/regenerate results.
+
+<p align="center">
+  <img 
+    src="./assets/banner.png" width="800"
+  />
+</p>
+
+## Features
+
+-   **Image Generation:** Uses OpenAI's `gpt-image-1` model to generate images from text prompts
+-   **Human Review:** Interactive interface to approve or regenerate images
+-   **State Machine:** Clean state-based workflow (`initial_input` → `user_feedback` → `final`)
+-   **PocketFlow Integration:** Uses PocketFlow `Node` and `Flow` for image generation with built-in retries
+-   **Session State Management:** Streamlit session state acts as PocketFlow's shared store
+-   **In-Memory Images:** Images stored as base64 strings, no disk storage required
+
+## How to Run
+
+1.  **Set OpenAI API Key:**
+    ```bash
+    export OPENAI_API_KEY="your-openai-api-key"
+    ```
+
+2.  **Install Dependencies:**
+    ```bash
+    pip install -r requirements.txt
+    ```
+
+3.  **Run the Streamlit Application:**
+    ```bash
+    streamlit run app.py
+    ```
+
+4.  **Access the Web UI:**
+    Open the URL provided by Streamlit (usually `http://localhost:8501`).
+
+## Usage
+
+1. **Enter Prompt**: Describe the image you want to generate
+2. **Generate**: Click "Generate Image" to create the image
+3. **Review**: View the generated image and choose:
+   - **Approve**: Accept the image and move to final result
+   - **Regenerate**: Generate a new image with the same prompt
+4. **Final**: View approved image and optionally start over
+
+## Files
+
+-   [`app.py`](./app.py): Main Streamlit application with state-based UI
+-   [`nodes.py`](./nodes.py): PocketFlow `GenerateImageNode` definition
+-   [`flow.py`](./flow.py): PocketFlow `Flow` for image generation
+-   [`utils/generate_image.py`](./utils/generate_image.py): OpenAI image generation utility
+-   [`requirements.txt`](./requirements.txt): Project dependencies
+-   [`docs/design.md`](./docs/design.md): System design documentation
+-   [`README.md`](./README.md): This file
@@ -0,0 +1,85 @@
+import streamlit as st
+import base64
+from flow import create_generation_flow
+
+st.title("PocketFlow Image Generation HITL")
+
+# Initialize session state for shared store
+if 'stage' not in st.session_state:
+    st.session_state.stage = "initial_input"
+    st.session_state.task_input = ""
+    st.session_state.generated_image = ""
+    st.session_state.final_result = ""
+    st.session_state.error_message = ""
+
+# Debug info
+with st.expander("Session State"):
+    st.json({k: v for k, v in st.session_state.items() if not k.startswith("_")})
+
+# State-based UI
+if st.session_state.stage == "initial_input":
+    st.header("1. Generate Image")
+    
+    prompt = st.text_area("Enter image prompt:", value=st.session_state.task_input, height=100)
+    
+    if st.button("Generate Image"):
+        if prompt.strip():
+            st.session_state.task_input = prompt
+            st.session_state.error_message = ""
+            
+            try:
+                with st.spinner("Generating image..."):
+                    flow = create_generation_flow()
+                    flow.run(st.session_state)
+                st.rerun()
+            except Exception as e:
+                st.session_state.error_message = str(e)
+        else:
+            st.error("Please enter a prompt")
+
+elif st.session_state.stage == "user_feedback":
+    st.header("2. Review Generated Image")
+    
+    if st.session_state.generated_image:
+        # Display image
+        image_bytes = base64.b64decode(st.session_state.generated_image)
+        st.image(image_bytes, caption=f"Prompt: {st.session_state.task_input}")
+        
+        col1, col2 = st.columns(2)
+        
+        with col1:
+            if st.button("Approve", use_container_width=True):
+                st.session_state.final_result = st.session_state.generated_image
+                st.session_state.stage = "final"
+                st.rerun()
+        
+        with col2:
+            if st.button("Regenerate", use_container_width=True):
+                try:
+                    with st.spinner("Regenerating image..."):
+                        flow = create_generation_flow()
+                        flow.run(st.session_state)
+                    st.rerun()
+                except Exception as e:
+                    st.session_state.error_message = str(e)
+
+elif st.session_state.stage == "final":
+    st.header("3. Final Result")
+    st.success("Image approved!")
+    
+    if st.session_state.final_result:
+        image_bytes = base64.b64decode(st.session_state.final_result)
+        st.image(image_bytes, caption=f"Final approved image: {st.session_state.task_input}")
+    
+    if st.button("Start Over", use_container_width=True):
+        st.session_state.stage = "initial_input"
+        st.session_state.task_input = ""
+        st.session_state.generated_image = ""
+        st.session_state.final_result = ""
+        st.session_state.error_message = ""
+        st.rerun()
+
+# Show errors
+if st.session_state.error_message:
+    st.error(st.session_state.error_message)
+
@@ -0,0 +1,105 @@
+# Design Doc: PocketFlow Streamlit Image Generation HITL
+
+> Human-in-the-Loop image generation application using PocketFlow and Streamlit
+
+## Requirements
+
+**User Story**: As a user, I want to:
+1. Enter a text prompt describing an image I want to generate
+2. Have the system generate an image based on my prompt using OpenAI's image generation API
+3. Review the generated image in the web interface
+4. Approve the image if I'm satisfied, OR regenerate with the same prompt if I want a different result
+5. See the final approved image as the completed result
+
+**Technical Requirements**:
+- Use OpenAI's image generation API (via responses.create with image_generation tool)
+- Keep generated images in memory (base64 format) - no disk storage
+- Provide clear approve/regenerate workflow
+- Handle API errors gracefully with retries
+- Maintain session state between generations
+
+## Flow Design
+
+### Applicable Design Pattern:
+
+**State Machine with Multiple Subflows**: Each state has its own user interface and workflow. Users interact with different UI elements in each state, and the app transitions to the next state based on user actions and feedback.
+
+### States & User Interface:
+
+1. **initial_input** - User sees text input field, enters prompt, clicks "Generate Image" button
+2. **user_feedback** - User sees generated image, has "Approve" and "Regenerate" buttons 
+3. **final** - User sees final approved image and "Start Over" button
+
+### Flow High-level Design & Transitions:
+
+```mermaid
+flowchart TD
+    Start([Start]) --> IS[initial_input]
+    IS --> GI[GenerateImage]
+    GI --> UF[user_feedback]
+    UF -->|Regenerate| GI
+    UF -->|Approve| F[final]
+    F --> IS
+    
+    %% Legend
+    classDef stateStyle fill:#e1f5fe,stroke:#01579b,stroke-width:2px
+    classDef nodeStyle fill:#fff3e0,stroke:#e65100,stroke-width:2px
+    
+    class IS,UF,F stateStyle
+    class GI nodeStyle
+```
+
+**Legend:**
+- 🔷 **Blue rectangles**: User interface states (initial_input, user_feedback, final)
+- 🔶 **Orange rectangles**: PocketFlow processing nodes (GenerateImage)
+
+## Utility Functions
+
+1. **Generate Image** (`utils/generate_image.py`)
+   - *Input*: prompt (str)
+   - *Output*: base64 image data (str)
+   - *Purpose*: Calls OpenAI's image generation API and returns base64 encoded image
+   - *Error Handling*: Includes retry logic for API failures
+
+## Node Design
+
+### Shared Memory
+
+**Using Streamlit Session State as Shared Store**: We use `st.session_state` directly as the shared store for PocketFlow, eliminating the need for separate data structures.
+
+The session state structure for the image generation workflow:
+
+```python
+st.session_state = {
+    # User input and workflow state
+    "task_input": "user's text prompt for image generation",
+    "stage": "current workflow stage (initial_input/user_feedback/final)",
+    "error_message": "any error messages for user feedback",
+    
+    # Processing data
+    "input_used_by_process": "prompt used for generation",
+    "generated_image": "base64 encoded image data",
+    "final_result": "final approved image data",
+    
+    # Streamlit built-in keys (managed automatically)
+    # "_streamlit_*": various internal streamlit state
+}
+```
+
+### Node Steps
+
+**Initial Input Flow Nodes:**
+
+1. **Image Generation Node**
+   - *Purpose*: Generate image using OpenAI API based on the prompt
+   - *Type*: Regular (with retries for API reliability)
+   - *Steps*:
+     - *prep*: Read "input_used_by_process" from st.session_state
+     - *exec*: Call generate_image utility with the prompt, return base64 image data
+     - *post*: Write base64 image data to "generated_image" in st.session_state
+
+**User Feedback Flow:**
+- Reuses the same `GenerateImage` node when user clicks "Regenerate"
+
+**Final Flow:**
+- No processing nodes needed - the `final` state simply displays the approved image from `generated_image` and provides UI for starting over
@@ -0,0 +1,9 @@
+from pocketflow import Flow
+from nodes import GenerateImageNode
+
+def create_generation_flow():
+    """Creates a flow for image generation (initial or regeneration)."""
+    generate_image_node = GenerateImageNode()
+    return Flow(start=generate_image_node)
+
+
@@ -0,0 +1,17 @@
+from pocketflow import Node
+from utils.generate_image import generate_image
+
+class GenerateImageNode(Node):
+    """Generates image from text prompt using OpenAI API."""
+    
+    def prep(self, shared):
+        return shared.get("task_input", "")
+
+    def exec(self, prompt):
+        return generate_image(prompt)
+
+    def post(self, shared, prep_res, exec_res):
+        shared["input_used_by_process"] = prep_res
+        shared["generated_image"] = exec_res
+        shared["stage"] = "user_feedback"
+        return "default"
@@ -0,0 +1,3 @@
+streamlit
+pocketflow
+openai
@@ -0,0 +1,30 @@
+from openai import OpenAI
+import os
+import base64
+
+def generate_image(prompt: str) -> str:
+    client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
+    
+    response = client.images.generate(
+        model="gpt-image-1",
+        prompt=prompt,
+        n=1,
+        size="1024x1024"
+    )
+    
+    image_b64 = response.data[0].b64_json
+    print(f"Generated image ({len(image_b64)} chars)")
+    return image_b64
+
+if __name__ == "__main__":
+    test_prompt = "A gray tabby cat hugging an otter with an orange scarf"
+    print(f"Generating image for prompt: {test_prompt[:50]}...")
+    
+    image_base64 = generate_image(test_prompt)
+    print(f"Success! Generated {len(image_base64)} characters of base64 data")
+    
+    # Write image to local file for testing
+    image_bytes = base64.b64decode(image_base64)
+    with open("test_generated_image.png", "wb") as f:
+        f.write(image_bytes)
+    print("Test image saved as test_generated_image.png")