BrainyFlow

https://github.com/zvictor/brainyflow
Admin
BrainyFlow: More AI with less coding 🎤🐤
Tokens:133,369
Snippets:726
Trust Score:8.8
License:MPL-2.0
Update:1 year ago
Show doc for...
Tokens:
Raw
### Install Dependencies and Run Example (Bash)

Source: https://github.com/zvictor/brainyflow/blob/main/cookbook/python-async-basic/README.md

Installs project dependencies using pip and then runs the main Python script. This is a standard way to set up and execute Python projects.

```bash
pip install -r requirements.txt
python main.py
```

--------------------------------

### Caskada Hello World in Python

Source: https://github.com/zvictor/brainyflow/blob/main/docs/cookbook/python.md

Demonstrates how to create a basic Caskada application from scratch. Includes project structure, setup instructions (virtual environment, dependency installation), and how to run the example.

```bash
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

pip install -r requirements.txt

python main.py
```

```python
import caskada

# Placeholder for flow.py content
class HelloWorldFlow(caskada.Flow):
    def __init__(self):
        super().__init__()

    async def run(self, shared):
        print("Hello, Caskada!")

# Placeholder for main.py content
async def main():
    flow = HelloWorldFlow()
    await flow.run({})

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

```

--------------------------------

### Install Caskada with Python

Source: https://github.com/zvictor/brainyflow/blob/main/docs/getting_started.md

Installs the Caskada framework using pip for Python projects. Ensure pip is available and up-to-date before running this command. No external dependencies are required for this installation step.

```bash
pip install caskada
```

--------------------------------

### Install Caskada with TypeScript/Node.js

Source: https://github.com/zvictor/brainyflow/blob/main/docs/getting_started.md

Installs the Caskada framework using npm, pnpm, or yarn for TypeScript/Node.js projects. Ensure your package manager is installed and configured. This command fetches the latest version of the Caskada package.

```bash
npm install caskada # or pnpm/yarn
```

--------------------------------

### Caskada Hello World Example

Source: https://github.com/zvictor/brainyflow/blob/main/cookbook/README.md

Demonstrates the creation of a basic Caskada application from scratch. It covers project structure, setup with virtual environments and dependencies, and running the example.

```bash
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

pip install -r requirements.txt

python main.py
```

```python
# flow.py (Conceptual structure)
from caskada.node import Node

class HelloWorldNode(Node):
    async def exec(self, shared):
        shared['message'] = 'Hello, Caskada!'

# main.py (Conceptual structure)
from caskada.flow import Flow
from flow import HelloWorldNode

async def main():
    flow = Flow([HelloWorldNode()])
    shared = {}
    await flow.run(shared)
    print(shared['message'])

if __name__ == '__main__':
    import asyncio
    asyncio.run(main())
```

--------------------------------

### Install and Run TypeScript Agent

Source: https://github.com/zvictor/brainyflow/blob/main/cookbook/typescript-agent/README.md

These commands demonstrate how to set up and execute the TypeScript agent. It requires installing dependencies and running the agent with a question, or using a default question if none is provided.

```bash
# from Caskada root directory
cd cookbook/typescript-agent

cp .env.example .env # add your API key

npm install
npm run agent -- "this is your question"
```

--------------------------------

### Install Dependencies and Run Brainyflow

Source: https://github.com/zvictor/brainyflow/blob/main/docs/cookbook/python.md

Installs project dependencies from requirements.txt, sets the ANTHROPIC_API_KEY environment variable, and runs the main script for testing or custom problems. The script supports specifying a problem and the number of tries for the majority vote.

```bash
pip install -r requirements.txt
export ANTHROPIC_API_KEY="your-api-key-here"
python main.py
python main.py --problem "Your complex reasoning problem here" --tries 5
```

--------------------------------

### Copy Caskada Python Source Code

Source: https://github.com/zvictor/brainyflow/blob/main/docs/installation.md

Allows for installation by directly copying the Python source code file into your project. This is suitable for projects where external dependencies are restricted.

```python
# Copy python/caskada.py from the repository
```

--------------------------------

### Run Streamlit HITL Application

Source: https://github.com/zvictor/brainyflow/blob/main/docs/cookbook/python.md

Installs dependencies for the Streamlit HITL application and runs the Streamlit application. Users can then access the web UI via a provided URL to submit text, review processed output, and provide feedback.

```bash
pip install -r requirements.txt
streamlit run app.py
```

--------------------------------

### Copy Caskada TypeScript Source Code

Source: https://github.com/zvictor/brainyflow/blob/main/docs/installation.md

Allows for installation by directly copying the TypeScript source code file into your project. This is an alternative to package managers for TypeScript projects.

```typescript
// Copy typescript/caskada.ts from the repository
```

--------------------------------

### Setup and Run Python Chat Application (Bash)

Source: https://github.com/zvictor/brainyflow/blob/main/cookbook/README.md

Steps to set up and run a basic chat application using Caskada and OpenAI's GPT-4o model. It involves setting the API key, installing dependencies, and running the main script.

```bash
export OPENAI_API_KEY="your-api-key-here"
pip install -r requirements.txt
python main.py
```

--------------------------------

### Install Caskada TypeScript Package

Source: https://github.com/zvictor/brainyflow/blob/main/docs/installation.md

Installs the Caskada TypeScript package using pnpm, npm, or yarn. This is the recommended method for TypeScript projects.

```bash
pnpm add caskada
```

```bash
npm install caskada
```

```bash
yarn add caskada
```

--------------------------------

### Install Dependencies and Set API Key (Bash)

Source: https://github.com/zvictor/brainyflow/blob/main/cookbook/README.md

This section provides shell commands to install project dependencies using pip and set the OpenAI API key as an environment variable. It also includes a command to verify the API key setup by running a utility script.

```bash
pip install -r requirements.txt
export OPENAI_API_KEY="your-api-key-here"
python utils.py
```

--------------------------------

### Get User Approval (Python)

Source: https://github.com/zvictor/brainyflow/blob/main/cookbook/python-async-basic/README.md

Asynchronously prompts the user for approval of a recipe suggestion and triggers subsequent actions based on the input. It uses `async/await` for waiting on user input.

```python
async def post(self, shared, prep_res, suggestion):
    # Async user input
    answer = await get_user_input(
        f"Accept {suggestion}? (y/n): "
    )
    self.trigger("accept" if answer == "y" else "retry")
```

--------------------------------

### Parallel Batch Translation Setup and Execution (Python)

Source: https://github.com/zvictor/brainyflow/blob/main/cookbook/README.md

Demonstrates setting up and running a parallel batch translation process using Caskada. It includes steps for installing dependencies, setting API keys, and running the main script. The output comparison highlights the time saved compared to a sequential approach.

```bash
pip install -r requirements.txt
```

```bash
export ANTHROPIC_API_KEY="your-api-key-here"
```

```bash
python utils.py
```

```bash
python main.py
```

```text
# --- Sequential Run Output (from python-batch) ---
Starting sequential translation into 8 languages...
Translated Chinese text
...
Translated Korean text
Saved translation to translations/README_CHINESE.md
...
Saved translation to translations/README_KOREAN.md

Total sequential translation time: ~1136 seconds

=== Translation Complete ===
Translations saved to: translations
============================


# --- Parallel Run Output (this example) ---
Starting parallel translation into 8 languages...
Translated French text
Translated Portuguese text
... # Messages may appear interleaved
Translated Spanish text
Saved translation to translations/README_CHINESE.md
...
Saved translation to translations/README_KOREAN.md

Total parallel translation time: ~209 seconds

=== Translation Complete ===
Translations saved to: translations
============================

_(Actual times will vary based on API response speed and system.)_
```

--------------------------------

### Connect Nodes to Create a QA Flow (Python & TypeScript)

Source: https://github.com/zvictor/brainyflow/blob/main/docs/getting_started.md

Instantiates GetQuestionNode and AnswerNode, connects them sequentially, and creates a Flow instance starting with GetQuestionNode. This demonstrates the basic flow construction in both Python and TypeScript.

```python
from .nodes import GetQuestionNode, AnswerNode
from caskada import Flow

def create_qa_flow():
    get_question_node = GetQuestionNode()
    answer_node = AnswerNode()

    get_question_node >> answer_node

    return Flow(start=get_question_node)
```

```typescript
import { Flow } from 'caskada'

function createQaFlow(): Flow {
  const getQuestionNode = new GetQuestionNode()
  const answerNode = new AnswerNode()

  getQuestionNode.next(answerNode)

  return new Flow(getQuestionNode)
}
```

--------------------------------

### Run Multi-Agent Taboo Game Example

Source: https://github.com/zvictor/brainyflow/blob/main/docs/cookbook/python.md

This section outlines the steps to set up and run a multi-agent Taboo word guessing game example using Caskada. It involves installing dependencies, setting an OpenAI API key, and executing the main Python script. The example demonstrates asynchronous communication between two AI agents (Hinter and Guesser) via message queues.

```bash
pip install -r requirements.txt
```

```bash
export OPENAI_API_KEY=your_api_key_here
```

```bash
python main.py
```

--------------------------------

### Bash: Python Virtual Environment and Dependency Installation

Source: https://github.com/zvictor/brainyflow/blob/main/cookbook/python-tool-database/README.md

Commands to set up a Python virtual environment and install project dependencies from a requirements file. This is a standard setup procedure for Python projects.

```bash
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate
pip install -r requirements.txt
```

--------------------------------

### MCP Demo: Install and Run

Source: https://github.com/zvictor/brainyflow/blob/main/cookbook/README.md

Installs project dependencies and runs the main script for the MCP Demo. Assumes the OPENAI_API_KEY environment variable is set or updated in utils.py.

```bash
pip install -r requirements.txt
python main.py

```

--------------------------------

### Run a QA Flow with Memory Management (Python & TypeScript)

Source: https://github.com/zvictor/brainyflow/blob/main/docs/getting_started.md

Executes a previously created QA flow using an asyncio event loop. It initializes an empty memory object, runs the flow, and then accesses the results stored within the memory object after completion. This example is provided for both Python and TypeScript.

```python
import asyncio
from .flow import create_qa_flow

async def main():
    memory = {}
    qa_flow = create_qa_flow()

    print("Running QA Flow...")
    await qa_flow.run(memory)

    print("\n--- Flow Complete ---")
    print(f"Question: {memory.question}")
    print(f"Answer: {memory.answer}")

if __name__ == '__main__':
    asyncio.run(main())
```

```typescript
import { createQaFlow, QAGlobalStore } from './flow'

async function main() {
  const globalStore: QAGlobalStore = {}
  const qaFlow = createQaFlow()

  console.log('Running QA Flow...')
  await qaFlow.run(globalStore)

  console.log('\n--- Flow Complete ---')
  console.log(`Question: ${globalStore.question ?? 'N/A'}`)
  console.log(`Answer: ${globalStore.answer ?? 'N/A'}`)
}

main().catch(console.error)
```

--------------------------------

### Web Human-in-the-Loop (HITL) Service with Python and FastAPI

Source: https://github.com/zvictor/brainyflow/blob/main/docs/cookbook/python.md

A minimal web application for human-in-the-loop workflows using Caskada, FastAPI, and Server-Sent Events (SSE). Users submit text, it's processed (simulated), and reviewed for approval or rejection, potentially triggering reprocessing. This requires installing dependencies and setting up the FastAPI application.

```python
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
from sse_starlette.sse import EventSourceResponse
import asyncio

from caskada.flows import SequentialFlow
from caskada.nodes import Node

# --- Mock Nodes for Demonstration ---
class MockProcessingNode(Node):
    async def handle(self, message):
        await asyncio.sleep(1) # Simulate processing time
        return {"processed_text": f"Processed: {message['text']}"}

class MockReviewNode(Node):
    async def handle(self, message):
        # In a real app, this would involve UI interaction and state management
        await asyncio.sleep(0.5) # Simulate review time
        return {"status": "approved", "final_text": message['processed_text']}

# --- FastAPI App Setup ---
app = FastAPI()
templates = Jinja2Templates(directory="templates") # Assuming a 'templates' folder exists

# --- In-memory state for demonstration ---
# In a real app, use a database or persistent storage
processed_data = {}

# --- Server-Sent Events (SSE) ---
async def event_generator(request_id):
    while True:
        # Check if data is ready for this request_id
        if request_id in processed_data and processed_data[request_id].get("status") == "approved":
            yield {"data": "Approved", "event": "status_update"}
            break
        if request_id in processed_data and processed_data[request_id].get("status") == "rejected":
            yield {"data": "Rejected", "event": "status_update"}
            break
        await asyncio.sleep(1) # Poll every second

@app.get("/stream/{request_id}")
async def stream_events(request_id: str, request: Request):
    # Ensure the request is still active before returning EventSourceResponse
    async def close_client():
        await asyncio.sleep(10) # Keep the connection open for a while
        # In a real app, you might have more sophisticated connection management
        if not request.state.disconnected:
            pass # Or handle disconnection logic

    request.state.disconnected = False
    request.scope['client'] = (None, None) # Mock client info to prevent unexpected disconnects
    asyncio.create_task(close_client())

    return EventSourceResponse(event_generator(request_id))

# --- API Endpoints ---
@app.post("/submit/")
async def submit_text(data: dict):
    text_to_process = data.get("text")
    if not text_to_process:
        return {"error": "No text provided"}, 400

    request_id = str(hash(text_to_process + str(asyncio.get_event_loop().time()))) # Simple unique ID

    # Define the Caskada flow
    processing_node = MockProcessingNode()
    review_node = MockReviewNode()
    flow = SequentialFlow(nodes=[processing_node, review_node])

    async def run_flow_in_background():
        try:
            result = await flow.run({"text": text_to_process})
            processed_data[request_id] = result
        except Exception as e:
            processed_data[request_id] = {"error": str(e)}

    asyncio.create_task(run_flow_in_background())

    return {"message": "Processing started", "request_id": request_id}

# --- HTML Interface ---
@app.get("/", response_class=HTMLResponse)
async def read_root(request: Request):
    return templates.TemplateResponse("index.html", {"request": request})

# --- Helper to run the app (for demonstration) ---
# To run this: save as main.py, create a 'templates' folder with 'index.html',
# then run `uvicorn main:app --reload`
# You'll need to install: fastapi uvicorn sse-starlette "python-multipart"

```

--------------------------------

### Python Logging with ExecutionTree Example

Source: https://github.com/zvictor/brainyflow/blob/main/docs/guides/visualization_logging.md

An example demonstrating how to log the `ExecutionTree` as JSON in Python after running a Caskada flow. It includes placeholder setup for logger, nodes, flow, and memory, showing where to capture and log the execution result.

```python
import logging
import json
from caskada import Node, Memory, Flow

# ... (setup logger, define nodes as in previous Python logging example) ...

async def main():
    # ... (define node_a, node_b, connect them)
    # node_a = LoggingNode()
    # node_b = LoggingNode()
    # node_a >> node_b

    # flow = Flow(start=node_a)
    # memory_obj = Memory(global_store={"initial": "data"})

    # execution_result_tree = await flow.run(memory_obj)

    # logger.info(f"Flow execution completed. Final Memory: {dict(memory_obj)}")
    # logger.info(f"Execution Tree: {json.dumps(execution_result_tree, indent=2)}")
    pass # Placeholder for actual setup
```

--------------------------------

### Resume Parser Structured Output Example

Source: https://github.com/zvictor/brainyflow/blob/main/docs/cookbook/python.md

An example of the structured output generated by the resume parser. It includes personal information, experience, and identified skills, formatted in YAML.

```yaml
name: JOHN SMTIH
email: johnsmtih1983@gnail.com
experience:
- {title: SALES MANAGER, company: ABC Corportaion}
- {title: ASST. MANAGER, company: XYZ Industries}
- {title: CUSTOMER SERVICE REPRESENTATIVE, company: Fast Solutions Inc}
skill_indexes: [0, 1, 2, 3, 4]

```

--------------------------------

### Implement Question-Answering Flow Nodes in Python

Source: https://github.com/zvictor/brainyflow/blob/main/docs/getting_started.md

Python implementation of `GetQuestionNode` and `AnswerNode` for a question-answering flow. `GetQuestionNode` captures user input, and `AnswerNode` processes it using an LLM. Dependencies include `asyncio`, `caskada`, and a `call_llm` utility.

```python
import asyncio
from caskada import Node, Flow, Memory
from utils import call_llm  # Your LLM implementation

class GetQuestionNode(Node):
    async def prep(self, memory):
        """Get text input from user."""
        memory.question = input("Enter your question: ")

class AnswerNode(Node):
    async def prep(self, memory):
        """Extract the question from memory."""
        return memory.question

    async def exec(self, question: str | None):
        """Call LLM to generate an answer."""

        prompt = f"Answer the following question: {question}"
        return await call_llm(prompt)

    async def post(self, memory, prep_res: str | None, exec_res: str):
        """Store the answer in memory."""
        memory.answer = exec_res
        print(f"AnswerNode: Stored answer '{exec_res}'")
```

--------------------------------

### Project Dependencies

Source: https://github.com/zvictor/brainyflow/blob/main/docs/cookbook/python.md

Lists the Python dependencies required for the project, including FastAPI for the web framework, Uvicorn as the ASGI server, Jinja2 for templating, and Caskada for workflow orchestration.

```text
fastapi
uvicorn[standard]
jinja2
caskada
sse-starlette
```

--------------------------------

### Suggest Recipe (Python)

Source: https://github.com/zvictor/brainyflow/blob/main/cookbook/python-async-basic/README.md

Asynchronously suggests a recipe by calling an LLM with a list of recipes. It utilizes `async/await` for potentially slow LLM interactions.

```python
async def exec(self, recipes):
    # Async LLM call
    suggestion = await call_llm_async(
        f"Choose best recipe from: {recipes}"
    )
    return suggestion
```

--------------------------------

### Install Dependencies (Bash)

Source: https://github.com/zvictor/brainyflow/blob/main/cookbook/README.md

Installs the necessary Python libraries for the voice chat application using pip. It also provides instructions for installing PortAudio on Linux, which is a dependency for the 'sounddevice' library.

```bash
pip install -r requirements.txt
```

```bash
sudo apt-get update && sudo apt-get install -y portaudio19-dev
```

--------------------------------

### Install PortAudio Development Files on Linux

Source: https://github.com/zvictor/brainyflow/blob/main/cookbook/python-voice-chat/README.md

This command installs the PortAudio development files, which may be a dependency for the 'sounddevice' library on some Linux systems. It includes updating the package list and then installing the portaudio19-dev package.

```bash
sudo apt-get update && sudo apt-get install -y portaudio19-dev
```

--------------------------------

### CSS for Frontend Styling

Source: https://github.com/zvictor/brainyflow/blob/main/docs/cookbook/python.md

Basic CSS file for styling the frontend user interface, including layout and appearance of elements like textareas and buttons.

```css
body {
    font-family: sans-serif;
    padding: 20px;
    max-width: 800px;
    margin: auto;
    background-color: #f4f4f4;
}

h1 {
    text-align: center;
    color: #333;
}

textarea {
    width: 100%;
    min-height: 100px;
    margin-bottom: 10px;
    padding: 10px;
    border: 1px solid #ccc;
    box-sizing: border-box;
}

button {
    padding: 10px 15px;
    background-color: #5cb85c;
    color: white;
    border: none;
    cursor: pointer;
    margin-right: 5px;
}

button#rejectBtn {
    background-color: #d9534f;
}

#status {
    margin-top: 15px;
    padding: 10px;
    background-color: #eee;
    border: 1px solid #ddd;
}

#output {
    margin-top: 15px;
    padding: 10px;
    background-color: #fff;
    border: 1px solid #ddd;
    white-space: pre-wrap; /* Preserve whitespace */
}

#feedbackButtons {
    margin-top: 15px;
    text-align: center;
}

```

--------------------------------

### Google Custom Search API Integration (Python)

Source: https://github.com/zvictor/brainyflow/blob/main/docs/utility_function/websearch.md

Python code example to perform a web search using the Google Custom Search JSON API. It requires GOOGLE_API_KEY and GOOGLE_CX_ID environment variables. The function sends a GET request to the Google API and prints the JSON response or an error message.

```Python
import requests
import os

API_KEY = os.environ.get("GOOGLE_API_KEY") # Use environment variables
CX_ID = os.environ.get("GOOGLE_CX_ID")     # Use environment variables
query = "example"

url = "https://www.googleapis.com/customsearch/v1"
params = {
    "key": API_KEY,
    "cx": CX_ID,
    "q": query
}

if not API_KEY or not CX_ID:
    print("Error: Please set GOOGLE_API_KEY and GOOGLE_CX_ID environment variables.")
else:
    try:
        response = requests.get(url, params=params)
        response.raise_for_status() # Raise an exception for bad status codes
        results = response.json()
        print(results)
    except requests.exceptions.RequestException as e:
        print(f"Error fetching Google search results: {e}")

```

--------------------------------

### Synthesize Speech with IBM Watson TTS (Python)

Source: https://github.com/zvictor/brainyflow/blob/main/docs/utility_function/text_to_speech.md

