works

.gitignore

@@ -0,0 +1 @@
+.env

app.py

@@ -1,244 +1,314 @@
 import os
 import re
-import streamlit as st
-from dotenv import load_dotenv
-
 import io
 import time
 import json
 import queue
 import logging
-from PIL import Image
+from typing import Any, Generator, Optional, List, Dict, Tuple
+from dataclasses import dataclass
 
-# ------------------------
-# LangSmith imports
-# ------------------------
+import streamlit as st
+from dotenv import load_dotenv
+from PIL import Image
 import openai
 from langsmith.wrappers import wrap_openai
 from langsmith import traceable
 
 # ------------------------
-# Configure logging (optional but recommended)
+# Configuration and Types
 # ------------------------
-def init_logging():
+@dataclass
+class AppConfig:
+    """Application configuration settings."""
+    page_title: str = "Solution Specifier A"
+    page_icon: str = "🤖"
+    layout: str = "centered"
+    
+@dataclass
+class Message:
+    """Chat message structure."""
+    role: str
+    content: str
+
+class StreamingError(Exception):
+    """Custom exception for streaming-related errors."""
+    pass
+
+# ------------------------
+# Logging Configuration
+# ------------------------
+def setup_logging() -> logging.Logger:
+    """Configure and return the application logger."""
     logging.basicConfig(
         format="[%(asctime)s] %(levelname)+8s: %(message)s",
         level=logging.INFO,
     )
-    return logging.getLogger()
+    return logging.getLogger(__name__)
 
-logger = init_logging()
+logger = setup_logging()
 
 # ------------------------
-# Load environment variables
+# Environment Setup
 # ------------------------
-load_dotenv()
-api_key = os.getenv("OPENAI_API_KEY")
-assistant_id = os.getenv("ASSISTANT_ID_SOLUTION_SPECIFIER_A")  # The assistant we want to call
-
-if not api_key or not assistant_id:
-    raise RuntimeError("Please set OPENAI_API_KEY and ASSISTANT_ID_SOLUTION_SPECIFIER_A in your environment")
+class EnvironmentManager:
+    """Manages environment variables and configuration."""
+    
+    @staticmethod
+    def load_environment() -> Tuple[str, str]:
+        """Load and validate environment variables."""
+        load_dotenv(override=True)
+        api_key = os.getenv("OPENAI_API_KEY")
+        assistant_id = os.getenv("ASSISTANT_ID_SOLUTION_SPECIFIER_A")
+        
+        if not api_key or not assistant_id:
+            raise RuntimeError(
+                "Missing required environment variables. Please set "
+                "OPENAI_API_KEY and ASSISTANT_ID_SOLUTION_SPECIFIER_A"
+            )
+        
+        return api_key, assistant_id
 
 # ------------------------
-# Wrap the OpenAI client for LangSmith traceability
+# State Management
 # ------------------------
-openai_client = openai.Client(api_key=api_key)
-client = wrap_openai(openai_client)
+class StateManager:
+    """Manages Streamlit session state."""
+    
+    @staticmethod
+    def initialize_state() -> None:
+        """Initialize session state variables."""
+        if "messages" not in st.session_state:
+            st.session_state.messages = []
+        if "thread" not in st.session_state:
+            st.session_state.thread = None
+        if "tool_requests" not in st.session_state:
+            st.session_state.tool_requests = queue.Queue()
+        if "run_stream" not in st.session_state:
+            st.session_state.run_stream = None
+
+    @staticmethod
+    def add_message(role: str, content: str) -> None:
+        """Add a message to the conversation history."""
+        st.session_state.messages.append(Message(role=role, content=content))
 
 # ------------------------
-# Streamlit session state
+# Text Processing
 # ------------------------
-if "messages" not in st.session_state:
-    st.session_state["messages"] = []
-
-if "thread" not in st.session_state:
-    st.session_state["thread"] = None
-
-if "tool_requests" not in st.session_state:
-    st.session_state["tool_requests"] = queue.Queue()
-
-tool_requests = st.session_state["tool_requests"]
+class TextProcessor:
+    """Handles text processing and formatting."""
+    
+    @staticmethod
+    def remove_citations(text: str) -> str:
+        """Remove citation markers from text."""
+        pattern = r"【\d+†\w+】"
+        return re.sub(pattern, "📚", text)
 
 # ------------------------
-# Utility to remove citations like: 【12†somefile】
-# You can adapt to your own "annotations" handling if needed
+# Streaming Handler
 # ------------------------
