add web hil
This commit is contained in:
zachary62
parent
9de04109d8
commit
89008a04a1
|
|
@ -83,6 +83,7 @@ From there, it's easy to implement popular design patterns like ([Multi-](https:
|
||||||
| [Thinking](https://github.com/The-Pocket/PocketFlow/tree/main/cookbook/pocketflow-thinking) | ★☆☆ <br> *Beginner* | Solve complex reasoning problems through Chain-of-Thought |
|
| [Thinking](https://github.com/The-Pocket/PocketFlow/tree/main/cookbook/pocketflow-thinking) | ★☆☆ <br> *Beginner* | Solve complex reasoning problems through Chain-of-Thought |
|
||||||
| [Memory](https://github.com/The-Pocket/PocketFlow/tree/main/cookbook/pocketflow-chat-memory) | ★☆☆ <br> *Beginner* | A chat bot with short-term and long-term memory |
|
| [Memory](https://github.com/The-Pocket/PocketFlow/tree/main/cookbook/pocketflow-chat-memory) | ★☆☆ <br> *Beginner* | A chat bot with short-term and long-term memory |
|
||||||
| [MCP](https://github.com/The-Pocket/PocketFlow/tree/main/cookbook/pocketflow-mcp) | ★☆☆ <br> *Beginner* | Agent using Model Context Protocol for numerical operations |
|
| [MCP](https://github.com/The-Pocket/PocketFlow/tree/main/cookbook/pocketflow-mcp) | ★☆☆ <br> *Beginner* | Agent using Model Context Protocol for numerical operations |
|
||||||
|
| [Web HITL](https://github.com/The-Pocket/PocketFlow/tree/main/cookbook/pocketflow-web-hitl) | ★☆☆ <br> *Beginner* | A minimal web service for a human review loop with SSE updates |
|
||||||
|
|
||||||
</div>
|
</div>
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -1,4 +1,4 @@
|
||||||
pocketflow>=0.0.5
|
pocketflow>=0.0.1
|
||||||
numpy>=1.20.0
|
numpy>=1.20.0
|
||||||
faiss-cpu>=1.7.0
|
faiss-cpu>=1.7.0
|
||||||
openai>=1.0.0
|
openai>=1.0.0
|
||||||
|
|
@ -1,42 +1,82 @@
|
||||||
# PocketFlow Hello World
|
# PocketFlow Web Human-in-the-Loop (HITL) Feedback Service
|
||||||
|
|
||||||
Your first PocketFlow application! This simple example demonstrates how to create a basic PocketFlow app from scratch.
|
This project demonstrates a minimal web application for human-in-the-loop workflows using PocketFlow, FastAPI, and Server-Sent Events (SSE). Users can submit text, have it processed (simulated), review the output, and approve or reject it, potentially triggering reprocessing until approved.
|
||||||
|
|
||||||
## Project Structure
|
<p align="center">
|
||||||
|
<img
|
||||||
|
src="./assets/banner.png" width="800"
|
||||||
|
/>
|
||||||
|
</p>
|
||||||
|
|
||||||
```
|
## Features
|
||||||
.
|
|
||||||
├── docs/ # Documentation files
|
- **Web UI:** Simple interface for submitting tasks and providing feedback.
|
||||||
├── utils/ # Utility functions
|
- **PocketFlow Workflow:** Manages the process -> review -> result/reprocess logic.
|
||||||
├── flow.py # PocketFlow implementation
|
- **FastAPI Backend:** Serves the UI and handles API requests asynchronously.
|
||||||
├── main.py # Main application entry point
|
- **Server-Sent Events (SSE):** Provides real-time status updates to the client without polling.
|
||||||
└── README.md # Project documentation
|
|
||||||
|
## How to Run
|
||||||
|
|
||||||
|
1. Install Dependencies:
|
||||||
|
```bash
|
||||||
|
pip install -r requirements.txt
|
||||||
|
```
|
||||||
|
|
||||||
|
2. Run the FastAPI Server:
|
||||||
|
Use Uvicorn (or another ASGI server):
|
||||||
|
```bash
|
||||||
|
uvicorn server:app --reload --port 8000
|
||||||
|
```
|
||||||
|
*(The `--reload` flag is useful for development.)*
|
||||||
|
|
||||||
|
3. Access the Web UI:
|
||||||
|
Open your web browser and navigate to `http://127.0.0.1:8000`.
|
||||||
|
|
||||||
|
4. Use the Application:
|
||||||
|
* Enter text into the textarea and click "Submit".
|
||||||
|
* Observe the status updates pushed via SSE.
|
||||||
|
* When prompted ("waiting_for_review"), use the "Approve" or "Reject" buttons.
|
||||||
|
* If rejected, the process loops back. If approved, the final result is displayed.
|
||||||
|
|
||||||
|
## How It Works
|
||||||
|
|
||||||
|
The application uses PocketFlow to define and execute the feedback loop workflow. FastAPI handles web requests and manages the real-time SSE communication.
|
||||||
|
|
||||||
|
**PocketFlow Workflow:**
|
||||||
|
|
||||||
|
The core logic is orchestrated by an `AsyncFlow` defined in `flow.py`:
|
||||||
|
|
||||||
|
```mermaid
|
||||||
|
flowchart TD
|
||||||
|
subgraph FeedbackFlow[MinimalFeedbackFlow]
|
||||||
|
Process[ProcessNode] -- default --> Review[ReviewNode]
|
||||||
|
Review -- approved --> Result[ResultNode]
|
||||||
|
Review -- rejected --> Process
|
||||||
|
end
|
||||||
```
|
```
|
||||||
|
|
||||||
## Setup
|
1. **`ProcessNode`**: Receives input text, calls the minimal `process_task` utility, and stores the output.
|
||||||
|
2. **`ReviewNode` (Async)**:
|
||||||
|
* Pushes a "waiting_for_review" status with the processed output to the SSE queue.
|
||||||
|
* Waits asynchronously for an external signal (triggered by the `/feedback` API endpoint).
|
||||||
|
* Based on the received feedback ("approved" or "rejected"), determines the next step in the flow. Stores the result if approved.
|
||||||
|
3. **`ResultNode`**: Logs the final approved result.
|
||||||
|
|
||||||
1. Create a virtual environment:
|
**FastAPI & SSE Integration:**
|
||||||
```bash
|
|
||||||
python -m venv venv
|
|
||||||
source venv/bin/activate # On Windows: venv\Scripts\activate
|
|
||||||
```
|
|
||||||
|
|
||||||
2. Install dependencies:
|
* The `/submit` endpoint creates a unique task, initializes the PocketFlow `shared` state (including an `asyncio.Event` for review and an `asyncio.Queue` for SSE), and schedules the flow execution using `BackgroundTasks`.
|
||||||
```bash
|
* Nodes within the flow (specifically `ReviewNode`'s prep logic) put status updates onto the task-specific `sse_queue`.
|
||||||
pip install -r requirements.txt
|
* The `/stream/{task_id}` endpoint uses `StreamingResponse` to read from the task's `sse_queue` and push formatted status updates to the connected client via Server-Sent Events.
|
||||||
```
|
* The `/feedback/{task_id}` endpoint receives the human's decision, updates the `shared` state, and sets the `asyncio.Event` to unblock the waiting `ReviewNode`.
|
||||||
|
|
||||||
3. Run the example:
|
This setup allows for a decoupled workflow logic (PocketFlow) and web interaction layer (FastAPI), with efficient real-time updates pushed to the user.
|
||||||
```bash
|
|
||||||
python main.py
|
|
||||||
```
|
|
||||||
|
|
||||||
## What This Example Demonstrates
|
## Files
|
||||||
|
|
||||||
- How to create your first PocketFlow application
|
- [`server.py`](./server.py): The main FastAPI application handling HTTP requests, SSE, state management, and background task scheduling.
|
||||||
- Basic PocketFlow concepts and usage
|
- [`nodes.py`](./nodes.py): Defines the PocketFlow `Node` classes (`ProcessNode`, `ReviewNode`, `ResultNode`) for the workflow steps.
|
||||||
- Simple example of PocketFlow's capabilities
|
- [`flow.py`](./flow.py): Defines the PocketFlow `AsyncFlow` that connects the nodes into the feedback loop.
|
||||||
|
- [`utils/process_task.py`](./utils/process_task.py): Contains the minimal simulation function for task processing.
|
||||||
## Additional Resources
|
- [`templates/index.html`](./templates/index.html): The HTML structure for the frontend user interface.
|
||||||
|
- [`static/style.css`](./static/style.css): Basic CSS for styling the frontend.
|
||||||
- [PocketFlow Documentation](https://the-pocket.github.io/PocketFlow/)
|
- [`requirements.txt`](./requirements.txt): Project dependencies (FastAPI, Uvicorn, Jinja2, PocketFlow).
|
||||||
|
|
|
||||||
Binary file not shown.
|
After Width: | Height: | Size: 447 KiB |
|
|
@ -4,7 +4,7 @@
|
||||||
|
|
||||||
* **Goal:** Create a web service for task submission, processing, human review (Approve/Reject loop via UI), and finalization.
|
* **Goal:** Create a web service for task submission, processing, human review (Approve/Reject loop via UI), and finalization.
|
||||||
* **Interface:** Simple web UI (HTML/JS) for input, status display, and feedback buttons.
|
* **Interface:** Simple web UI (HTML/JS) for input, status display, and feedback buttons.
|
||||||
* **Backend:** Flask application using PocketFlow for workflow management.
|
* **Backend:** FastAPI using PocketFlow for workflow management.
|
||||||
* **Real-time Updates:** Use Server-Sent Events (SSE) to push status changes (pending, running, waiting_for_review, completed, failed) and intermediate results to the client without page reloads.
|
* **Real-time Updates:** Use Server-Sent Events (SSE) to push status changes (pending, running, waiting_for_review, completed, failed) and intermediate results to the client without page reloads.
|
||||||
* **State:** Use in-memory storage for task state (Warning: Not suitable for production).
|
* **State:** Use in-memory storage for task state (Warning: Not suitable for production).
|
||||||
|
|
||||||
|
|
@ -35,7 +35,7 @@ flowchart TD
|
||||||
|
|
||||||
## 3. Utilities
|
## 3. Utilities
|
||||||
|
|
||||||
For this specific example, the core "utility" is the processing logic itself. Let's simulate it with a simple function. The Flask server acts as the external interface.
|
For this specific example, the core "utility" is the processing logic itself. Let's simulate it with a simple function. The FastAPI server acts as the external interface.
|
||||||
|
|
||||||
* `process_task(input_data)`: A placeholder function. In a real scenario, this might call an LLM (`utils/call_llm.py`).
|
* `process_task(input_data)`: A placeholder function. In a real scenario, this might call an LLM (`utils/call_llm.py`).
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -1,176 +0,0 @@
|
||||||
import asyncio
|
|
||||||
import uuid
|
|
||||||
import json
|
|
||||||
import os
|
|
||||||
from flask import Flask, request, jsonify, render_template, send_from_directory, Response
|
|
||||||
from flow import create_feedback_flow
|
|
||||||
|
|
||||||
# --- Configuration ---
|
|
||||||
template_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), 'templates'))
|
|
||||||
static_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), 'static'))
|
|
||||||
app = Flask(__name__, template_folder=template_dir, static_folder=static_dir)
|
|
||||||
|
|
||||||
# --- State Management (In-Memory - NOT FOR PRODUCTION) ---
|
|
||||||
tasks = {} # task_id -> {"shared": dict, "status": str, "task_obj": asyncio.Task}
|
|
||||||
|
|
||||||
# --- Background Flow Runner ---
|
|
||||||
async def run_flow_background(task_id, flow, shared):
|
|
||||||
"""Runs the flow in background, uses queue in shared for SSE."""
|
|
||||||
if task_id not in tasks: return # Should not happen
|
|
||||||
queue = shared.get("sse_queue")
|
|
||||||
if not queue:
|
|
||||||
print(f"ERROR: Task {task_id} missing sse_queue in shared store!")
|
|
||||||
tasks[task_id]["status"] = "failed"
|
|
||||||
# Cannot easily report via SSE if queue is missing
|
|
||||||
return
|
|
||||||
|
|
||||||
tasks[task_id]["status"] = "running"
|
|
||||||
await queue.put({"status": "running"})
|
|
||||||
print(f"Task {task_id}: Flow starting.")
|
|
||||||
|
|
||||||
final_status = "unknown"
|
|
||||||
error_message = None
|
|
||||||
try:
|
|
||||||
await flow.run_async(shared)
|
|
||||||
# Check final state
|
|
||||||
if shared.get("final_result") is not None:
|
|
||||||
final_status = "completed"
|
|
||||||
else:
|
|
||||||
# If flow ends without setting final_result (e.g., error before ResultNode)
|
|
||||||
final_status = "finished_incomplete"
|
|
||||||
print(f"Task {task_id}: Flow finished with status: {final_status}")
|
|
||||||
|
|
||||||
except Exception as e:
|
|
||||||
final_status = "failed"
|
|
||||||
error_message = str(e)
|
|
||||||
print(f"Task {task_id}: Flow failed: {e}")
|
|
||||||
finally:
|
|
||||||
if task_id in tasks:
|
|
||||||
tasks[task_id]["status"] = final_status
|
|
||||||
final_update = {"status": final_status}
|
|
||||||
if final_status == "completed":
|
|
||||||
final_update["final_result"] = shared.get("final_result")
|
|
||||||
elif error_message:
|
|
||||||
final_update["error"] = error_message
|
|
||||||
await queue.put(final_update)
|
|
||||||
# Signal end of stream
|
|
||||||
await queue.put(None)
|
|
||||||
print(f"Task {task_id}: Background task ended. Final update put on queue.")
|
|
||||||
|
|
||||||
# --- Flask Routes ---
|
|
||||||
@app.route('/')
|
|
||||||
async def index():
|
|
||||||
return render_template('index.html')
|
|
||||||
|
|
||||||
@app.route('/static/<path:filename>')
|
|
||||||
async def static_files(filename):
|
|
||||||
return send_from_directory(app.static_folder, filename)
|
|
||||||
|
|
||||||
@app.route('/submit', methods=['POST'])
|
|
||||||
async def submit_task():
|
|
||||||
if not request.is_json or 'data' not in request.json:
|
|
||||||
return jsonify({"error": "Requires JSON with 'data' field"}), 400
|
|
||||||
|
|
||||||
task_id = str(uuid.uuid4())
|
|
||||||
feedback_event = asyncio.Event()
|
|
||||||
status_queue = asyncio.Queue() # Queue for SSE
|
|
||||||
|
|
||||||
# Initial shared state for the flow
|
|
||||||
shared = {
|
|
||||||
"task_input": request.json['data'],
|
|
||||||
"processed_output": None,
|
|
||||||
"feedback": None,
|
|
||||||
"review_event": feedback_event,
|
|
||||||
"sse_queue": status_queue, # Make queue accessible to nodes
|
|
||||||
"final_result": None,
|
|
||||||
"task_id": task_id
|
|
||||||
}
|
|
||||||
|
|
||||||
flow = create_feedback_flow()
|
|
||||||
|
|
||||||
# Store task state
|
|
||||||
tasks[task_id] = {
|
|
||||||
"shared": shared,
|
|
||||||
"status": "pending",
|
|
||||||
# "flow": flow, # Not strictly needed if we don't re-use it
|
|
||||||
"task_obj": None # Will hold the background task
|
|
||||||
}
|
|
||||||
|
|
||||||
await status_queue.put({"status": "pending", "task_id": task_id})
|
|
||||||
|
|
||||||
# Start flow execution in background
|
|
||||||
task_obj = asyncio.create_task(run_flow_background(task_id, flow, shared))
|
|
||||||
tasks[task_id]["task_obj"] = task_obj
|
|
||||||
|
|
||||||
print(f"Task {task_id}: Submitted.")
|
|
||||||
return jsonify({"message": "Task submitted", "task_id": task_id}), 202
|
|
||||||
|
|
||||||
@app.route('/feedback/<task_id>', methods=['POST'])
|
|
||||||
async def provide_feedback(task_id):
|
|
||||||
if task_id not in tasks:
|
|
||||||
return jsonify({"error": "Task not found"}), 404
|
|
||||||
|
|
||||||
task_info = tasks[task_id]
|
|
||||||
shared = task_info["shared"]
|
|
||||||
queue = shared.get("sse_queue")
|
|
||||||
|
|
||||||
async def report_error(message, status_code=400):
|
|
||||||
print(f"Task {task_id}: Feedback error - {message}")
|
|
||||||
if queue: await queue.put({"status": "feedback_error", "error": message})
|
|
||||||
return jsonify({"error": message}), status_code
|
|
||||||
|
|
||||||
if not request.is_json or 'feedback' not in request.json:
|
|
||||||
return await report_error("Requires JSON with 'feedback' field")
|
|
||||||
feedback = request.json.get('feedback')
|
|
||||||
if feedback not in ["approved", "rejected"]:
|
|
||||||
return await report_error("Invalid feedback value")
|
|
||||||
|
|
||||||
review_event = shared.get("review_event")
|
|
||||||
if not review_event or review_event.is_set():
|
|
||||||
return await report_error("Task not awaiting feedback or feedback already sent", 409)
|
|
||||||
|
|
||||||
print(f"Task {task_id}: Received feedback: {feedback}")
|
|
||||||
if queue: await queue.put({"status": "processing_feedback"})
|
|
||||||
tasks[task_id]["status"] = "processing_feedback"
|
|
||||||
|
|
||||||
shared["feedback"] = feedback
|
|
||||||
review_event.set() # Signal the waiting ReviewNode
|
|
||||||
|
|
||||||
return jsonify({"message": f"Feedback '{feedback}' received"}), 200
|
|
||||||
|
|
||||||
# --- SSE Endpoint ---
|
|
||||||
@app.route('/stream/<task_id>')
|
|
||||||
async def stream(task_id):
|
|
||||||
if task_id not in tasks or "sse_queue" not in tasks[task_id]["shared"]:
|
|
||||||
return Response("data: {\"status\": \"error\", \"error\": \"Task or queue not found\"}\n\n",
|
|
||||||
mimetype='text/event-stream', status=404)
|
|
||||||
|
|
||||||
queue = tasks[task_id]["shared"]["sse_queue"]
|
|
||||||
|
|
||||||
async def event_generator():
|
|
||||||
print(f"SSE Stream: Client connected for {task_id}")
|
|
||||||
try:
|
|
||||||
while True:
|
|
||||||
update = await queue.get()
|
|
||||||
if update is None: # Sentinel for end of stream
|
|
||||||
print(f"SSE Stream: Sentinel received for {task_id}, closing.")
|
|
||||||
yield f"data: {json.dumps({'status': 'stream_closed'})}\n\n"
|
|
||||||
break
|
|
||||||
sse_data = json.dumps(update)
|
|
||||||
print(f"SSE Stream: Sending for {task_id}: {sse_data}")
|
|
||||||
yield f"data: {sse_data}\n\n"
|
|
||||||
queue.task_done()
|
|
||||||
except asyncio.CancelledError:
|
|
||||||
print(f"SSE Stream: Client disconnected for {task_id}.")
|
|
||||||
finally:
|
|
||||||
print(f"SSE Stream: Generator finished for {task_id}.")
|
|
||||||
# Optional: Cleanup task entry after stream ends?
|
|
||||||
# if task_id in tasks: del tasks[task_id] # Careful if task state needed elsewhere
|
|
||||||
|
|
||||||
headers = {'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache'}
|
|
||||||
return Response(event_generator(), mimetype='text/event-stream', headers=headers)
|
|
||||||
|
|
||||||
# --- Main ---
|
|
||||||
# Use an ASGI server like Hypercorn: `hypercorn server:app`
|
|
||||||
if __name__ == '__main__':
|
|
||||||
print("Run using an ASGI server, e.g., 'hypercorn flask_server:app'")
|
|
||||||
|
|
@ -1,4 +1,3 @@
|
||||||
import asyncio
|
|
||||||
from pocketflow import Node, AsyncNode
|
from pocketflow import Node, AsyncNode
|
||||||
from utils.process_task import process_task
|
from utils.process_task import process_task
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -1,3 +1,4 @@
|
||||||
pocketflow>=0.0.1
|
pocketflow>=0.0.1
|
||||||
flask
|
fastapi
|
||||||
hypercorn
|
uvicorn[standard] # ASGI server for FastAPI
|
||||||
|
jinja2 # For HTML templating
|
||||||
|
|
@ -0,0 +1,253 @@
|
||||||
|
import asyncio
|
||||||
|
import uuid
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
from fastapi import FastAPI, Request, HTTPException, status, BackgroundTasks # Import BackgroundTasks
|
||||||
|
from fastapi.responses import HTMLResponse, StreamingResponse
|
||||||
|
from fastapi.staticfiles import StaticFiles
|
||||||
|
from fastapi.templating import Jinja2Templates
|
||||||
|
from pydantic import BaseModel, Field # Import Pydantic for request/response models
|
||||||
|
from typing import Dict, Any, Literal # For type hinting
|
||||||
|
|
||||||
|
from flow import create_feedback_flow # PocketFlow imports
|
||||||
|
|
||||||
|
# --- Configuration ---
|
||||||
|
app = FastAPI(title="Minimal Feedback Loop API")
|
||||||
|
|
||||||
|
static_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), 'static'))
|
||||||
|
if os.path.isdir(static_dir):
|
||||||
|
app.mount("/static", StaticFiles(directory=static_dir), name="static")
|
||||||
|
else:
|
||||||
|
print(f"Warning: Static directory '{static_dir}' not found.")
|
||||||
|
|
||||||
|
template_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), 'templates'))
|
||||||
|
if os.path.isdir(template_dir):
|
||||||
|
templates = Jinja2Templates(directory=template_dir)
|
||||||
|
else:
|
||||||
|
print(f"Warning: Template directory '{template_dir}' not found.")
|
||||||
|
templates = None
|
||||||
|
|
||||||
|
# --- State Management (In-Memory - NOT FOR PRODUCTION) ---
|
||||||
|
# Global dictionary to store task state. In production, use Redis, DB, etc.
|
||||||
|
tasks: Dict[str, Dict[str, Any]] = {}
|
||||||
|
# Structure: task_id -> {"shared": dict, "status": str, "task_obj": asyncio.Task | None}
|
||||||
|
|
||||||
|
|
||||||
|
# --- Background Flow Runner ---
|
||||||
|
# This function remains mostly the same, as it defines the work to be done.
|
||||||
|
# It will be scheduled by FastAPI's BackgroundTasks now.
|
||||||
|
async def run_flow_background(task_id: str, flow, shared: Dict[str, Any]):
|
||||||
|
"""Runs the flow in background, uses queue in shared for SSE."""
|
||||||
|
# Check if task exists (might have been cancelled/deleted)
|
||||||
|
if task_id not in tasks:
|
||||||
|
print(f"Background task {task_id}: Task not found, aborting.")
|
||||||
|
return
|
||||||
|
queue = shared.get("sse_queue")
|
||||||
|
if not queue:
|
||||||
|
print(f"ERROR: Task {task_id} missing sse_queue in shared store!")
|
||||||
|
tasks[task_id]["status"] = "failed"
|
||||||
|
# Cannot report failure via SSE if queue is missing
|
||||||
|
return
|
||||||
|
|
||||||
|
tasks[task_id]["status"] = "running"
|
||||||
|
await queue.put({"status": "running"})
|
||||||
|
print(f"Task {task_id}: Background flow starting.")
|
||||||
|
|
||||||
|
final_status = "unknown"
|
||||||
|
error_message = None
|
||||||
|
try:
|
||||||
|
# Execute the potentially long-running PocketFlow
|
||||||
|
await flow.run_async(shared)
|
||||||
|
|
||||||
|
# Determine final status based on shared state after flow completion
|
||||||
|
if shared.get("final_result") is not None:
|
||||||
|
final_status = "completed"
|
||||||
|
else:
|
||||||
|
# If flow ends without setting final_result
|
||||||
|
final_status = "finished_incomplete"
|
||||||
|
print(f"Task {task_id}: Flow finished with status: {final_status}")
|
||||||
|
|
||||||
|
except Exception as e:
|
||||||
|
final_status = "failed"
|
||||||
|
error_message = str(e)
|
||||||
|
print(f"Task {task_id}: Flow execution failed: {e}")
|
||||||
|
# Consider logging traceback here in production
|
||||||
|
finally:
|
||||||
|
# Ensure task still exists before updating state
|
||||||
|
if task_id in tasks:
|
||||||
|
tasks[task_id]["status"] = final_status
|
||||||
|
final_update = {"status": final_status}
|
||||||
|
if final_status == "completed":
|
||||||
|
final_update["final_result"] = shared.get("final_result")
|
||||||
|
elif error_message:
|
||||||
|
final_update["error"] = error_message
|
||||||
|
# Put final status update onto the queue
|
||||||
|
await queue.put(final_update)
|
||||||
|
|
||||||
|
# Signal the end of the SSE stream by putting None
|
||||||
|
# Must happen regardless of whether task was deleted mid-run
|
||||||
|
if queue:
|
||||||
|
await queue.put(None)
|
||||||
|
print(f"Task {task_id}: Background task ended. Final update sentinel put on queue.")
|
||||||
|
# Remove the reference to the completed/failed asyncio Task object
|
||||||
|
if task_id in tasks:
|
||||||
|
tasks[task_id]["task_obj"] = None
|
||||||
|
|
||||||
|
# --- Pydantic Models for Request/Response Validation ---
|
||||||
|
class SubmitRequest(BaseModel):
|
||||||
|
data: str = Field(..., min_length=1, description="Input data for the task")
|
||||||
|
|
||||||
|
class SubmitResponse(BaseModel):
|
||||||
|
message: str = "Task submitted"
|
||||||
|
task_id: str
|
||||||
|
|
||||||
|
class FeedbackRequest(BaseModel):
|
||||||
|
feedback: Literal["approved", "rejected"] # Use Literal for specific choices
|
||||||
|
|
||||||
|
class FeedbackResponse(BaseModel):
|
||||||
|
message: str
|
||||||
|
|
||||||
|
# --- FastAPI Routes ---
|
||||||
|
@app.get("/", response_class=HTMLResponse, include_in_schema=False)
|
||||||
|
async def get_index(request: Request):
|
||||||
|
"""Serves the main HTML frontend."""
|
||||||
|
if templates is None:
|
||||||
|
raise HTTPException(status_code=500, detail="Templates directory not configured.")
|
||||||
|
return templates.TemplateResponse("index.html", {"request": request})
|
||||||
|
|
||||||
|
@app.post("/submit", response_model=SubmitResponse, status_code=status.HTTP_202_ACCEPTED)
|
||||||
|
async def submit_task(
|
||||||
|
submit_request: SubmitRequest, # Use Pydantic model for validation
|
||||||
|
background_tasks: BackgroundTasks # Inject BackgroundTasks instance
|
||||||
|
):
|
||||||
|
"""
|
||||||
|
Submits a new task. The actual processing runs in the background.
|
||||||
|
Returns immediately with the task ID.
|
||||||
|
"""
|
||||||
|
task_id = str(uuid.uuid4())
|
||||||
|
feedback_event = asyncio.Event()
|
||||||
|
status_queue = asyncio.Queue()
|
||||||
|
|
||||||
|
shared = {
|
||||||
|
"task_input": submit_request.data,
|
||||||
|
"processed_output": None,
|
||||||
|
"feedback": None,
|
||||||
|
"review_event": feedback_event,
|
||||||
|
"sse_queue": status_queue,
|
||||||
|
"final_result": None,
|
||||||
|
"task_id": task_id
|
||||||
|
}
|
||||||
|
|
||||||
|
flow = create_feedback_flow()
|
||||||
|
|
||||||
|
# Store task state BEFORE scheduling background task
|
||||||
|
tasks[task_id] = {
|
||||||
|
"shared": shared,
|
||||||
|
"status": "pending",
|
||||||
|
"task_obj": None # Placeholder for the asyncio Task created by BackgroundTasks
|
||||||
|
}
|
||||||
|
|
||||||
|
await status_queue.put({"status": "pending", "task_id": task_id})
|
||||||
|
|
||||||
|
# Schedule the flow execution using FastAPI's BackgroundTasks
|
||||||
|
# This runs AFTER the response has been sent
|
||||||
|
background_tasks.add_task(run_flow_background, task_id, flow, shared)
|
||||||
|
# Note: We don't get a direct reference to the asyncio Task object this way,
|
||||||
|
# which is fine for this minimal example. If cancellation were needed,
|
||||||
|
# managing asyncio.create_task manually would be necessary.
|
||||||
|
|
||||||
|
print(f"Task {task_id}: Submitted, scheduled for background execution.")
|
||||||
|
return SubmitResponse(task_id=task_id)
|
||||||
|
|
||||||
|
|
||||||
|
@app.post("/feedback/{task_id}", response_model=FeedbackResponse)
|
||||||
|
async def provide_feedback(task_id: str, feedback_request: FeedbackRequest):
|
||||||
|
"""Provides feedback (approved/rejected) to potentially unblock a waiting task."""
|
||||||
|
if task_id not in tasks:
|
||||||
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Task not found")
|
||||||
|
|
||||||
|
task_info = tasks[task_id]
|
||||||
|
shared = task_info["shared"]
|
||||||
|
queue = shared.get("sse_queue")
|
||||||
|
review_event = shared.get("review_event")
|
||||||
|
|
||||||
|
async def report_error(message, status_code=status.HTTP_400_BAD_REQUEST):
|
||||||
|
# Helper to log, put status on queue, and raise HTTP exception
|
||||||
|
print(f"Task {task_id}: Feedback error - {message}")
|
||||||
|
if queue: await queue.put({"status": "feedback_error", "error": message})
|
||||||
|
raise HTTPException(status_code=status_code, detail=message)
|
||||||
|
|
||||||
|
if not review_event:
|
||||||
|
# This indicates an internal setup error if the task exists but has no event
|
||||||
|
await report_error("Task not configured for feedback", status.HTTP_500_INTERNAL_SERVER_ERROR)
|
||||||
|
if review_event.is_set():
|
||||||
|
# Prevent processing feedback multiple times or if the task isn't waiting
|
||||||
|
await report_error("Task not awaiting feedback or feedback already sent", status.HTTP_409_CONFLICT)
|
||||||
|
|
||||||
|
feedback = feedback_request.feedback # Already validated by Pydantic
|
||||||
|
print(f"Task {task_id}: Received feedback via POST: {feedback}")
|
||||||
|
|
||||||
|
# Update status *before* setting the event, so client sees 'processing' first
|
||||||
|
if queue: await queue.put({"status": "processing_feedback", "feedback_value": feedback})
|
||||||
|
tasks[task_id]["status"] = "processing_feedback" # Update central status tracker
|
||||||
|
|
||||||
|
# Store feedback and signal the waiting ReviewNode
|
||||||
|
shared["feedback"] = feedback
|
||||||
|
review_event.set()
|
||||||
|
|
||||||
|
return FeedbackResponse(message=f"Feedback '{feedback}' received")
|
||||||
|
|
||||||
|
|
||||||
|
# --- SSE Endpoint ---
|
||||||
|
@app.get("/stream/{task_id}")
|
||||||
|
async def stream_status(task_id: str):
|
||||||
|
"""Streams status updates for a given task using Server-Sent Events."""
|
||||||
|
if task_id not in tasks or "sse_queue" not in tasks[task_id]["shared"]:
|
||||||
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Task or queue not found")
|
||||||
|
|
||||||
|
queue = tasks[task_id]["shared"]["sse_queue"]
|
||||||
|
|
||||||
|
async def event_generator():
|
||||||
|
"""Yields SSE messages from the task's queue."""
|
||||||
|
print(f"SSE Stream: Client connected for {task_id}")
|
||||||
|
try:
|
||||||
|
while True:
|
||||||
|
# Wait for the next status update from the queue
|
||||||
|
update = await queue.get()
|
||||||
|
if update is None: # Sentinel value indicates end of stream
|
||||||
|
print(f"SSE Stream: Sentinel received for {task_id}, closing stream.")
|
||||||
|
yield f"data: {json.dumps({'status': 'stream_closed'})}\n\n"
|
||||||
|
break
|
||||||
|
|
||||||
|
sse_data = json.dumps(update)
|
||||||
|
print(f"SSE Stream: Sending for {task_id}: {sse_data}")
|
||||||
|
yield f"data: {sse_data}\n\n" # SSE format: "data: <json>\n\n"
|
||||||
|
queue.task_done() # Acknowledge processing the queue item
|
||||||
|
|
||||||
|
except asyncio.CancelledError:
|
||||||
|
# This happens if the client disconnects
|
||||||
|
print(f"SSE Stream: Client disconnected for {task_id}.")
|
||||||
|
except Exception as e:
|
||||||
|
# Log unexpected errors during streaming
|
||||||
|
print(f"SSE Stream: Error in generator for {task_id}: {e}")
|
||||||
|
# Optionally send an error message to the client if possible
|
||||||
|
try:
|
||||||
|
yield f"data: {json.dumps({'status': 'stream_error', 'error': str(e)})}\n\n"
|
||||||
|
except Exception: # Catch errors if yield fails (e.g., connection already closed)
|
||||||
|
pass
|
||||||
|
finally:
|
||||||
|
print(f"SSE Stream: Generator finished for {task_id}.")
|
||||||
|
# Consider cleanup here (e.g., removing task if no longer needed)
|
||||||
|
# if task_id in tasks: del tasks[task_id]
|
||||||
|
|
||||||
|
# Use FastAPI/Starlette's StreamingResponse for SSE
|
||||||
|
headers = {'Cache-Control': 'no-cache', 'X-Accel-Buffering': 'no'}
|
||||||
|
return StreamingResponse(event_generator(), media_type="text/event-stream", headers=headers)
|
||||||
|
|
||||||
|
# --- Main Execution Guard (for running with uvicorn) ---
|
||||||
|
if __name__ == "__main__":
|
||||||
|
print("Starting FastAPI server using Uvicorn is recommended:")
|
||||||
|
print("uvicorn server:app --reload --host 0.0.0.0 --port 8000")
|
||||||
|
# Example using uvicorn programmatically (less common than CLI)
|
||||||
|
# import uvicorn
|
||||||
|
# uvicorn.run(app, host="0.0.0.0", port=8000)
|
||||||
|
|
@ -1,13 +1,137 @@
|
||||||
/* Minimal functional styling */
|
body {
|
||||||
body { font-family: sans-serif; margin: 15px; background-color: #fdfdfd; }
|
font-family: sans-serif;
|
||||||
.container, .status-container { background: #fff; padding: 15px; border: 1px solid #eee; margin-bottom: 15px; border-radius: 4px; max-width: 600px;}
|
margin: 0; /* Remove default body margin */
|
||||||
textarea { width: 95%; padding: 8px; margin-bottom: 10px; border: 1px solid #ccc; font-size: 1em; min-height: 50px; }
|
padding: 20px; /* Add some padding around the content */
|
||||||
button { padding: 8px 12px; margin-right: 5px; cursor: pointer; border: 1px solid #ccc; border-radius: 3px;}
|
background-color: #f8f9fa; /* Lighter grey background */
|
||||||
button:disabled { cursor: not-allowed; opacity: 0.6; }
|
display: flex; /* Enable Flexbox */
|
||||||
#task-id-display { font-size: 0.85em; color: #666; margin-bottom: 5px; }
|
flex-direction: column; /* Stack children vertically */
|
||||||
#status-display { font-weight: bold; margin-bottom: 10px; padding: 8px; background-color: #f0f0f0; border-radius: 3px;}
|
align-items: center; /* Center children horizontally */
|
||||||
.hidden { display: none; }
|
min-height: 100vh; /* Ensure body takes at least full viewport height */
|
||||||
.review-box, .result-box { border: 1px solid #ddd; padding: 10px; margin-top: 10px; background-color: #f9f9f9; }
|
box-sizing: border-box; /* Include padding in height calculation */
|
||||||
pre { background-color: #eee; padding: 10px; border: 1px solid #ddd; white-space: pre-wrap; word-wrap: break-word; max-height: 200px; overflow-y: auto;}
|
}
|
||||||
.approve { background-color: #d4edda; border-color: #c3e6cb; }
|
|
||||||
.reject { background-color: #f8d7da; border-color: #f5c6cb; }
|
h1 {
|
||||||
|
text-align: center; /* Center the main title */
|
||||||
|
color: #343a40;
|
||||||
|
margin-bottom: 25px;
|
||||||
|
}
|
||||||
|
|
||||||
|
/* Style the main containers */
|
||||||
|
.container, .status-container {
|
||||||
|
background: #ffffff;
|
||||||
|
padding: 20px 25px; /* More padding */
|
||||||
|
border: 1px solid #dee2e6; /* Softer border */
|
||||||
|
margin-bottom: 20px;
|
||||||
|
border-radius: 6px; /* Slightly rounder corners */
|
||||||
|
width: 90%; /* Responsive width */
|
||||||
|
max-width: 650px; /* Max width for readability */
|
||||||
|
box-shadow: 0 2px 5px rgba(0,0,0,0.05); /* Subtle shadow */
|
||||||
|
box-sizing: border-box; /* Include padding/border in width */
|
||||||
|
}
|
||||||
|
|
||||||
|
textarea {
|
||||||
|
width: 100%; /* Take full width of parent container */
|
||||||
|
padding: 10px;
|
||||||
|
margin-bottom: 10px;
|
||||||
|
border: 1px solid #ced4da;
|
||||||
|
border-radius: 4px;
|
||||||
|
font-size: 1em;
|
||||||
|
min-height: 60px;
|
||||||
|
box-sizing: border-box;
|
||||||
|
}
|
||||||
|
|
||||||
|
button {
|
||||||
|
padding: 9px 15px; /* Slightly adjusted padding */
|
||||||
|
margin-right: 8px;
|
||||||
|
cursor: pointer;
|
||||||
|
border: none; /* Remove default border */
|
||||||
|
border-radius: 4px;
|
||||||
|
font-weight: 500;
|
||||||
|
transition: background-color 0.2s ease;
|
||||||
|
}
|
||||||
|
|
||||||
|
button:disabled {
|
||||||
|
cursor: not-allowed;
|
||||||
|
opacity: 0.6;
|
||||||
|
}
|
||||||
|
|
||||||
|
/* Specific button styling */
|
||||||
|
#submit-button {
|
||||||
|
background-color: #0d6efd; /* Bootstrap primary blue */
|
||||||
|
color: white;
|
||||||
|
}
|
||||||
|
#submit-button:hover:not(:disabled) {
|
||||||
|
background-color: #0b5ed7;
|
||||||
|
}
|
||||||
|
|
||||||
|
.approve {
|
||||||
|
background-color: #198754; /* Bootstrap success green */
|
||||||
|
color: white;
|
||||||
|
}
|
||||||
|
.approve:hover:not(:disabled) {
|
||||||
|
background-color: #157347;
|
||||||
|
}
|
||||||
|
|
||||||
|
.reject {
|
||||||
|
background-color: #dc3545; /* Bootstrap danger red */
|
||||||
|
color: white;
|
||||||
|
}
|
||||||
|
.reject:hover:not(:disabled) {
|
||||||
|
background-color: #bb2d3b;
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
#task-id-display {
|
||||||
|
font-size: 0.9em;
|
||||||
|
color: #6c757d; /* Bootstrap secondary text color */
|
||||||
|
margin-bottom: 8px;
|
||||||
|
word-wrap: break-word;
|
||||||
|
}
|
||||||
|
|
||||||
|
#status-display {
|
||||||
|
font-weight: bold;
|
||||||
|
margin-bottom: 15px;
|
||||||
|
padding: 10px;
|
||||||
|
background-color: #e9ecef; /* Light grey background */
|
||||||
|
border: 1px solid #dee2e6;
|
||||||
|
border-radius: 4px;
|
||||||
|
color: #495057;
|
||||||
|
}
|
||||||
|
|
||||||
|
.hidden {
|
||||||
|
display: none;
|
||||||
|
}
|
||||||
|
|
||||||
|
/* Review/Result Box Styling */
|
||||||
|
.review-box, .result-box {
|
||||||
|
border: 1px solid #dee2e6;
|
||||||
|
padding: 15px;
|
||||||
|
margin-top: 15px;
|
||||||
|
border-radius: 4px;
|
||||||
|
background-color: #f8f9fa; /* Very light background */
|
||||||
|
}
|
||||||
|
|
||||||
|
h2, h3 {
|
||||||
|
margin-top: 0; /* Remove default top margin */
|
||||||
|
margin-bottom: 15px;
|
||||||
|
color: #495057;
|
||||||
|
}
|
||||||
|
|
||||||
|
h3 {
|
||||||
|
border-bottom: 1px solid #eee;
|
||||||
|
padding-bottom: 8px;
|
||||||
|
}
|
||||||
|
|
||||||
|
pre {
|
||||||
|
background-color: #e9ecef;
|
||||||
|
padding: 12px;
|
||||||
|
border: 1px solid #ced4da;
|
||||||
|
border-radius: 4px;
|
||||||
|
white-space: pre-wrap;
|
||||||
|
word-wrap: break-word;
|
||||||
|
max-height: 250px; /* Adjusted height */
|
||||||
|
overflow-y: auto;
|
||||||
|
font-family: monospace;
|
||||||
|
font-size: 0.95em;
|
||||||
|
color: #212529;
|
||||||
|
}
|
||||||
|
|
@ -3,11 +3,11 @@
|
||||||
<head>
|
<head>
|
||||||
<meta charset="UTF-8">
|
<meta charset="UTF-8">
|
||||||
<meta name="viewport" content="width=device-width, initial-scale=1.0">
|
<meta name="viewport" content="width=device-width, initial-scale=1.0">
|
||||||
<title>Minimal Feedback Task (SSE)</title>
|
<title>Pocket Flow Web Feedback</title>
|
||||||
<link rel="stylesheet" href="{{ url_for('static_files', filename='style.css') }}">
|
<link rel="stylesheet" href="{{ url_for('static', path='style.css') }}">
|
||||||
</head>
|
</head>
|
||||||
<body>
|
<body>
|
||||||
<h1>Minimal Task Submitter (SSE)</h1>
|
<h1>Pocket Flow Web Feedback</h1>
|
||||||
|
|
||||||
<div class="container">
|
<div class="container">
|
||||||
<textarea id="task-input" rows="3" placeholder="Enter text to process..."></textarea>
|
<textarea id="task-input" rows="3" placeholder="Enter text to process..."></textarea>
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue