llm_clients/manual.py

# llm_clients/manual.py
from typing import Generator, Any, Dict
from .base import LlmClient


class ManualClient(LlmClient):
    """
    A manual LLM client that prompts the user to enter responses manually.
    This is useful for testing output guardrails.
    """

    def __init__(self, config: Dict[str, Any], system_prompt: str):
        super().__init__(config, system_prompt)
        print("✅ Manual LLM Client initialized for output testing.")

    def generate_content(self, prompt: str) -> str:
        """Prompts the user to manually enter a response."""
        print(f"\n{'='*60}")
        print("📝 MANUAL OUTPUT MODE")
        print(f"{'='*60}")
        print(f"💭 Input prompt: {prompt}")
        print("\n🤖 Please enter the LLM output you want to test with output guardrails:")
        print("(Press Enter twice to finish your input)\n")
        
        lines = []
        empty_line_count = 0
        
        while True:
            try:
                line = input()
                if line == "":
                    empty_line_count += 1
                    if empty_line_count >= 2:
                        break
                    lines.append(line)
                else:
                    empty_line_count = 0
                    lines.append(line)
            except KeyboardInterrupt:
                print("\n❌ Input cancelled by user.")
                return "User cancelled input."
        
        response = "\n".join(lines).strip()
        if not response:
            response = "No output provided."
            
        print(f"\n✅ Captured output ({len(response)} characters)")
        return response

    def generate_content_stream(self, prompt: str) -> Generator[str, None, None]:
        """
        Prompts the user to manually enter a response and simulates streaming.
        """
        print(f"\n{'='*60}")
        print("📝 MANUAL OUTPUT MODE (Streaming)")
        print(f"{'='*60}")
        print(f"💭 Input prompt: {prompt}")
        print("\n🤖 Please enter the LLM output you want to test with output guardrails:")
        print("(Press Enter twice to finish your input)\n")
        
        lines = []
        empty_line_count = 0
        
        while True:
            try:
                line = input()
                if line == "":
                    empty_line_count += 1
                    if empty_line_count >= 2:
                        break
                    lines.append(line)
                else:
                    empty_line_count = 0
                    lines.append(line)
            except KeyboardInterrupt:
                print("\n❌ Input cancelled by user.")
                yield "User cancelled input."
                return
        
        full_response = "\n".join(lines).strip()
        if not full_response:
            full_response = "No output provided."
            
        print(f"\n✅ Captured output ({len(full_response)} characters)")
        print("\n🔄 Simulating streaming output for guardrail testing...")
        
        # Simulate streaming by yielding words with small delays
        import time
        words = full_response.split()
        
        for i, word in enumerate(words):
            if i == 0:
                yield word
            else:
                yield " " + word
            time.sleep(0.1)  # Small delay to simulate streaming
            
        # If there were newlines in the original, yield them at the end
        if "\n" in full_response:
            yield "\n"
    
    def _generate_content_impl(self, prompt: str) -> str:
        """Implementation for base class compatibility."""
        return self.generate_content(prompt)

    def _generate_content_stream_impl(self, prompt: str) -> Generator[str, None, None]:
        """Implementation for base class compatibility."""
        return self.generate_content_stream(prompt)