-def remove_citation(text: str) -> str:
-    pattern = r"【\d+†\w+】"
-    return re.sub(pattern, "📚", text)
+class StreamHandler:
+    """Handles streaming of assistant responses."""
+    
+    def __init__(self, client: Any):
+        self.client = client
+        self.text_processor = TextProcessor()
+    
+    def stream_data(self) -> Generator[Any, None, None]:
+        """Stream data from the assistant run."""
+        st.toast("Thinking...", icon="🤔")
+        content_produced = False
+
+        try:
+            for event in st.session_state.run_stream:
+                match event.event:
+                    case "thread.message.delta":
+                        yield from self._handle_message_delta(event, content_produced)
+                    case "thread.run.requires_action":
+                        yield from self._handle_action_request(event, content_produced)
+                    case "thread.run.failed":
+                        logger.error(f"Run failed: {event}")
+                        raise StreamingError(f"Assistant run failed: {event}")
+                    
+            st.toast("Completed", icon="✅")
+        except Exception as e:
+            logger.error(f"Streaming error: {e}")
+            st.error(f"An error occurred while streaming: {str(e)}")
+            raise
+
+    def _handle_message_delta(self, event: Any, content_produced: bool) -> Generator[Any, None, None]:
+        """Handle message delta events."""
+        content = event.data.delta.content[0]
+        match content.type:
+            case "text":
+                yield self.text_processor.remove_citations(content.text.value)
+            case "image_file":
+                image_content = io.BytesIO(self.client.files.content(content.image_file.file_id).read())
+                yield Image.open(image_content)
+
+    def _handle_action_request(self, event: Any, content_produced: bool) -> Generator[str, None, None]:
+        """Handle action request events."""
+        logger.info(f"[Tool Request] {event}")
+        st.session_state.tool_requests.put(event)
+        if not content_produced:
+            yield "[Processing function call...]"
 
 # ------------------------
-# Helper: data streamer for text & images
-# Adapted from the Medium article approach
-# to handle text deltas, images, or function calls
+# Tool Request Handler
 # ------------------------
-def data_streamer():
-    """
-    Streams data from the assistant run. Yields text or images
-    and enqueues tool requests (function calls) to tool_requests.
-    """
-    st.toast("Thinking...", icon=":material/emoji_objects:")
-    content_produced = False
-
-    for event in st.session_state["run_stream"]:
-        match event.event:
-            case "thread.message.delta":
-                # A chunk of text or an image
-                content = event.data.delta.content[0]
-                match content.type:
-                    case "text":
-                        text_value = content.text.value
-                        content_produced = True
-                        # Optionally remove citations, etc.
-                        yield remove_citation(text_value)
-
-                    case "image_file":
-                        # If the assistant returns an image
-                        file_id = content.image_file.file_id
-                        content_produced = True
-                        image_content = io.BytesIO(client.files.content(file_id).read())
-                        yield Image.open(image_content)
-
-            case "thread.run.requires_action":
-                # The assistant is requesting a function call
-                logger.info(f"[Tool Request] {event}")
-                tool_requests.put(event)
-                if not content_produced:
-                    # We can yield a placeholder if the model hasn't said anything yet
-                    yield "[LLM is requesting a function call]"
-                return
-
-            case "thread.run.failed":
-                # The run failed for some reason
-                logger.error(f"Run failed: {event}")
-                return
-
-    # If we successfully streamed everything
-    st.toast("Completed", icon=":material/emoji_objects:")
-
-# ------------------------
-# Helper: display the streaming content
-# This wraps data_streamer in st.write_stream
-# so you can see partial tokens in real-time
-# ------------------------
-def display_stream(run_stream, create_context=True):
-    """
-    Grabs tokens from data_streamer() and displays them in real-time.
-    If `create_context=True`, messages are displayed as an assistant block.
-    """
-    st.session_state["run_stream"] = run_stream
-    if create_context:
-        with st.chat_message("assistant"):
-            streamed_result = st.write_stream(data_streamer)
-    else:
-        streamed_result = st.write_stream(data_streamer)
-
-    # Return whatever the final token stream is
-    return streamed_result
-
-# ------------------------
-# Example of handling a function call (requires_action)
-# If your Assistant uses function calling (e.g. code interpreter),
-# you'd parse arguments, run the function, and return output here.
-# ------------------------
-def handle_tool_request(event):
-    """
-    Demonstrates how you might handle a function call.
-    In practice, you'd parse the arguments from the event
-    and run your custom logic. Then return outputs as JSON.
-    """
-    st.toast("Running a function (this is user-defined code)", icon=":material/function:")
-    tool_outputs = []
-    data = event.data
-    for tool_call in data.required_action.submit_tool_outputs.tool_calls:
-        if tool_call.function.arguments:
-            function_args = json.loads(tool_call.function.arguments)
-        else:
-            function_args = {}
-
+class ToolRequestHandler:
+    """Handles tool requests from the assistant."""
+    
+    @staticmethod
+    def handle_request(event: Any) -> Tuple[List[Dict[str, str]], str, str]:
+        """Process tool requests and return outputs."""
+        st.toast("Processing function call...", icon="⚙️")
+        tool_outputs = []
+        data = event.data
+        
+        for tool_call in data.required_action.submit_tool_outputs.tool_calls:
+            output = ToolRequestHandler._process_tool_call(tool_call)
+            tool_outputs.append(output)
+            
+        return tool_outputs, data.thread_id, data.id
+    
+    @staticmethod
+    def _process_tool_call(tool_call: Any) -> Dict[str, str]:
+        """Process individual tool calls."""
+        function_args = json.loads(tool_call.function.arguments) if tool_call.function.arguments else {}
+        
         match tool_call.function.name:
             case "hello_world":
