mirror of
https://github.com/arkorty/B.Tech-Project-III.git
synced 2026-04-19 12:41:48 +00:00
124 lines
5.1 KiB
Python
124 lines
5.1 KiB
Python
from features.base_feature import BaseFeature
|
|
from tools.tavily_search import TavilySearchTool
|
|
|
|
_tavily = TavilySearchTool()
|
|
|
|
|
|
class RoommateFeature(BaseFeature):
|
|
|
|
async def get_context(self, preferences_a: dict, preferences_b: dict, user_a_id: int = None, user_b_id: int = None) -> str:
|
|
"""
|
|
Fetch real product/plan options via Tavily (e.g., actual WiFi plans,
|
|
furniture prices) so agents propose real options.
|
|
"""
|
|
raw_a = preferences_a.get("raw_details", {})
|
|
raw_b = preferences_b.get("raw_details", {})
|
|
|
|
decision_type = (
|
|
raw_a.get("decision_type") or raw_b.get("decision_type")
|
|
or "shared living decision"
|
|
)
|
|
city = (
|
|
raw_a.get("city") or raw_b.get("city")
|
|
or raw_a.get("location") or raw_b.get("location")
|
|
or "India"
|
|
)
|
|
budget_a = raw_a.get("budget") or raw_a.get("max_budget") or ""
|
|
budget_b = raw_b.get("budget") or raw_b.get("max_budget") or ""
|
|
|
|
# Build a Tavily query based on decision type
|
|
if "wifi" in str(decision_type).lower() or "internet" in str(decision_type).lower():
|
|
query = f"best WiFi broadband plans {city} 2026 price speed"
|
|
elif "furniture" in str(decision_type).lower():
|
|
query = f"furniture prices India 2026 online shopping"
|
|
elif "chore" in str(decision_type).lower() or "cleaning" in str(decision_type).lower():
|
|
query = f"chore schedule roommates fair division strategies"
|
|
else:
|
|
query = f"{decision_type} options India 2026"
|
|
|
|
search_text = ""
|
|
try:
|
|
result = await _tavily.execute(query)
|
|
answer = result.get("answer", "")
|
|
results = result.get("results", [])[:4]
|
|
parts = []
|
|
if answer:
|
|
parts.append(f"Summary: {answer[:300]}")
|
|
for r in results:
|
|
title = r.get("title", "")
|
|
content = r.get("content", "")[:150]
|
|
if title:
|
|
parts.append(f" • {title}: {content}")
|
|
search_text = "\n".join(parts)
|
|
except Exception as e:
|
|
search_text = f"Search unavailable. Use your knowledge of {decision_type} options in {city}."
|
|
|
|
lines = [
|
|
"ROOMMATE DECISION DOMAIN RULES:",
|
|
"• Only propose options (plans, products) that appear in the real data below.",
|
|
"• Budget ceiling = lower of both parties' stated budgets.",
|
|
"• Unequal cost splits need usage-based justification.",
|
|
"• Both parties must stay within their stated budget constraints.",
|
|
"• If no option satisfies both budgets, propose cheapest viable option + fair split.",
|
|
]
|
|
if decision_type:
|
|
lines.append(f"\nDecision type: {decision_type}")
|
|
if city:
|
|
lines.append(f"Location: {city}")
|
|
if budget_a:
|
|
lines.append(f"Person A max budget: ₹{budget_a}/month")
|
|
if budget_b:
|
|
lines.append(f"Person B max budget: ₹{budget_b}/month")
|
|
|
|
if search_text:
|
|
lines.append(f"\nREAL OPTIONS from web search (only propose from this list):\n{search_text}")
|
|
|
|
return "\n".join(lines)
|
|
|
|
def format_resolution(
|
|
self, resolution: dict, preferences_a: dict, preferences_b: dict
|
|
) -> str:
|
|
status = resolution.get("status", "resolved")
|
|
final = resolution.get("final_proposal", {})
|
|
details = final.get("details", {})
|
|
rounds = resolution.get("rounds_taken", "?")
|
|
summary = resolution.get("summary", "")
|
|
|
|
raw_a = preferences_a.get("raw_details", {})
|
|
raw_b = preferences_b.get("raw_details", {})
|
|
decision_type = raw_a.get("decision_type") or raw_b.get("decision_type") or "Decision"
|
|
|
|
if status == "escalated":
|
|
return (
|
|
f"⚠️ *{decision_type.title()} — Human Decision Needed*\n\n"
|
|
f"_{summary}_\n\n"
|
|
f"Agents proposed options but couldn't finalize in {rounds} round(s)."
|
|
)
|
|
|
|
chosen = (
|
|
details.get("chosen_option") or details.get("plan") or details.get("option")
|
|
or details.get("decision") or final.get("summary", "")
|
|
)
|
|
cost = details.get("monthly_cost") or details.get("cost") or details.get("price") or ""
|
|
split = details.get("split") or details.get("each_pays") or ""
|
|
rules = details.get("rules") or details.get("terms") or []
|
|
|
|
lines = [f"🏠 *{decision_type.title()} — Decision Made!*\n"]
|
|
if chosen:
|
|
lines.append(f"✅ *Choice:* {chosen}")
|
|
if cost:
|
|
lines.append(f"💰 *Cost:* ₹{cost}/month")
|
|
if split:
|
|
lines.append(f"💳 *Each pays:* ₹{split}")
|
|
if rules and isinstance(rules, list):
|
|
lines.append("📋 *Agreed rules:*")
|
|
for rule in rules[:4]:
|
|
lines.append(f" • {rule}")
|
|
elif rules:
|
|
lines.append(f"📋 *Terms:* {rules}")
|
|
lines.append(f"\n⏱ Decided in {rounds} round(s)")
|
|
if summary and summary != "Agreement reached":
|
|
lines.append(f"_{summary}_")
|
|
|
|
return "\n".join(lines)
|