Synthesizes speech from text using IBM Watson Text-to-Speech. Requires installation of 'ibm_watson' and 'ibm_cloud_sdk_core' pip packages and setting environment variables for IBM_API_KEY and IBM_SERVICE_URL. Outputs audio to a specified MP3 file.

```python
# Requires: pip install ibm_watson
from ibm_watson import TextToSpeechV1
from ibm_cloud_sdk_core.authenticators import IAMAuthenticator
import os

def synthesize_ibm_tts(text: str, output_filename: str = "ibm_tts_output.mp3"):
    """Synthesizes speech using IBM Watson TTS."""
    api_key = os.environ.get("IBM_API_KEY")
    service_url = os.environ.get("IBM_SERVICE_URL")

    if not api_key or not service_url:
        print("Error: IBM_API_KEY or IBM_SERVICE_URL not set.")
        return

    try:
        authenticator = IAMAuthenticator(api_key)
        text_to_speech = TextToSpeechV1(authenticator=authenticator)
        text_to_speech.set_service_url(service_url)

        response = text_to_speech.synthesize(
            text=text,
            voice='en-US_AllisonV3Voice', # Example voice
            accept='audio/mp3' # Specify desired format
        ).get_result()

        # The result object has a 'content' attribute with the audio data
        with open(output_filename, 'wb') as audio_file:
            audio_file.write(response.content)
        print(f"Audio saved to {output_filename}")

    except Exception as e:
        print(f"Error calling IBM Watson TTS: {e}")

# Example:

```

--------------------------------

### Include Caskada JavaScript in Browser

Source: https://github.com/zvictor/brainyflow/blob/main/docs/installation.md

Includes the Caskada JavaScript file directly in a browser environment using a script tag. This method allows for immediate use of Caskada functionalities in web applications.

```html
<script type="module">
  import * as caskada from 'https://unpkg.com/caskada@latest/dist/caskada.js'

  new caskada.Node(...)
</script>
```

```html
<script type="module" src="https://unpkg.com/caskada@latest/dist/caskada.js"></script>
<script>
  new globalThis.caskada.Node(...)
</script>
```

--------------------------------

### Install Project Dependencies

Source: https://github.com/zvictor/brainyflow/blob/main/cookbook/python-agent/README.md

Installs all necessary Python packages for the Research Agent project using a requirements file. This is a prerequisite for running the agent.

```bash
pip install -r requirements.txt

```

--------------------------------

### Parallel Batch Translation with Python and Caskada

Source: https://github.com/zvictor/brainyflow/blob/main/docs/cookbook/python.md

Demonstrates translating a document into multiple languages concurrently using Caskada's async and parallel features (`ParallelFlow`, `Node`). This approach leverages `asyncio` and parallel execution to speed up I/O-bound tasks like multiple API calls. It requires setting an Anthropic API key and installing dependencies.

```python
import asyncio
import os

from caskada.nodes import Node
from caskada.flows import ParallelFlow
from utils import TranslateTextNodeParallel, get_languages

async def main():
    # Get target languages
    languages = get_languages()

    # Prepare nodes for parallel translation
    translate_node = TranslateTextNodeParallel(languages)

    # Create a parallel flow
    flow = ParallelFlow(nodes=[translate_node])

    # Run the flow
    await flow.run()

if __name__ == "__main__":
    # Ensure API key is set
    if "ANTHROPIC_API_KEY" not in os.environ:
        raise ValueError("ANTHROPIC_API_KEY environment variable not set.")

    # Run the main async function
    asyncio.run(main())

```

--------------------------------

### Install Dependencies and Set Environment Variable (Bash)

Source: https://github.com/zvictor/brainyflow/blob/main/cookbook/python-a2a/README.md

Installs project dependencies using pip and sets the OpenAI API key as an environment variable. Ensure Python 3.10+ is installed.

```bash
pip install -r requirements.txt
export OPENAI_API_KEY="your-api-key-here"
```

--------------------------------

### HTML Structure for the Frontend UI

Source: https://github.com/zvictor/brainyflow/blob/main/docs/cookbook/python.md

The basic HTML file for the web user interface. It includes a textarea for input, buttons for feedback, and areas to display status updates and results.