-                # Example: implement a user-defined function
                 name = function_args.get("name", "anonymous")
-                time.sleep(2)  # Simulate a long-running function
                 output_val = f"Hello, {name}! This was from a local function."
-                tool_outputs.append({"tool_call_id": tool_call.id, "output": output_val})
             case _:
-                # If unknown function name
-                msg = {"status": "error", "message": "Unknown function request."}
-                tool_outputs.append({"tool_call_id": tool_call.id, "output": json.dumps(msg)})
-    return tool_outputs, data.thread_id, data.id
-
-# ------------------------
-# Main chat logic
-# ------------------------
-@traceable  # Make this function traceable via LangSmith
-def generate_assistant_reply(user_input: str):
-    """
-    1. If no thread exists, create a new one.
-    2. Insert user message into the thread.
-    3. Use the Assistants API to create a run + stream the response.
-    4. If the assistant requests a function call, handle it and stream again.
-    """
-    # Create or retrieve thread
-    if not st.session_state["thread"]:
-        st.session_state["thread"] = client.beta.threads.create()
-    thread = st.session_state["thread"]
-
-    # Add user message to the thread
-    client.beta.threads.messages.create(
-        thread_id=thread.id,
-        role="user",
-        content=user_input
-    )
+                output_val = json.dumps({"status": "error", "message": "Unknown function request."})
+                
+        return {"tool_call_id": tool_call.id, "output": output_val}
 
-    # Start streaming assistant response
-    with client.beta.threads.runs.stream(
-        thread_id=thread.id,
-        assistant_id=assistant_id,
-    ) as run_stream:
-        display_stream(run_stream)
-
-    # If the assistant requested any tool calls, handle them now
-    while not tool_requests.empty():
-        event = tool_requests.get()
-        tool_outputs, t_id, run_id = handle_tool_request(event)
-        # Submit tool outputs
-        with client.beta.threads.runs.submit_tool_outputs_stream(
-            thread_id=t_id, run_id=run_id, tool_outputs=tool_outputs
-        ) as next_stream:
-            display_stream(next_stream, create_context=False)
+# ------------------------
+# Assistant Manager
+# ------------------------
+class AssistantManager:
+    """Manages interactions with the OpenAI Assistant."""
+    
+    def __init__(self, client: Any, assistant_id: str):
+        self.client = client
+        self.assistant_id = assistant_id
+        self.stream_handler = StreamHandler(client)
+        self.tool_handler = ToolRequestHandler()
+    
+    @traceable
+    def generate_reply(self, user_input: str) -> None:
+        """Generate and stream assistant's reply."""
+        # Ensure thread exists
+        if not st.session_state.thread:
+            st.session_state.thread = self.client.beta.threads.create()
+        
+        # Add user message
+        self.client.beta.threads.messages.create(
+            thread_id=st.session_state.thread.id,
+            role="user",
+            content=user_input
+        )
+        
+        # Stream initial response
+        with self.client.beta.threads.runs.stream(
+            thread_id=st.session_state.thread.id,
+            assistant_id=self.assistant_id,
+        ) as run_stream:
+            self._display_stream(run_stream)
+        
+        # Handle any tool requests
+        self._process_tool_requests()
+    
+    def _display_stream(self, run_stream: Any, create_context: bool = True) -> None:
+        """Display streaming content."""
+        st.session_state.run_stream = run_stream
+        if create_context:
+            with st.chat_message("assistant"):
+                st.write_stream(self.stream_handler.stream_data)
+        else:
+            st.write_stream(self.stream_handler.stream_data)
+    
+    def _process_tool_requests(self) -> None:
+        """Process any pending tool requests."""
+        while not st.session_state.tool_requests.empty():
+            event = st.session_state.tool_requests.get()
+            tool_outputs, thread_id, run_id = self.tool_handler.handle_request(event)
+            
+            with self.client.beta.threads.runs.submit_tool_outputs_stream(
+                thread_id=thread_id,
+                run_id=run_id,
+                tool_outputs=tool_outputs
+            ) as next_stream:
+                self._display_stream(next_stream, create_context=False)
 
 # ------------------------
