Skip to content

Revert "fixing research tool" #56

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
SumanthPal merged 1 commit into from
Feb 11, 2026
Merged

Revert "fixing research tool" #56

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
52 changes: 17 additions & 35 deletions backend/app/agents/executor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -186,9 +186,8 @@ def _build_agent_executor(self) -> AgentExecutor:
name="research_token",
func=self._execute_research_token_wrapper,
description=(
"Research a token for news and market sentiment. Uses a shared cache — "
"calling it is cheap, so use it whenever you need research. "
"Input: token symbol and optional recency (e.g., 'ETH 7d' for last 7 days)"
"Research a token for news and any market sentiments, "
"Input: token symbol and recency (e.g., 'ETH 7d' for last 7 days)"
),
),
Tool(
Expand Down Expand Up @@ -358,7 +357,7 @@ def _build_agent_executor(self) -> AgentExecutor:
- Do not create duplicate plans - cancel outdated ones first
- DOING NOTHING IS VALID: If no plans are due and market conditions don't warrant action, it's perfectly fine to take no action this cycle. Don't trade just to trade.
- Before making significant trades, research tokens using the research_token tool to check recent news and sentiment
- The research_token tool uses a shared cache automatically. Just call it when you need research.
- If recent research already exists, you may reuse it instead of researching again

TOOLS:
------
Expand Down Expand Up @@ -580,57 +579,40 @@ def _get_technical_indicator_wrapper(self, input_str: str) -> str:
return error_msg
def _execute_research_token_wrapper(self, input_str: str) -> str:
"""
Cache-first research wrapper.
Wrapper for executing research token from LangChain tool.

1. Parse input -> token + recency
2. Check shared cache via database_tool.get_fresh_research()
3. Cache HIT -> return cached summary (zero API cost)
4. Cache MISS -> call Perplexity, derive opinion, upsert to cache
Args:
input_str: "ETH" or "ETH 7d"

Returns:
Research summary string
"""
from .tools.research_tool import research_result_to_dict, derive_opinion
from .tools.research_tool import research_result_to_dict

try:
parts = input_str.split()
if len(parts) < 1:
return "Error: Invalid format. Expected 'TOKEN' or 'TOKEN RECENCY'"

token = parts[0].upper()
_valid_recencies = {"1d", "7d", "30d", "365d"}
raw_recency = parts[1] if len(parts) > 1 else "7d"
recency = raw_recency if raw_recency in _valid_recencies else "7d"

# ── Cache check ──────────────────────────────────────────
if self.database_tool:
cached = self._run_async(
self.database_tool.get_fresh_research(token, recency)
)
if cached is not None:
logger.info(f"Research cache HIT for {token} — skipping API call")
return cached["summary_markdown"]
recency = parts[1] if len(parts) > 1 else "7d"

# ── Cache MISS — call Perplexity ─────────────────────────
# Call sync research method
research_result = self.research_tool.research_token(
token_symbol=token,
recency=recency,
)

# Derive opinion from summary
opinion = derive_opinion(research_result.summary_markdown)

# Persist to shared cache (skip raw_results to keep rows lean)
if self.database_tool:
# Persist result to DB (convert dataclass to dict)
if self.database_tool and self.agent_uuid:
result_dict = research_result_to_dict(research_result)
self._run_async(
self.database_tool.upsert_research_result(
crypto_token=token,
self.database_tool.save_research_result(
agent_uuid=self.agent_uuid,
query=f"Research {token}",
summary_markdown=result_dict["summary_markdown"],
citations=result_dict["citations"],
result=result_dict,
recency=recency,
provider=result_dict.get("provider", "perplexity"),
related_tokens=[token],
agent_opinion=opinion,
last_researched_by=self.agent_uuid,
)
)

Expand Down
2 changes: 0 additions & 2 deletions backend/app/agents/tools/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,6 @@
from .tweet_post_tool import TweetPostTool
from .database_tool import DatabaseTool
from .plan_tool import PlanTool
from .research_tool import ResearchTool

# Deprecated: These tools are kept for backwards compatibility
# Use MakeTradeTool instead for simulated trading
Expand All @@ -30,7 +29,6 @@
"TweetPostTool",
"DatabaseTool",
"PlanTool",
"ResearchTool",
# Deprecated
"TradeTool",
"PortfolioTool",
Expand Down
204 changes: 62 additions & 142 deletions backend/app/agents/tools/database_tool.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@
import logging
from typing import Optional, Dict, Any, List
from uuid import UUID
from datetime import datetime, timezone
from datetime import datetime, timezone, timedelta
from decimal import Decimal

from sqlalchemy.ext.asyncio import AsyncSession
Expand All @@ -20,19 +20,6 @@

logger = logging.getLogger(__name__)

# Freshness thresholds per recency tier (seconds)
FRESHNESS_THRESHOLDS = {
"1d": 2 * 3600, # 2 hours
"7d": 6 * 3600, # 6 hours
"30d": 24 * 3600, # 24 hours
"365d": 72 * 3600, # 72 hours
}
DEFAULT_FRESHNESS = 6 * 3600 # 6 hours

# Tightness ordering — lower = tighter (more recent Perplexity search window).
# A cache row fetched with "7d" should NOT satisfy a "1d" request.
_RECENCY_TIGHTNESS = {"1d": 0, "7d": 1, "30d": 2, "365d": 3}


class DatabaseTool:
"""
Expand Down Expand Up @@ -415,139 +402,72 @@ async def create_agent_if_not_exists(
logger.error(f"Failed to create agent: {e}")
raise

# ── Shared research cache ──────────────────────────────────────────

async def get_fresh_research(
self, crypto_token: str, recency: str = "7d"
) -> Optional[Dict]:
"""
Return cached research for *crypto_token* if it is still fresh.

Freshness is determined by comparing the row's `updated_at` against
the threshold for the requested *recency* tier.

Returns:
Dict with cached research data if fresh, None if stale/missing.
"""
async def save_research_result(
self,
agent_uuid: UUID,
query: str,
result: Dict,
recency: Optional[str] = None,
related_tokens: Optional[List[str]] = None
) -> None:
#Saving the research to Postgres (agent_research_artifact table)
try:
stmt = select(AgentResearchArtifact).where(
AgentResearchArtifact.crypto_token == crypto_token
)
result = await self.session.execute(stmt)
artifact = result.scalar_one_or_none()

if artifact is None:
logger.info(f"Research cache MISS for {crypto_token} (no row)")
return None

# Tightness check: cached "7d" data can't satisfy a "1d" request
cached_tight = _RECENCY_TIGHTNESS.get(artifact.recency, 99)
requested_tight = _RECENCY_TIGHTNESS.get(recency, 1)
if cached_tight > requested_tight:
logger.info(
f"Research cache MISS for {crypto_token} "
f"(cached recency={artifact.recency} too loose for requested={recency})"
)
return None

age_seconds = (
datetime.now(timezone.utc) - artifact.updated_at
).total_seconds()
threshold = FRESHNESS_THRESHOLDS.get(recency, DEFAULT_FRESHNESS)

if age_seconds > threshold:
logger.info(
f"Research cache STALE for {crypto_token} "
f"(age={age_seconds:.0f}s, threshold={threshold}s)"
)
return None

logger.info(
f"Research cache HIT for {crypto_token} "
f"(age={age_seconds:.0f}s, threshold={threshold}s)"
stmt = insert(AgentResearchArtifact).values(
agent_id=agent_uuid,
query=query,
recency=recency,
provider=result.get("provider", "perplexity"),
summary_markdown=result["summary_markdown"],
citations=result.get("citations", []),
raw_results=result.get("raw_results", {}),
related_tokens=related_tokens or [],
created_at=datetime.now(timezone.utc),
)
return {
"crypto_token": artifact.crypto_token,
"query": artifact.query,
"recency": artifact.recency,
"provider": artifact.provider,
"summary_markdown": artifact.summary_markdown,
"citations": artifact.citations,
"related_tokens": artifact.related_tokens,
"agent_opinion": artifact.agent_opinion,
"updated_at": artifact.updated_at.isoformat(),
}

await self.session.execute(stmt)
await self.session.commit()
logger.info(f"Research result saved for agent={agent_uuid}, query='{query}'")
except Exception as e:
logger.error(f"Failed to check research cache: {e}")
raise

async def upsert_research_result(
self,
crypto_token: str,
query: str,
summary_markdown: str,
citations: List[Dict],
recency: Optional[str] = None,
provider: str = "perplexity",
raw_results: Optional[Dict] = None,
related_tokens: Optional[List[str]] = None,
agent_opinion: Optional[str] = None,
last_researched_by: Optional[UUID] = None,
) -> None:
"""
Insert or update the shared research row for *crypto_token*.

Uses INSERT ... ON CONFLICT (crypto_token) DO UPDATE so every token
has exactly one row that gets refreshed in place.
"""
now = datetime.now(timezone.utc)
await self.session.rollback()
logger.error(f"Failed to save research result: {e}")
raise

async def get_recent_research_results(
self,
agent_uuid: UUID,
token: Optional[str] = None,
days: int = 7
) -> List[Dict]:
#Fetch the recent research data for an agent from Postgres
try:
stmt = (
insert(AgentResearchArtifact)
.values(
crypto_token=crypto_token,
last_researched_by=last_researched_by,
query=query,
recency=recency,
provider=provider,
summary_markdown=summary_markdown,
citations=citations,
raw_results=raw_results,
related_tokens=related_tokens,
agent_opinion=agent_opinion,
created_at=now,
updated_at=now,
)
.on_conflict_do_update(
index_elements=["crypto_token"],
set_={
"last_researched_by": last_researched_by,
"query": query,
"recency": recency,
"provider": provider,
"summary_markdown": summary_markdown,
"citations": citations,
"raw_results": raw_results,
"related_tokens": related_tokens,
"agent_opinion": agent_opinion,
"updated_at": now,
},
)
)

await self.session.execute(stmt)
await self.session.commit()
logger.info(
f"Research upserted for {crypto_token} "
f"(by={last_researched_by}, opinion={agent_opinion})"
cutoff_date = datetime.now(timezone.utc) - timedelta(days=days)
stmt = select(AgentResearchArtifact).where(
AgentResearchArtifact.agent_id == agent_uuid,
AgentResearchArtifact.created_at >= cutoff_date
)

if token:
stmt = stmt.where(AgentResearchArtifact.related_tokens.contains([token]))

result = await self.session.execute(stmt)
artifacts = result.scalars().all()

research_results = []
for artifact in artifacts:
research_results.append({
"query": artifact.query,
"recency": artifact.recency,
"provider": artifact.provider,
"summary_markdown": artifact.summary_markdown,
"citations": artifact.citations,
"raw_results": artifact.raw_results,
"created_at": artifact.created_at.isoformat(),
})

logger.info(f"Retrieved {len(research_results)} research results for agent={agent_uuid}")
return research_results

except Exception as e:
await self.session.rollback()
logger.error(f"Failed to upsert research result: {e}")
logger.error(f"Failed to retrieve research results: {e}")
raise

async def create_plan_item(self, agent_uuid: UUID, tournament_uuid: UUID, action_type: PlanActionEnum,
execute_at: datetime, payload, idempotency_key: str, max_attempts: int = 3) -> PlanItem:
stmt = (insert(PlanItem).values(agent_id = agent_uuid,
Expand Down Expand Up @@ -606,7 +526,7 @@ async def list_plan_items(self, agent_uuid: UUID, tournament_uuid: UUID,
async def cancel_plan_item(self, plan_item_id: UUID, reason: str | None = None) -> None:
values = {
"status": PlanStatusEnum.cancelled,
"updated_at": datetime.now(timezone.utc),
"updated_at": datetime.utcnow(),
}

if reason:
Expand All @@ -633,7 +553,7 @@ async def reschedule_plan_item(self, plan_item_id: UUID, new_execute_at: datetim
)
.values(
execute_at=new_execute_at,
updated_at=datetime.now(timezone.utc),
updated_at=datetime.utcnow(),
)
.returning(PlanItem)
)
Expand Down
Loading
Loading