Spaces:
Sleeping
Sleeping
| import logging | |
| from pathlib import Path | |
| from typing import Any, Dict, List, Optional | |
| from dotenv import load_dotenv | |
| from langchain_openai import ChatOpenAI | |
| from src.core.settings import get_settings | |
| logger = logging.getLogger(__name__) | |
| class ResponseGenerator: | |
| """ | |
| Shared response-generation helper for agents. | |
| Supports: | |
| - Injecting "prompt_content" (agent-specific format rules, constraints, etc.) | |
| - Injecting guardrails from `src/data/guardrail_prompts.md` | |
| """ | |
| def __init__( | |
| self, | |
| model: str | None = None, | |
| include_guardrails: bool = True, | |
| guardrails_path: Optional[str] = None, | |
| ): | |
| load_dotenv() | |
| settings = get_settings() | |
| chosen_model = model or settings.models.agent_model | |
| self.client = ChatOpenAI(model=chosen_model) | |
| self.include_guardrails = include_guardrails | |
| self.guardrails_text = "" | |
| if include_guardrails: | |
| if guardrails_path: | |
| path = Path(guardrails_path) | |
| else: | |
| # src/core/ResponseGenerator.py -> src/ | |
| src_dir = Path(__file__).resolve().parents[1] | |
| path = src_dir / "data" / "guardrail_prompts.md" | |
| try: | |
| self.guardrails_text = path.read_text(encoding="utf-8").strip() | |
| except Exception: | |
| logger.exception("Failed reading guardrails from %s", str(path)) | |
| self.guardrails_text = "" | |
| def generate( | |
| self, | |
| *, | |
| user_query: str, | |
| context_blocks: Optional[List[str]] = None, | |
| system_preamble: str = "", | |
| prompt_content: str = "", | |
| temperature: float = 0.2, | |
| extra_metadata: Optional[Dict[str, Any]] = None, | |
| ) -> str: | |
| """ | |
| Generate a response using a consistent system prompt + optional guardrails + prompt_content. | |
| - system_preamble: short role instructions ("education only", etc.) | |
| - prompt_content: agent-specific formatting/rules to append (what the user asked for) | |
| - context_blocks: retrieved excerpts/snippets | |
| """ | |
| system_parts: List[str] = [] | |
| if system_preamble: | |
| system_parts.append(system_preamble.strip()) | |
| if self.include_guardrails and self.guardrails_text: | |
| system_parts.append("Guardrails (must follow):\n" + self.guardrails_text) | |
| system = "\n\n".join(system_parts).strip() or "You are a helpful assistant." | |
| ctx = "\n\n".join([c for c in (context_blocks or []) if c]).strip() | |
| user_parts: List[str] = [] | |
| if prompt_content: | |
| user_parts.append(prompt_content.strip()) | |
| user_parts.append(f"User question:\n{user_query.strip()}") | |
| if ctx: | |
| user_parts.append("Context:\n" + ctx) | |
| if extra_metadata: | |
| user_parts.append(f"Metadata:\n{extra_metadata}") | |
| user = "\n\n".join(user_parts).strip() | |
| logger.info( | |
| "ResponseGenerator: include_guardrails=%s prompt_content_len=%d context_blocks=%d", | |
| self.include_guardrails, | |
| len(prompt_content or ""), | |
| len(context_blocks or []), | |
| ) | |
| msg = self.client.invoke( | |
| [{"role": "system", "content": system}, {"role": "user", "content": user}], | |
| temperature=temperature, | |
| ) | |
| return str(msg.content or "").strip() | |