```html
<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>BrainyFlow HITL</title>
    <link rel="stylesheet" href="/static/style.css">
</head>
<body>
    <h1>Web Human-in-the-Loop Feedback</h1>
    
    <textarea id="inputText" placeholder="Enter text here..."></textarea>
    <button id="submitBtn">Submit</button>
    
    <div id="status">Status: Idle</div>
    <div id="output"></div>
    <div id="feedbackButtons" style="display: none;">
        <button id="approveBtn">Approve</button>
        <button id="rejectBtn">Reject</button>
    </div>

    <script>
        const inputText = document.getElementById('inputText');
        const submitBtn = document.getElementById('submitBtn');
        const statusDiv = document.getElementById('status');
        const outputDiv = document.getElementById('output');
        const feedbackButtons = document.getElementById('feedbackButtons');
        const approveBtn = document.getElementById('approveBtn');
        const rejectBtn = document.getElementById('rejectBtn');

        let taskId = null;

        submitBtn.onclick = async () => {
            const text = inputText.value;
            if (!text) return;

            const response = await fetch('/submit', {
                method: 'POST',
                headers: {'Content-Type': 'application/json'},
                body: JSON.stringify({ text_input: text })
            });
            const data = await response.json();
            taskId = data.task_id;
            statusDiv.textContent = `Status: Submitted (Task ID: ${taskId})`;
            inputText.value = ''; // Clear input
            feedbackButtons.style.display = 'none'; // Hide feedback buttons initially
            connectSSE(taskId);
        };

        function connectSSE(taskId) {
            const eventSource = new EventSource(`/stream/${taskId}`);

            eventSource.onmessage = (event) => {
                const data = JSON.parse(event.data);
                statusDiv.textContent = `Status: ${data.status}`;
                outputDiv.innerHTML = `<p>${data.data}</p>`;

                if (data.status === 'waiting_for_review') {
                    feedbackButtons.style.display = 'block';
                } else if (data.status === 'completed' || data.status === 'approved' || data.status === 'rejected') {
                    feedbackButtons.style.display = 'none';
                    eventSource.close(); // Close connection on completion or final state
                }
            };

            eventSource.onerror = (err) => {
                console.error('EventSource failed:', err);
                statusDiv.textContent = 'Status: Connection error';
                eventSource.close();
            };
        }

        approveBtn.onclick = async () => {
            await fetch(`/feedback/${taskId}`, {
                method: 'POST',
                headers: {'Content-Type': 'application/json'},
                body: JSON.stringify({ feedback: 'approved' })
            });
            feedbackButtons.style.display = 'none';
        };

        rejectBtn.onclick = async () => {
            await fetch(`/feedback/${taskId}`, {
                method: 'POST',
                headers: {'Content-Type': 'application/json'},
                body: JSON.stringify({ feedback: 'rejected' })
            });
            feedbackButtons.style.display = 'none';
        };

    </script>
</body>
</html>

```

--------------------------------

### Text Summarization with Caskada in Python

Source: https://github.com/zvictor/brainyflow/blob/main/docs/cookbook/python.md

Builds a text summarization tool using Caskada, featuring LLM integration, automatic retries on API failures, and graceful error handling. This example showcases Node architecture and shared store usage.

```bash
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

pip install -r requirements.txt

# Configure your LLM API key (check utils/call_llm.py)

python main.py
```

```python
import caskada

# Placeholder for flow.py content
class SummarizeNode(caskada.Node):
    async def exec(self, shared):
        text_to_summarize = shared.get("data", "")
        # In a real scenario, this would call an LLM API
        # For demonstration, we'll simulate a summary
        summary = f"Summary of '{text_to_summarize[:20]}...'"
        shared["summary"] = summary
        print(f"Generated summary: {summary}")

    async def exec_fallback(self, shared):
        shared["summary"] = "Fallback summary: Could not summarize."
        print("Fallback executed.")

class TextSummarizationFlow(caskada.Flow):
    def __init__(self):
        super().__init__()
        self.add_node(SummarizeNode(), name="summarizer", retries=3)

    async def run(self, shared):
        await self.nodes.get("summarizer").run(shared)

# Placeholder for main.py content
async def main():
    flow = TextSummarizationFlow()
    shared_data = {"data": "This is a long piece of text that needs to be summarized by the Caskada application. We will test its summarization capabilities."}
    await flow.run(shared_data)
    print("Final summary:", shared_data.get("summary"))

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

```

--------------------------------

### Install Project Dependencies

Source: https://github.com/zvictor/brainyflow/blob/main/cookbook/python-batch/README.md

Installs the necessary Python packages for the batch translation process. This command reads the list of dependencies from the requirements.txt file.

```bash
pip install -r requirements.txt
```

--------------------------------

### Install Caskada, Faiss, and OpenAI Libraries

Source: https://github.com/zvictor/brainyflow/blob/main/cookbook/python_demo.ipynb

Installs necessary Python packages: caskada for AI workflows, faiss-cpu for efficient similarity search, and openai for accessing OpenAI's API.

```python
! pip install caskada
! pip install faiss-cpu
! pip install openai
```

--------------------------------

### Fetch Recipes (Python)

Source: https://github.com/zvictor/brainyflow/blob/main/cookbook/python-async-basic/README.md

Asynchronously fetches recipes from an external API based on a given ingredient. It uses `async/await` for non-blocking I/O operations.

```python
async def prep(self, shared):
    ingredient = input("Enter ingredient: ")
    return ingredient

async def exec(self, ingredient):
    # Async API call
    recipes = await fetch_recipes(ingredient)
    return recipes
```

--------------------------------

### Install and Run Terminal Chat Interface (Bash)

Source: https://github.com/zvictor/brainyflow/blob/main/cookbook/typescript-chat/README.md

Commands to install dependencies and run the Node.js/TypeScript terminal chat interface. Ensure you have Node.js and npm installed. Copy the .env.example file and add your OpenAI API key.

```bash
# from Caskada root directory
cd cookbook/typescript-chat

cp .env.example .env # add your API key

npm install
npm run chat
```

--------------------------------

### Example Output of Article Workflow

Source: https://github.com/zvictor/brainyflow/blob/main/cookbook/python-workflow/README.md

Demonstrates the typical output generated by the article writing workflow, including the structured YAML outline, parsed outline, content for each section, and the final styled article.

