Update agent-video example

Author: leawn

agent_video/pyproject.toml

@@ -8,9 +8,9 @@ dependencies = [
     "openai>=1.59.9",
     "pipecat-ai[daily,deepgram,openai,silero,cartesia]>=0.0.58",
     "python-dotenv>=1.0.1",
-    "restack-ai>=0.0.77",
     "pydantic>=2.10.6",
     "watchfiles>=1.0.4",
+    "restack-ai>=0.0.78",
 ]
 
 [project.scripts]
@@ -23,6 +23,9 @@ include = ["src"]
 [tool.hatch.build.targets.wheel]
 include = ["src"]
 
+[tool.uv.sources]
+restack-ai = { path = "../../../sdk/engine/libraries/python/dist/restack_ai-0.0.78-py3-none-any.whl" }
+
 [build-system]
 requires = ["hatchling"]
 build-backend = "hatchling.build"

agent_video/src/agents/agent.py

@@ -4,8 +4,8 @@
 from restack_ai.agent import agent, import_functions, log
 
 with import_functions():
-    from src.functions.llm_chat import LlmChatInput, Message, llm_chat
     from src.functions.context_docs import context_docs
+    from src.functions.llm_chat import LlmChatInput, Message, llm_chat
 
 class MessagesEvent(BaseModel):
     messages: list[Message]

agent_video/src/functions/context_docs.py

@@ -1,23 +1,22 @@
-from restack_ai.function import function, log
 import aiohttp
+from restack_ai.function import function, log
 
 
 async def fetch_content_from_url(url: str) -> str:
     async with aiohttp.ClientSession() as session:
         async with session.get(url) as response:
             if response.status == 200:
                 return await response.text()
-            else:
-                log.error("Failed to fetch content", status=response.status)
-                raise Exception(f"Failed to fetch content: {response.status}")
+            log.error("Failed to fetch content", status=response.status)
+            raise Exception(f"Failed to fetch content: {response.status}")
 
 
 @function.defn()
 async def context_docs() -> str:
     try:
         docs_content = await fetch_content_from_url("https://docs.restack.io/llms-full.txt")
         log.info("Fetched content from URL", content=len(docs_content))
-        
+
         return docs_content
 
     except Exception as e:

agent_video/src/functions/pipeline.py

@@ -21,7 +21,7 @@
 from pipecat.services.tavus import TavusVideoService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
 from pydantic import BaseModel
-from restack_ai.function import FunctionFailure, function, log
+from restack_ai.function import NonRetryableError, function, log
 
 load_dotenv(override=True)
 
@@ -159,7 +159,7 @@ async def run_pipeline():
                     )
                     # Cancel the pipeline task if an error occurs within the pipeline runner.
                     await task.cancel()
-                    raise e
+                    raise NonRetryableError(f"Pipecat pipeline failed: {e}") from e
 
             # Launch the pipeline runner as a background task so it doesn't block the return.
             asyncio.create_task(run_pipeline())
@@ -170,4 +170,4 @@ async def run_pipeline():
             return room_url
     except Exception as e:
         log.error("Pipecat pipeline failed", error=e)
-        raise FunctionFailure(f"Pipecat pipeline failed: {e}", non_retryable=True)
+        raise NonRetryableError(f"Pipecat pipeline failed: {e}") from e

agent_video/src/services.py

@@ -7,10 +7,10 @@
 
 from src.agents.agent import AgentVideo
 from src.client import client
+from src.functions.context_docs import context_docs
 from src.functions.llm_chat import llm_chat
 from src.functions.pipeline import pipecat_pipeline
 from src.workflows.room import RoomWorkflow
-from src.functions.context_docs import context_docs
 
 
 async def main() -> None:

agent_video/src/workflows/room.py

@@ -2,6 +2,7 @@
 
 from pydantic import BaseModel
 from restack_ai.workflow import (
+    NonRetryableError,
     ParentClosePolicy,
     import_functions,
     log,
@@ -24,27 +25,35 @@ class RoomWorkflow:
     @workflow.run
     async def run(self) -> RoomWorkflowOutput:
         agent_id = f"{workflow_info().workflow_id}-agent"
-        agent = await workflow.child_start(
-            agent=AgentVideo,
-            agent_id=agent_id,
-            start_to_close_timeout=timedelta(minutes=20),
-            parent_close_policy=ParentClosePolicy.ABANDON,
-        )
-
-        log.info("Agent started", agent=agent)
-
-        room_url = await workflow.step(
-            function=pipecat_pipeline,
-            function_input=PipecatPipelineInput(
-                agent_name=AgentVideo.__name__,
-                agent_id=agent.id,
-                agent_run_id=agent.run_id,
-            ),
-            start_to_close_timeout=timedelta(minutes=20),
-        )
-
-        log.info("Pipecat pipeline started")
-
-        log.info("RoomWorkflow completed", room_url=room_url)
-
-        return RoomWorkflowOutput(room_url=room_url)
+        try:
+            agent = await workflow.child_start(
+                agent=AgentVideo,
+                agent_id=agent_id,
+                start_to_close_timeout=timedelta(minutes=20),
+                parent_close_policy=ParentClosePolicy.ABANDON,
+            )
+        except Exception as e:
+            error_message = f"Error during child_start: {e}"
+            raise NonRetryableError(error_message) from e
+        else:
+            log.info("Agent started", agent=agent)
+
+            try:
+                room_url = await workflow.step(
+                    function=pipecat_pipeline,
+                    function_input=PipecatPipelineInput(
+                        agent_name=AgentVideo.__name__,
+                        agent_id=agent.id,
+                        agent_run_id=agent.run_id,
+                    ),
+                    start_to_close_timeout=timedelta(minutes=20),
+                )
+            except Exception as e:
+                error_message = f"Error during pipecat_pipeline: {e}"
+                raise NonRetryableError(error_message) from e
+            else:
+                log.info("Pipecat pipeline started")
+
+                log.info("RoomWorkflow completed", room_url=room_url)
+
+                return RoomWorkflowOutput(room_url=room_url)