-# Streamlit UI
+# Main Application
 # ------------------------
-def main():
-    st.set_page_config(page_title="Solution Specifier A", layout="centered")
-    st.title("Solution Specifier A")
-
-    # Display existing conversation
-    for msg in st.session_state["messages"]:
-        with st.chat_message(msg["role"]):
-            st.write(msg["content"])
-
-    user_input = st.chat_input("Type your message here...")
-    if user_input:
-        # Show user's message
-        with st.chat_message("user"):
-            st.write(user_input)
-
-        # Keep in session state
-        st.session_state["messages"].append({"role": "user", "content": user_input})
-
-        # Generate assistant reply
-        generate_assistant_reply(user_input)
-
-        # In a real app, you might keep track of the final text
-        # from the streamed tokens. For simplicity, we store
-        # the entire streamed result as one block in session state:
-        st.session_state["messages"].append(
-            {"role": "assistant", "content": "[assistant reply streamed above]"}
+class ChatApplication:
+    """Main chat application class."""
+    
+    def __init__(self):
+        self.config = AppConfig()
+        api_key, assistant_id = EnvironmentManager.load_environment()
+        
+        # Initialize OpenAI client
+        openai_client = openai.Client(api_key=api_key)
+        self.client = wrap_openai(openai_client)
+        
+        # Initialize components
+        self.state_manager = StateManager()
+        self.assistant_manager = AssistantManager(self.client, assistant_id)
+    
+    def setup_page(self) -> None:
+        """Configure the Streamlit page."""
+        st.set_page_config(
+            page_title=self.config.page_title,
+            page_icon=self.config.page_icon,
+            layout=self.config.layout
         )
+        st.title(self.config.page_title)
+    
+    def display_chat_history(self) -> None:
+        """Display the chat history."""
+        for msg in st.session_state.messages:
+            with st.chat_message(msg.role):
+                st.write(msg.content)
+    
+    def run(self) -> None:
+        """Run the chat application."""
+        self.setup_page()
+        self.state_manager.initialize_state()
+        self.display_chat_history()
+        
+        user_input = st.chat_input("Type your message here...")
+        if user_input:
+            # Display and store user message
+            with st.chat_message("user"):
+                st.write(user_input)
+            self.state_manager.add_message("user", user_input)
+            
+            # Generate and display assistant reply
+            try:
+                self.assistant_manager.generate_reply(user_input)
+                self.state_manager.add_message(
+                    "assistant",
+                    "[Assistant reply streamed above]"
+                )
+            except Exception as e:
+                st.error(f"Error generating response: {str(e)}")
+                logger.exception("Error in assistant reply generation")
+
+def main():
+    """Application entry point."""
+    try:
+        app = ChatApplication()
+        app.run()
+    except Exception as e:
+        st.error(f"Application error: {str(e)}")
+        logger.exception("Fatal application error")
 
 if __name__ == "__main__":
     main()

requirements.txt

@@ -1,4 +1,59 @@
-python-dotenv
-langsmith
-openai
-Pillow
+altair==5.5.0
+annotated-types==0.7.0
+anyio==4.8.0
+attrs==25.1.0
+blinker==1.9.0
+cachetools==5.5.2
+certifi==2025.1.31
+charset-normalizer==3.4.1
+click==8.1.8
+distro==1.9.0
+gitdb==4.0.12
+GitPython==3.1.44
+h11==0.14.0
+httpcore==1.0.7
+httpx==0.28.1
+idna==3.10
+Jinja2==3.1.5
+jiter==0.8.2
+jsonschema==4.23.0
+jsonschema-specifications==2024.10.1
+langsmith==0.3.10
+markdown-it-py==3.0.0
+MarkupSafe==3.0.2
+mdurl==0.1.2
+narwhals==1.28.0
+numpy==2.2.3
+openai==1.64.0
+orjson==3.10.15
+packaging==24.2
+pandas==2.2.3
+pillow==11.1.0
+protobuf==5.29.3
+pyarrow==19.0.1
+pydantic==2.10.6
+pydantic_core==2.27.2
+pydeck==0.9.1
+Pygments==2.19.1
+python-dateutil==2.9.0.post0
+python-dotenv==1.0.1
+pytz==2025.1
+referencing==0.36.2
+requests==2.32.3
+requests-toolbelt==1.0.0
+rich==13.9.4
+rpds-py==0.23.1
+setuptools==75.8.0
+six==1.17.0
+smmap==5.0.2
+sniffio==1.3.1
+streamlit==1.42.2
+tenacity==9.0.0
+toml==0.10.2
+tornado==6.4.2
+tqdm==4.67.1
+typing_extensions==4.12.2
+tzdata==2025.1
+urllib3==2.3.0
+wheel==0.45.1
+zstandard==0.23.0