```text
=== Starting Article Workflow on Topic: AI Safety ===


===== OUTLINE (YAML) =====

sections:
- Introduction to AI Safety
- Key Challenges in AI Safety
- Strategies for Ensuring AI Safety


===== PARSED OUTLINE =====

1. Introduction to AI Safety
2. Key Challenges in AI Safety
3. Strategies for Ensuring AI Safety

=========================


===== SECTION CONTENTS =====

--- Introduction to AI Safety ---
AI Safety is about making sure that artificial intelligence (AI) systems are helpful and not harmful. Imagine teaching a robot to help with chores. AI Safety is like setting ground rules for the robot so it doesn't accidentally cause trouble, like mistaking a pet for a toy. By ensuring AI systems understand their tasks and limitations, we can trust them to act safely. It's about creating guidelines and checks to ensure AI assists us without unintended consequences.

--- Key Challenges in AI Safety ---
AI safety is about ensuring that artificial intelligence systems operate in ways that are beneficial and not harmful. One key challenge is making sure AI makes decisions that align with human values. Imagine teaching a robot to fetch coffee, but it ends up knocking things over because it doesn't understand the mess it creates. Similarly, if AI systems don't fully grasp human intentions, they might act in unexpected ways. The task is to make AI smart enough to achieve goals without causing problems, much like training a puppy to follow rules without chewing on your shoes.

--- Strategies for Ensuring AI Safety ---
Ensuring AI safety is about making sure artificial intelligence behaves as expected and doesn’t cause harm. Imagine AI as a new driver on the road; we need rules and safeguards to prevent accidents. By testing AI systems under different conditions, setting clear rules for their behavior, and keeping human oversight, we can manage risks. For instance, just as cars have brakes to ensure safety, AI systems need to have fail-safes. This helps in building trust and avoiding unexpected issues, keeping both humans and AI on the right track.

===========================


===== FINAL ARTICLE =====

# Welcome to the World of AI Safety

Have you ever wondered what it would be like to have your very own robot helping you around the house? Sounds like a dream, right? But let’s hit pause for a moment. What if this robot mistook your fluffy cat for a toy? That’s exactly where AI Safety comes in. Think of AI Safety as setting some friendly ground rules for your household helper, ensuring that it knows the difference between doing chores and causing a bit of chaos. It’s all about making sure our AI allies play by the rules, making life easier without those pesky accidental hiccups.

# Navigating the Maze of AI Challenges

Picture this: you've asked your trusty robot to grab you a cup of coffee. But instead, it sends mugs flying and spills coffee because it doesn’t quite get the concept of a mess. Frustrating, isn’t it? One of the biggest hurdles in AI Safety is aligning AI decisions with our human values and intentions. It’s like training a puppy not to gnaw on your favorite pair of shoes. Our job is to teach AI how to reach its goals without stepping on our toes, all while being as reliable and lovable as a well-trained pup.

```

--------------------------------

### Text-to-SQL Workflow with Debug Loop (Mermaid)

Source: https://github.com/zvictor/brainyflow/blob/main/docs/cookbook/python.md

This diagram illustrates the Caskada workflow for converting natural language queries to SQL. It shows the sequence of nodes: Get Schema, Generate SQL, Execute SQL, and Debug SQL. The workflow includes a loop for retrying SQL execution upon encountering SQLite errors and defines success and failure end conditions.

```mermaid
graph LR
    A[Get Schema] --> B[Generate SQL]
    B --> C[Execute SQL]
    C -- Success --> E[End]
    C -- SQLite Error --> D{Debug SQL Attempt}
    D -- Corrected SQL --> C
    C -- Max Retries Reached --> F[End with Error]

    style E fill:#dff,stroke:#333,stroke-width:2px
    style F fill:#fdd,stroke:#333,stroke-width:2px

```

--------------------------------

### Run Streamlit Application

Source: https://github.com/zvictor/brainyflow/blob/main/cookbook/python-streamlit-hitl/README.md

Launches the Streamlit web application. This command starts the development server, making the UI accessible via a web browser.

```bash
streamlit run app.py
```

--------------------------------

### Synthesize Speech with Amazon Polly (Python)

Source: https://github.com/zvictor/brainyflow/blob/main/docs/utility_function/text_to_speech.md

Synthesizes speech from text using AWS Polly. Requires the boto3 library and configured AWS credentials. Outputs an MP3 audio file.

```python
# Requires: pip install boto3
import boto3
import os

def synthesize_polly(text: str, output_filename: str = "polly_output.mp3", region: str | None = None):
    """Synthesizes speech using AWS Polly."""
    # Assumes AWS credentials are configured (e.g., via env vars, ~/.aws/credentials)
    aws_region = region or os.environ.get("AWS_REGION", "us-east-1")
    try:
        polly = boto3.client("polly", region_name=aws_region)
        response = polly.synthesize_speech(
            Text=text,
            OutputFormat="mp3",
            VoiceId="Joanna" # Example voice
        )

        # Check if AudioStream is present
        if "AudioStream" in response:
            with open(output_filename, "wb") as f:
                f.write(response["AudioStream"].read())
            print(f"Audio saved to {output_filename}")
        else:
            print("Error: Could not stream audio from Polly.")

    except Exception as e:
        print(f"Error calling AWS Polly: {e}")

# Example:
# synthesize_polly("Hello from AWS Polly!")

```

--------------------------------

### Web Human-in-the-Loop (HITL) Feedback Service Setup (Python)

Source: https://github.com/zvictor/brainyflow/blob/main/cookbook/README.md

Outlines the setup for a minimal web application facilitating human-in-the-loop (HITL) workflows. It uses Caskada, FastAPI, and Server-Sent Events (SSE) to manage a process where users can submit text, have it processed, and then approve or reject the output.

