Merge pull request #115 from restackio/re-act-example

leawn · web-flow · commit 74fb1a88fba1 · 2025-01-03T14:50:17.000+01:00
Re_act example
diff --git a/re_act/README.md b/re_act/README.md
@@ -0,0 +1,53 @@
+# Restack AI - ReAct Example
+
+## Prerequisites
+
+- Python 3.11
+- Poetry (for dependency management)
+- Docker (for running the Restack services)
+
+## Usage
+
+2. Open the web UI to see the workflows:
+
+   ```bash
+   http://localhost:5233
+   ```
+
+3. Clone this repository:
+
+   ```bash
+   git clone https://github.com/restackio/examples-python
+   cd examples-python/re_act
+   ```
+
+4. Install dependencies using Poetry:
+
+   ```bash
+   poetry env use 3.11
+   ```
+
+   ```bash
+   poetry shell
+   ```
+
+   ```bash
+   poetry install
+   ```
+
+   ```bash
+   poetry env info # Optional: copy the interpreter path to use in your IDE (e.g. Cursor, VSCode, etc.)
+   ```
+
+5. Run the services:
+
+   ```bash
+   poetry run services
+   ```
+
+6. Schedule workflow
+
+   ```bash
+   poetry run schedule_workflow
+   ```
+
diff --git a/re_act/pyproject.toml b/re_act/pyproject.toml
@@ -0,0 +1,23 @@
+[tool.poetry]
+name = "re_act_example"
+version = "0.0.1"
+description = "ReAct Example"
+authors = [
+    "Restack Team <service@restack.io>",
+]
+packages = [{include = "src"}]
+
+[tool.poetry.dependencies]
+python = ">=3.10,<4.0"
+openai = "^1.57.2"
+python-dotenv = "1.0.1"
+restack-ai = "0.0.51"
+sendgrid = "^6.11.0"
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"
+
+[tool.poetry.scripts]
+services = "src.services:run_services"
+schedule_workflow = "schedule_workflow:run_schedule_workflow"
diff --git a/re_act/schedule_workflow.py b/re_act/schedule_workflow.py
@@ -0,0 +1,30 @@
+import asyncio
+import time
+from restack_ai import Restack
+
+async def main():
+
+    client = Restack()
+
+    workflow_id = f"{int(time.time() * 1000)}-ParentWorkflow"
+    runId = await client.schedule_workflow(
+        workflow_name="ParentWorkflow",
+        workflow_id=workflow_id,
+        input={
+            "email": "john.doe@example.com",
+            "current_accepted_applicants_count": 9
+        }
+    )
+
+    await client.get_workflow_result(
+        workflow_id=workflow_id,
+        run_id=runId
+    )
+
+    exit(0)
+
+def run_schedule_workflow():
+    asyncio.run(main())
+
+if __name__ == "__main__":
+    run_schedule_workflow()
diff --git a/re_act/src/client.py b/re_act/src/client.py
@@ -0,0 +1,3 @@
+from restack_ai import Restack
+
+client = Restack()
diff --git a/re_act/src/functions/decide.py b/re_act/src/functions/decide.py
@@ -0,0 +1,55 @@
+from restack_ai.function import function, log, FunctionFailure
+from dataclasses import dataclass
+from openai import OpenAI
+import os
+from dotenv import load_dotenv
+
+load_dotenv()
+
+@dataclass
+class DecideInput:
+    email: str
+    current_accepted_applicants_count: int
+
+@function.defn()
+async def decide(input: DecideInput):
+    try:
+        if (os.environ.get("OPENAI_API_KEY") is None):
+                raise FunctionFailure("OPENAI_API_KEY is not set", non_retryable=True)
+            
+        client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
+
+        response = client.chat.completions.create(
+            model="gpt-4o-mini",
+            messages=[
+                {
+                    "role": "system",
+                    "content": f"""
+                    You are a helpful assistant for event registration that decides if the applicant should be accepted or rejected.
+                    """
+                },
+                {
+                    "role": "user",
+                    "content": f"""
+                    The event is called "Restack AI Summit 2025"
+                    Restack is the main sponsor of the event.
+                    The applicant has the following email: {input.email}
+                    The current number of accepted applicants is: {input.current_accepted_applicants_count}
+                    The maximum number of accepted applicants is: 10
+
+                    Decide if the applicant should be accepted or rejected.
+                    Return only a JSON object with these exact fields:
+                    {{
+                        "accepted": boolean
+                    }}
+                    """,
+                }
+            ],
+            response_format={"type": "json_object"},
+        )
+        
+        return response.choices[0].message.content
+    except Exception as e:
+        log.error("Failed to decide", error=e)
+        raise FunctionFailure("Failed to decide", non_retryable=True)
+
diff --git a/re_act/src/functions/generate_email_content.py b/re_act/src/functions/generate_email_content.py
@@ -0,0 +1,41 @@
+from restack_ai.function import function, log, FunctionFailure
+from dataclasses import dataclass
+from openai import OpenAI
+import os
+from dotenv import load_dotenv
+
+load_dotenv()
+
+@dataclass
+class GenerateEmailInput:
+    email_context: str
+
+@function.defn()
+async def generate_email_content(input: GenerateEmailInput):
+
+    try:
+        client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
+
+        response = client.chat.completions.create(
+            model="gpt-4o-mini",
+            messages=[
+                {
+                    "role": "system",
+                    "content": f"""
+                    You are a helpful assistant that generates short emails based on the provided context.
+                    """
+                },
+                {
+                    "role": "user",
+                    "content": f"""Generate a short email based on the following context: {input.email_context}
+                    """
+                }
+            ],
+            max_tokens=150
+        )
+        
+        return response.choices[0].message.content
+    except Exception as e:
+        log.error("Failed to generate email content", error=e)
+        raise FunctionFailure("Failed to generate email content", non_retryable=True)
+
diff --git a/re_act/src/functions/send_email.py b/re_act/src/functions/send_email.py
@@ -0,0 +1,39 @@
+import sendgrid
+from sendgrid.helpers.mail import Mail
+from restack_ai.function import function, log, FunctionFailure
+from dotenv import load_dotenv
+import os
+from dataclasses import dataclass
+
+load_dotenv()
+
+@dataclass
+class SendEmailInput:
+    subject: str
+    body: str
+
+@function.defn()
+async def send_email(input: SendEmailInput):
+    from_email = os.environ.get("FROM_EMAIL")
+
+    if not from_email:
+        raise FunctionFailure('FROM_EMAIL is not set', non_retryable=True)
+    
+    sendgrid_api_key = os.getenv('SENDGRID_API_KEY')
+
+    if not sendgrid_api_key:
+        raise FunctionFailure('SENDGRID_API_KEY is not set', non_retryable=True)
+    
+    message = Mail(
+        from_email=from_email,
+        to_emails=from_email,
+        subject=input.subject,
+        plain_text_content=input.body
+    )
+
+    try:
+        sg = sendgrid.SendGridAPIClient(api_key=sendgrid_api_key)
+        sg.send(message)
+    except Exception as e:
+        log.error("Failed to send email", error=e)
+        raise FunctionFailure("Failed to send email", non_retryable=True)
diff --git a/re_act/src/services.py b/re_act/src/services.py
@@ -0,0 +1,22 @@
+import asyncio
+from src.client import client
+from src.functions.decide import decide
+from src.functions.generate_email_content import generate_email_content
+from src.functions.send_email import send_email
+from src.workflows.parent_workflow import ParentWorkflow
+from src.workflows.child_workflow_a import ChildWorkflowA
+from src.workflows.child_workflow_b import ChildWorkflowB
+
+async def main():
+    await asyncio.gather(
+        client.start_service(
+            workflows=[ParentWorkflow, ChildWorkflowA, ChildWorkflowB],
+            functions=[decide, generate_email_content, send_email]
+        )
+    )
+
+def run_services():
+    asyncio.run(main())
+
+if __name__ == "__main__":
+    run_services()
diff --git a/re_act/src/workflows/child_workflow_a.py b/re_act/src/workflows/child_workflow_a.py
@@ -0,0 +1,30 @@
+from restack_ai.workflow import workflow, log, import_functions
+from datetime import timedelta
+from dataclasses import dataclass
+
+with import_functions():
+    from src.functions.generate_email_content import generate_email_content, GenerateEmailInput
+    from src.functions.send_email import send_email, SendEmailInput
+
+@workflow.defn()
+class ChildWorkflowA:
+    @workflow.run
+    async def run(self, email: str):
+        log.info(f"Sending email to {email}")
+
+        text = await workflow.step(
+            generate_email_content,
+            input=GenerateEmailInput(
+                email_context="Application was accepted"
+            ),
+            start_to_close_timeout=timedelta(seconds=120)
+        )
+
+        await workflow.step(
+            send_email,
+            input=SendEmailInput(
+                subject="Restack AI Summit 2025",
+                body=text
+            ),
+            start_to_close_timeout=timedelta(seconds=120)
+        )
diff --git a/re_act/src/workflows/child_workflow_b.py b/re_act/src/workflows/child_workflow_b.py
@@ -0,0 +1,30 @@
+from restack_ai.workflow import workflow, log, import_functions
+from datetime import timedelta
+from dataclasses import dataclass
+
+with import_functions():
+    from src.functions.generate_email_content import generate_email_content, GenerateEmailInput
+    from src.functions.send_email import send_email, SendEmailInput
+
+@workflow.defn()
+class ChildWorkflowB:
+    @workflow.run
+    async def run(self, email: str):
+        log.info(f"Sending email to {email}")
+
+        text = await workflow.step(
+            generate_email_content,
+            input=GenerateEmailInput(
+                email_context="Application was rejected"
+            ),
+            start_to_close_timeout=timedelta(seconds=120)
+        )
+
+        await workflow.step(
+            send_email,
+            input=SendEmailInput(
+                subject="Restack AI Summit 2025",
+                body=text
+            ),
+            start_to_close_timeout=timedelta(seconds=120)
+        )
diff --git a/re_act/src/workflows/parent_workflow.py b/re_act/src/workflows/parent_workflow.py
@@ -0,0 +1,49 @@
+from restack_ai.workflow import workflow, log, workflow_info, import_functions
+from datetime import timedelta
+import json
+from dataclasses import dataclass
+from .child_workflow_a import ChildWorkflowA
+from .child_workflow_b import ChildWorkflowB
+
+with import_functions():
+    from src.functions.decide import decide, DecideInput
+
+@dataclass
+class ParentWorkflowInput:
+    email: str
+    current_accepted_applicants_count: int
+
+@workflow.defn()
+class ParentWorkflow:
+    @workflow.run
+    async def run(self, input: ParentWorkflowInput):
+        parent_workflow_id = workflow_info().workflow_id
+
+        decide_result = await workflow.step(
+            decide,
+            input=DecideInput(
+                email=input.email,
+                current_accepted_applicants_count=input.current_accepted_applicants_count
+            ),
+            start_to_close_timeout=timedelta(seconds=120)
+        )
+
+        decision = json.loads(decide_result)["accepted"]
+
+        child_workflow_result = None
+        if decision:
+            child_workflow_result = await workflow.child_execute(
+                ChildWorkflowA,
+                workflow_id=f"{parent_workflow_id}-child-a",
+                input=input.email
+            )
+        elif not decision:
+            child_workflow_result = await workflow.child_execute(
+                ChildWorkflowB,
+                workflow_id=f"{parent_workflow_id}-child-b",
+                input=input.email
+            )
+
+        return child_workflow_result
+
+

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,3 @@`
	`1`	`+from restack_ai import Restack`
	`2`	`+`
	`3`	`+client = Restack()`