doc: Add more app documentation; add LLM temperature setting (#185)

Author: yoomlam

app/src/app.py

@@ -1,3 +1,9 @@
+"""
+This application uses Hayhooks to create an API service.
+API requests are handled by Hayhooks and triggers specific Haystack pipelines (defined in pipeline_wrapper.py files).
+Phoenix is used for prompt templates and OpenTelemetry-based tracing of API requests.
+"""
+
 import logging
 from typing import Dict
 

app/src/app_config.py

@@ -10,6 +10,11 @@
 
 
 class AppConfig(PydanticBaseEnvConfig):
+    """
+    Application configuration settings, overridable by environment variables of the same name.
+    See docker-compose.yml, local.env, and override.env for environment variable being set.
+    """
+
     environment: str = "local"
     # Preview environment bucket names look like 'p-###-labs-referral-pilot-app-dev'
     bucket_name: str = "local"
@@ -23,6 +28,9 @@ class AppConfig(PydanticBaseEnvConfig):
     phoenix_collector_endpoint: str = "https://phoenix:6006"
     batch_otel: bool = True
 
+    # For sending emails, AWS SES region configuration where email address is verified.
+    aws_ses_region: str = "us-east-1"
+    # From field of the email sent via AWS SES. This email address needs to be verified in SES.
     aws_ses_from_email: str = "no-reply@test.com"
 
     @cached_property
@@ -33,7 +41,7 @@ def db_session(self) -> db.Session:
         return self.db_client.get_session()
 
     # These versions should only be used for the deployed Phoenix instance.
-    # Version ids are base64 encodings of 'PromptVersion:N' where N is simply a counter,
+    # Be aware: Version ids are base64 encodings of 'PromptVersion:N' where N is simply a counter,
     # so they are not unique across different Phoenix instances.
     PROMPT_VERSIONS: dict = {
         "generate_referrals_centraltx": "UHJvbXB0VmVyc2lvbjoxNDM=",
@@ -65,12 +73,11 @@ def db_session(self) -> db.Session:
 
     generate_referrals_rag_model_version: str = "gpt-5.1"
     generate_referrals_rag_reasoning_level: str = "none"
+    generate_referrals_rag_temperature: float = 0.9
 
     generate_action_plan_model_version: str = "gpt-5.1"
     generate_action_plan_reasoning_level: str = "none"
-
-    generate_referrals_from_doc_model_version: str = "gpt-5.1"
-    generate_referrals_from_doc_reasoning_level: str = "none"
+    generate_action_plan_temperature: float = 0.9
 
     def chroma_client(self) -> ClientAPI:
         return chromadb.HttpClient(host=self.rag_db_host, port=self.rag_db_port)

app/src/common/components.py

@@ -39,42 +39,6 @@
 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:
     """
@@ -213,6 +177,7 @@ def __init__(self) -> None:
         Initialize the OpenAI web search generator.
         """
 
+        self.client = OpenAI()
         # Declare this attribute so it can be set when streaming_generator() is called
         self.streaming_callback: Callable | None = None
 
@@ -224,13 +189,18 @@ def run(
         model: str = config.default_openai_model_version,
         reasoning_effort: str = config.default_openai_reasoning_level,
         streaming: bool = False,
+        temperature: float = 1.0,
     ) -> dict:
         """
         Run the OpenAI web search generator.
 
         Args:
             messages: List of ChatMessage objects to send to the API
             domain: Domain to restrict web search to
+            model: LLM model to use
+            reasoning_effort: Reasoning effort level
+            streaming: Whether to use streaming response
+            temperature: temperature for the LLM
 
         Returns:
             Dictionary with response key containing string of response
@@ -253,15 +223,12 @@ def run(
             "input": prompt,
             "reasoning": {"effort": reasoning_effort},
             "tools": [{"type": "web_search"}],
-            # Add other parameters, like temperature
+            "temperature": temperature,
         }
 
         if domain:
             api_params["tools"][0]["filters"] = {"allowed_domains": [domain]}
 
-        client = OpenAI()
-
-        # Use streaming if callback is provided
         if streaming:
             logger.info(
                 "Starting OpenAI streaming request (model=%s, reasoning_effort=%s)",
@@ -271,70 +238,15 @@ def run(
             api_params["stream"] = True
 
             try:
-                response = client.responses.create(**api_params)
-            except Exception as e:
-                logger.error("Failed to create OpenAI stream: %s", e, exc_info=True)
-                raise
-
-            # Collect full response while streaming
-            full_text = ""
-            chunk_count = 0
-
-            try:
-                for openai_chunk in response:
-                    chunk_count += 1
-                    chunk_text = ""
-
-                    # Extract text from OpenAI Responses API events
-                    if hasattr(openai_chunk, "type"):
-                        # Check delta attribute (for text delta events)
-                        if not chunk_text and hasattr(openai_chunk, "delta"):
-                            delta = openai_chunk.delta
-                            if isinstance(delta, str):
-                                chunk_text = delta
-                            elif isinstance(delta, list):
-                                chunk_text = "".join(str(item) for item in delta)
-                            elif hasattr(delta, "content"):
-                                chunk_text = delta.content or ""
-                            elif hasattr(delta, "text"):
-                                chunk_text = delta.text or ""
-
-                    # Fallback for non-Responses API format
-                    if not chunk_text and hasattr(openai_chunk, "output_text"):
-                        chunk_text = openai_chunk.output_text or ""
-
-                    if chunk_text:
-                        full_text += chunk_text
-                        # Convert to Haystack StreamingChunk and call the callback
-                        streaming_chunk = StreamingChunk(content=chunk_text)
-                        assert (
-                            self.streaming_callback is not None
-                        ), "Expected streaming_callback to be set by Hayhooks"
-                        self.streaming_callback(streaming_chunk)
-
-                    # Capture metadata from OpenAI chunk; handle each type of Response*Event
-                    if isinstance(openai_chunk, ResponseOutputItemDoneEvent):
-                        if isinstance(openai_chunk.item, ResponseFunctionWebSearch):
-                            self._add_child_spans([openai_chunk.item])
-                    elif isinstance(openai_chunk, ResponseCreatedEvent):
-                        resp = openai_chunk.response
-                        span = trace.get_current_span()
-                        span.set_attribute("model", str(resp.model))
-                        span.set_attribute("reasoning_effort", str(resp.reasoning))
-                        span.set_attribute("temperature", str(resp.temperature))
-
+                response = self.client.responses.create(**api_params)
+                full_text = self._stream_response(response)
+                return {"replies": [ChatMessage.from_assistant(full_text)]}
             except Exception as e:
-                logger.error("Error during streaming: %s", e, exc_info=True)
+                logger.error("Failed to stream response: %s", e, exc_info=True)
                 raise
-
-            logger.info("Streaming complete: %d chunks, %d characters", chunk_count, len(full_text))
-            if not full_text:
-                logger.warning("No text collected during streaming")
-
-            return {"replies": [ChatMessage.from_assistant(full_text)]}
         else:
             # Non-streaming response
-            response = client.responses.create(**api_params)
+            response = self.client.responses.create(**api_params)
 
             web_search_responses = [
                 item for item in response.output if isinstance(item, ResponseFunctionWebSearch)
@@ -348,6 +260,58 @@ def run(
                 "web_search": [str(result) for result in web_search_responses],
             }
 
+    def _stream_response(self, response: Any) -> str:
+        # Collect full response while streaming
+        full_text = ""
+        chunk_count = 0
+
+        for openai_chunk in response:
+            chunk_count += 1
+            chunk_text = ""
+
+            # Extract text from OpenAI Responses API events
+            if hasattr(openai_chunk, "type"):
+                # Check delta attribute (for text delta events)
+                if not chunk_text and hasattr(openai_chunk, "delta"):
+                    delta = openai_chunk.delta
+                    if isinstance(delta, str):
+                        chunk_text = delta
+                    elif isinstance(delta, list):
+                        chunk_text = "".join(str(item) for item in delta)
+                    elif hasattr(delta, "content"):
+                        chunk_text = delta.content or ""
+                    elif hasattr(delta, "text"):
+                        chunk_text = delta.text or ""
+
+            # Fallback for non-Responses API format
+            if not chunk_text and hasattr(openai_chunk, "output_text"):
+                chunk_text = openai_chunk.output_text or ""
+
+            if chunk_text:
+                full_text += chunk_text
+                # Convert to Haystack StreamingChunk and call the callback
+                streaming_chunk = StreamingChunk(content=chunk_text)
+                assert (
+                    self.streaming_callback is not None
+                ), "Expected streaming_callback to be set by Hayhooks"
+                self.streaming_callback(streaming_chunk)
+
+            # Capture metadata from OpenAI chunk; handle each type of Response*Event
+            if isinstance(openai_chunk, ResponseOutputItemDoneEvent):
+                if isinstance(openai_chunk.item, ResponseFunctionWebSearch):
+                    self._add_child_spans([openai_chunk.item])
+            elif isinstance(openai_chunk, ResponseCreatedEvent):
+                resp = openai_chunk.response
+                span = trace.get_current_span()
+                span.set_attribute("model", str(resp.model))
+                span.set_attribute("reasoning_effort", str(resp.reasoning))
+                span.set_attribute("temperature", str(resp.temperature))
+
+        logger.info("Streaming complete: %d chunks, %d characters", chunk_count, len(full_text))
+        if not full_text:
+            logger.warning("No text collected during streaming")
+        return full_text
+
     def _add_child_spans(self, web_search_responses: list[ResponseFunctionWebSearch]) -> None:
         for tool_call in web_search_responses:
             with phoenix_utils.tracer().start_as_current_span(  # pylint: disable=not-context-manager,unexpected-keyword-arg
@@ -422,6 +386,11 @@ def run(self, email: str, resources_dict: dict, action_plan_dict: dict) -> dict:
 
         # Validate that at least one type of content is provided
         if not has_resources and not has_action_plan:
+            logger.error(
+                "EmailResponses: No content to email resources_dict=%r, action_plan_dict=%r",
+                resources_dict,
+                action_plan_dict,
+            )
             raise ValueError(
                 "At least one of resources_dict or action_plan_dict must contain valid data. "
                 f"Received resources_dict={bool(resources_dict)}, action_plan_dict={bool(action_plan_dict)}"
@@ -441,11 +410,16 @@ def run(self, email: str, resources_dict: dict, action_plan_dict: dict) -> dict:
         message_parts = [EMAIL_INTRO]
 
         if has_resources:
-            formatted_resources = format_resources(resources_dict.get("resources", []))
+            formatted_resources = "\n\n".join(
+                [
+                    self._format_resource(resource)
+                    for resource in resources_dict.get("resources", [])
+                ]
+            )
             message_parts.append(formatted_resources)
 
         if has_action_plan:
-            formatted_action_plan = format_action_plan(action_plan_dict)
+            formatted_action_plan = self._format_action_plan(action_plan_dict)
             message_parts.append(formatted_action_plan)
 
         message = "\n\n".join(message_parts)
@@ -465,11 +439,41 @@ def run(self, email: str, resources_dict: dict, action_plan_dict: dict) -> dict:
         logger.info("Email send status: %s", status)
         return {"status": status, "email": email, "message": message}
 
+    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']))}",
+            ]
+        )
+
+    def _format_action_plan(self, 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)
+
 
 BaseModelT = TypeVar("BaseModelT", bound=BaseModel)
 
 
-# TODO: Replace with https://docs.haystack.deepset.ai/docs/jsonschemavalidator
+# Consider replacing this with https://docs.haystack.deepset.ai/docs/jsonschemavalidator
 @component
 class LlmOutputValidator:
     def __init__(self, pydantic_model: type[BaseModelT]):