Spaces:
Sleeping
Sleeping
| import logging | |
| from typing import Dict, List | |
| from dotenv import load_dotenv | |
| from langchain_openai import ChatOpenAI | |
| from src.core.AgentCommand import AgentCommand | |
| from src.core.FinanceState import FinanceState | |
| from src.core.errors import add_error | |
| from src.core.settings import get_settings | |
| logger = logging.getLogger(__name__) | |
| class GoalPlanningAgent(AgentCommand): | |
| """ | |
| Education-focused goal planning. Produces a plan and assumptions without directing trades. | |
| """ | |
| def __init__(self, state: FinanceState): | |
| load_dotenv() | |
| self.state = state | |
| settings = get_settings() | |
| self.client = ChatOpenAI(model=settings.models.agent_model) | |
| def process(self): | |
| trace_id = str(self.state.get("trace_id") or "") | |
| logger.info("[trace=%s] GoalPlanningAgent.start", trace_id) | |
| user_query = self.state.get("user_query", "") | |
| user_profile = self.state.get("user_profile", {}) or {} | |
| history: List[Dict[str, str]] = self.state.get("conversation_history", []) or [] | |
| system = ( | |
| "You are a financial education assistant. You must NOT provide personalized financial advice, " | |
| "trade instructions, or guarantees. Provide general education and planning frameworks only. " | |
| "Always include a short disclaimer: 'Educational only, not financial advice.' " | |
| "Be conservative and highlight risks and assumptions." | |
| ) | |
| # Keep history short to avoid blowing context. We only pass the last few turns. | |
| recent = history[-6:] if isinstance(history, list) else [] | |
| history_text = "\n".join( | |
| [f"{m.get('role','')}: {m.get('content','')}" for m in recent] | |
| ).strip() | |
| prompt = f""" | |
| Create an education-only financial goal plan for the user. | |
| User profile (may be incomplete): | |
| {user_profile} | |
| Recent conversation: | |
| {history_text if history_text else "(none)"} | |
| User request: | |
| {user_query} | |
| Output format: | |
| 1) Quick disclaimer line | |
| 2) Clarifying questions (max 5) if needed | |
| 3) A simple plan with assumptions: | |
| - Goal | |
| - Time horizon | |
| - Monthly/annual contribution estimate approach (no promises) | |
| - Asset allocation ranges by risk level (broad ranges only) | |
| - What to track monthly | |
| 4) Risks and what could change | |
| """ | |
| try: | |
| msg = self.client.invoke( | |
| [ | |
| {"role": "system", "content": system}, | |
| {"role": "user", "content": prompt}, | |
| ], | |
| temperature=0.2, | |
| ) | |
| self.state["response"] = msg.content | |
| except Exception as e: | |
| logger.exception("[trace=%s] GoalPlanningAgent failed", trace_id) | |
| add_error( | |
| self.state, | |
| code="goal_planning_error", | |
| message=str(e), | |
| agent="goal_planning_agent", | |
| ) | |
| self.state["response"] = ( | |
| "Educational only, not financial advice.\n\n" | |
| f"I couldn't generate a goal plan due to an internal error: {e}" | |
| ) | |
| return self.state | |