FEAT: add the ability to email action plan (#123)

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Bypassing review due to low engineering coverage over the holidays.

Author: michelle-hadfield-nava

app/src/common/components.py

@@ -30,6 +30,42 @@
 logger = logging.getLogger(__name__)
 
 
+def format_resources(resources: list[dict]) -> str:
+    return "\n\n".join([format_resource(resource) for resource in resources])
+
+
+def format_resource(resource: dict) -> str:
+    return "\n".join(
+        [
+            f"### {resource.get('name', 'Unnamed Resource')}",
+            f"- Referral Type: {resource.get('referral_type', 'None')}",
+            f"- Description: {resource.get('description', 'None')}",
+            f"- Website: {resource.get('website', 'None')}",
+            f"- Phone: {', '.join(resource.get('phones', ['None']))}",
+            f"- Email: {', '.join(resource.get('emails', ['None']))}",
+            f"- Addresses: {', '.join(resource.get('addresses', ['None']))}",
+        ]
+    )
+
+
+def format_action_plan(action_plan: dict) -> str:
+    """Format the action plan for email display. Returns empty string if no action plan."""
+    if not action_plan:
+        return ""
+
+    title = action_plan.get("title", "Your Action Plan")
+    summary = action_plan.get("summary", "")
+    content = action_plan.get("content", "")
+
+    parts = [f"## {title}"]
+    if summary:
+        parts.append(f"\n{summary}")
+    if content:
+        parts.append(f"\n{content}")
+
+    return "\n".join(parts)
+
+
 @component
 class EchoNode:
     """
@@ -209,6 +245,32 @@ def run(
 """
 
 
+@component
+class EmailFullResult:
+    """
+    Formats JSON object (representing a list of resources and action plan) and sends it to email address.
+    """
+
+    @component.output_types(status=str, email=str, message=str)
+    def run(self, email: str, resources_dict: dict, action_plan_dict: dict) -> dict:
+        logger.info("Emailing result to %s", email)
+        logger.debug("Resources JSON content:\n%s", json.dumps(resources_dict, indent=2))
+        if action_plan_dict:
+            logger.debug("Action plan JSON content:\n%s", json.dumps(action_plan_dict, indent=2))
+
+        formatted_resources = format_resources(resources_dict.get("resources", []))
+        formatted_action_plan = format_action_plan(action_plan_dict)
+
+        message = f"{EMAIL_INTRO}\n{formatted_resources}\n\n{formatted_action_plan}"
+
+        # Send email via AWS SES
+        subject = "Your Requested Resources and Action Plan"
+        success = send_email(recipient=email, subject=subject, body=message)
+        status = "success" if success else "failed"
+
+        return {"status": status, "email": email, "message": message}
+
+
 @component
 class EmailResult:
     """
@@ -219,7 +281,7 @@ class EmailResult:
     def run(self, email: str, json_dict: dict) -> dict:
         logger.info("Emailing result to %s", email)
         logger.debug("JSON content:\n%s", json.dumps(json_dict, indent=2))
-        formatted_resources = self.format_resources(json_dict.get("resources", []))
+        formatted_resources = format_resources(json_dict.get("resources", []))
         message = f"{EMAIL_INTRO}\n{formatted_resources}"
 
         # Send email via AWS SES
@@ -229,22 +291,6 @@ def run(self, email: str, json_dict: dict) -> dict:
 
         return {"status": status, "email": email, "message": message}
 
-    def format_resources(self, resources: list[dict]) -> str:
-        return "\n\n".join([self.format_resource(resource) for resource in resources])
-
-    def format_resource(self, resource: dict) -> str:
-        return "\n".join(
-            [
-                f"### {resource.get('name', 'Unnamed Resource')}",
-                f"- Referral Type: {resource.get('referral_type', 'None')}",
-                f"- Description: {resource.get('description', 'None')}",
-                f"- Website: {resource.get('website', 'None')}",
-                f"- Phone: {', '.join(resource.get('phones', ['None']))}",
-                f"- Email: {', '.join(resource.get('emails', ['None']))}",
-                f"- Addresses: {', '.join(resource.get('addresses', ['None']))}",
-            ]
-        )
-
 
 BaseModelT = TypeVar("BaseModelT", bound=BaseModel)
 

app/src/pipelines/email_full_result/__init__.py

@@ -0,0 +1,93 @@
+import logging
+from pprint import pformat
+
+from fastapi import HTTPException
+from hayhooks import BasePipelineWrapper
+from haystack import Pipeline
+from haystack.core.errors import PipelineRuntimeError
+from openinference.instrumentation import _tracers, using_metadata
+from opentelemetry.trace.status import Status, StatusCode
+
+from src.common import components, phoenix_utils
+
+logger = logging.getLogger(__name__)
+tracer = phoenix_utils.tracer_provider.get_tracer(__name__)
+
+
+class PipelineWrapper(BasePipelineWrapper):
+    name = "email_full_result"
+
+    def setup(self) -> None:
+        pipeline = Pipeline()
+        pipeline.add_component("load_resources", components.LoadResult())
+        pipeline.add_component("load_action_plan", components.LoadResult())
+        pipeline.add_component("email_full_result", components.EmailFullResult())
+
+        pipeline.connect("load_resources.result_json", "email_full_result.resources_dict")
+        pipeline.connect("load_action_plan.result_json", "email_full_result.action_plan_dict")
+        pipeline.add_component("logger", components.ReadableLogger())
+
+        self.pipeline = pipeline
+
+    def run_api(self, resources_result_id: str, action_plan_result_id: str, email: str) -> dict:
+        with using_metadata({"email": email}):
+            # Must set using_metadata context before calling tracer.start_as_current_span()
+            assert isinstance(tracer, _tracers.OITracer), f"Got unexpected {type(tracer)}"
+            with tracer.start_as_current_span(  # pylint: disable=not-context-manager,unexpected-keyword-arg
+                self.name, openinference_span_kind="chain"
+            ) as span:
+                result = self._run(resources_result_id, action_plan_result_id, email)
+                span.set_input(
+                    {
+                        "resources_result_id": resources_result_id,
+                        "action_plan_result_id": action_plan_result_id,
+                    }
+                )
+                span.set_output(result["email_full_result"]["status"])
+                span.set_status(Status(StatusCode.OK))
+                return result
+
+    def _run(self, resources_result_id: str, action_plan_result_id: str, email: str) -> dict:
+        try:
+            run_data = {
+                "logger": {
+                    "messages_list": [
+                        {
+                            "resources_result_id": resources_result_id,
+                            "action_plan_result_id": action_plan_result_id,
+                            "email": email,
+                        }
+                    ],
+                },
+                "load_resources": {
+                    "result_id": resources_result_id,
+                },
+                "email_full_result": {
+                    "email": email,
+                },
+            }
+
+            run_data["load_action_plan"] = {
+                "result_id": action_plan_result_id,
+            }
+
+            response = self.pipeline.run(
+                run_data,
+                include_outputs_from={"email_full_result"},
+            )
+            logger.debug("Results: %s", pformat(response, width=160))
+            return response
+        except PipelineRuntimeError as re:
+            error_msg = str(re)
+            if re.component_type == components.LoadResult:
+                if "Invalid JSON format in result" in error_msg:
+                    status_code = 500  # Internal error
+                else:
+                    status_code = 400  # User error
+            else:
+                status_code = 500  # Internal error
+
+            raise HTTPException(
+                status_code=status_code,
+                detail=f"Error occurred: {error_msg}",
+            ) from re

app/src/pipelines/email_full_result/pipeline_wrapper.py

@@ -28,15 +28,15 @@ def setup(self) -> None:
 
         self.pipeline = pipeline
 
-    def run_api(self, result_id: str, email: str) -> dict:
+    def run_api(self, resources_result_id: str, email: str) -> dict:
         with using_metadata({"email": email}):
             # Must set using_metadata context before calling tracer.start_as_current_span()
             assert isinstance(tracer, _tracers.OITracer), f"Got unexpected {type(tracer)}"
             with tracer.start_as_current_span(  # pylint: disable=not-context-manager,unexpected-keyword-arg
                 self.name, openinference_span_kind="chain"
             ) as span:
-                result = self._run(result_id, email)
-                span.set_input(result_id)
+                result = self._run(resources_result_id, email)
+                span.set_input(resources_result_id)
                 span.set_output(result["email_result"]["status"])
                 span.set_status(Status(StatusCode.OK))
                 return result

app/src/pipelines/email_result/pipeline_wrapper.py

@@ -9,7 +9,12 @@
 from pydantic import BaseModel
 
 from src.common import haystack_utils, phoenix_utils
-from src.common.components import OpenAIWebSearchGenerator, ReadableLogger
+from src.common.components import (
+    LlmOutputValidator,
+    OpenAIWebSearchGenerator,
+    ReadableLogger,
+    SaveResult,
+)
 from src.pipelines.generate_referrals.pipeline_wrapper import Resource
 
 logger = logging.getLogger(__name__)
@@ -50,7 +55,12 @@ def setup(self) -> None:
             ),
             name="prompt_builder",
         )
+        pipeline.add_component("output_validator", LlmOutputValidator(ActionPlan))
+        pipeline.add_component("save_result", SaveResult())
+
         pipeline.connect("prompt_builder", "llm.messages")
+        pipeline.connect("llm.replies", "output_validator")
+        pipeline.connect("output_validator.valid_replies", "save_result.replies")
 
         pipeline.add_component("logger", ReadableLogger())
         pipeline.connect("llm", "logger")
@@ -90,10 +100,13 @@ def _run(self, resource_objects: list[Resource], user_email: str, user_query: st
                 },
                 "llm": {"model": "gpt-5-mini", "reasoning_effort": "low"},
             },
-            include_outputs_from={"llm"},
+            include_outputs_from={"llm", "save_result"},
         )
         logger.debug("Results: %s", pformat(response, width=160))
-        return {"response": response["llm"]["replies"][0]._content[0].text}
+        return {
+            "response": response["llm"]["replies"][0]._content[0].text,
+            "save_result": response["save_result"],
+        }
 
 
 def get_resources(resources: list[Resource] | list[dict]) -> list[Resource]: