Spaces:
Sleeping
Sleeping
File size: 3,103 Bytes
e5e35a3 08b77a1 e5e35a3 cc8beab e5e35a3 08b77a1 e5e35a3 cc8beab e5e35a3 6710fbe e5e35a3 6710fbe cc8beab e5e35a3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 | 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
|