```bash
# Example setup commands would go here, assuming similar to other Python projects
# e.g., pip install -r requirements.txt (if requirements.txt exists)
```

--------------------------------

### Python Online Flow Execution Example

Source: https://github.com/zvictor/brainyflow/blob/main/docs/design_pattern/rag.md

Provides a Python example of executing the 'OnlineFlow'. It initializes memory with a user question, runs the flow using `OnlineFlow.run()`, and then retrieves the final answer from memory. Includes a `main` function to orchestrate offline and online stages, with mock functions for dependencies.

```python
# --- Online Flow Execution ---
async def run_online(memory_from_offline: dict):
    # Add the user's question to the memory from the offline stage
    memory_from_offline["question"] = "Why do people like cats?"

    print(f"\nStarting online RAG flow for question: \"{memory_from_offline['question']}\"
    await OnlineFlow.run(memory_from_offline) # Pass memory object
    # final answer in memory_from_offline["answer"]
    print("Final Answer:", memory_from_offline.get("answer", "N/A")) # Read from memory
    return memory_from_offline

# Example usage combining both stages
async def main():
    # Mock external functions if not defined
    # global get_embedding, create_index, search_index, call_llm
    # get_embedding = ...
    # create_index = ...
    # search_index = ...
    # call_llm = ...

    memory_after_offline = await run_offline()
    if memory_after_offline.get("index"): # Only run online if index exists
        await run_online(memory_after_offline)
    else:
        print("Skipping online flow due to missing index.")

if __name__ == "__main__":
    # Note: Ensure dummy files exist or are created before running
    # For simplicity, file creation moved to run_offline
    asyncio.run(main())
```

--------------------------------

### Brave Search API

Source: https://github.com/zvictor/brainyflow/blob/main/docs/utility_function/websearch.md

This API allows you to perform web searches using Brave Search. It requires a subscription token for authentication.

```APIDOC
## GET /res/v1/web/search

### Description
Performs a web search using the Brave Search API.

### Method
GET

### Endpoint
https://api.search.brave.com/res/v1/web/search

### Parameters
#### Query Parameters
- **q** (string) - Required - The search query.

#### Headers
- **X-Subscription-Token** (string) - Required - Your Brave Search API subscription token.
- **Accept** (string) - Optional - Specifies the desired response format, e.g., 'application/json'.

### Request Example
```json
{
  "query": "example"
}
```

### Response
#### Success Response (200)
- **results** (object) - The search results from Brave Search.

#### Response Example
```json
{
  "web": {
    "results": [
      {
        "title": "Example Domain",
        "url": "http://example.com/",
        "description": "This domain is for use in illustrative examples in documents."
      }
    ]
  }
}
```
```

--------------------------------

### Text-to-SQL Workflow Example (Python)

Source: https://github.com/zvictor/brainyflow/blob/main/cookbook/README.md

Demonstrates a Caskada workflow for converting natural language questions into SQL queries for an SQLite database. It includes an LLM-powered loop for debugging failed queries.

```python
# This is a conceptual representation. The actual code would be within the project files.
# Example of a Caskada node for Text-to-SQL conversion
from caskada.nodes import BaseNode

class TextToSQLNode(BaseNode):
    def __init__(self, db_path: str, llm_client):
        super().__init__()
        self.db_path = db_path
        self.llm_client = llm_client

    def handle(self, input_data: dict) -> dict:
        natural_language_query = input_data.get('query')
        if not natural_language_query:
            raise ValueError("Input must contain a 'query' key.")

        # Placeholder for LLM call to convert text to SQL
        sql_query = self.llm_client.generate_sql(natural_language_query, self.db_path)
        input_data['sql_query'] = sql_query
        
        # Further steps would involve executing the SQL and potentially debugging
        return input_data

# Example of a Caskada node for executing SQL and handling errors
class SQLExecutionNode(BaseNode):
    def __init__(self, db_path: str, llm_client):
        super().__init__()
        self.db_path = db_path
        self.llm_client = llm_client

    def handle(self, input_data: dict) -> dict:
        sql_query = input_data.get('sql_query')
        if not sql_query:
            raise ValueError("Input must contain a 'sql_query' key.")

        try:
            # Placeholder for executing SQL query against SQLite
            results = self.execute_sql(self.db_path, sql_query)
            input_data['results'] = results
            input_data['error'] = None
        except Exception as e:
            # Placeholder for LLM-powered debugging loop
            error_message = str(e)
            input_data['error'] = error_message
            # In a real scenario, you'd use the LLM to analyze the error and the original query
            # and potentially generate a corrected SQL query.
            input_data['sql_query'] = self.llm_client.debug_and_correct_sql(sql_query, error_message)
            # Then re-attempt execution or return the error
            
        return input_data

    def execute_sql(self, db_path, query):
        # This is a placeholder. Actual implementation would use sqlite3 module.
        print(f"Executing SQL: {query} on {db_path}")
        # Simulate success or failure
        if "error" in query.lower():
            raise Exception("Simulated SQL execution error")
        return [{"column1": "value1", "column2": 123}] # Simulated results


# To run this, you would typically define a Caskada workflow:
# workflow = Workflow([TextToSQLNode(...), SQLExecutionNode(...)])
# result = workflow.run({'query': 'How many users are there?'})

```
Tokens:
Raw