import json
import os
from typing import Dict, List
import requests
from agentopera.engine.types.msg_context import MessageContext
from agentopera.engine.agent import message_handler
from agentopera.models.openai import OpenAIChatCompletionClient
from agentopera.chatflow.agents import AssistantAgent
from agentopera.chatflow.messages import TextMessage
from agentopera.engine.agent.routed_agent import RoutedAgent
from agentopera.engine.types.models import ModelInfo
from agentopera.engine.types.models import ModelFamily
from agentopera.engine.function_call import FunctionTool
from agentopera.utils.streaming_utils import stream_chunks_and_publish
from ...utils.tools.onchain_meme_trending_tool import fetch_onchain_trending
from ...utils.logger import logger
[docs]
class OnchainTrendAgent(RoutedAgent):
"""A specialized agent for conducting onchain trend analysis via an external API."""
def __init__(self, name: str) -> None:
super().__init__("Onchain Trend Agent for onchain trend analysis")
self.name = name
"""init the agent instance"""
# Initialize the model client
self.model_name = "Qwen2.5-72B-Instruct"
model_info: ModelInfo = {"vision": False, "function_calling": True, "json_output": False, "family": ModelFamily.UNKNOWN}
model_client = OpenAIChatCompletionClient(
base_url=os.getenv("TENSOROPERA_API_GATEWAY"),
api_key=os.environ['LLM_GATEWAY_KEY'],
model_info=model_info,
model=self.model_name,
temperature=0.6,
top_p=0.95,
)
trend_data_tool = FunctionTool(fetch_onchain_trending, description="Get sorted onchain data from CoinGecko")
self.agent = AssistantAgent(
name="onchain_trend_agent",
model_client=model_client,
system_message="You are a financial assistant specialized in cryptocurrency market analysis.\
You are limited to recommending **at most 20 coins**. If the user asks for more than 20, you must **only return 20 coins** and clearly state:\
**“I only care about the top 20 coins for quality and relevance.”**\
You must use the `fetch_onchain_trending` tool with the following parameter:\
- `number_of_coins` (str): the number of cryptocurrencies to recommend. Must be between 1 and 20.\
Recommendation rules:\
- If the user asks for coin recommendations **without specifying a number**, set `number_of_coins = '10'`.\
- If the user asks for **more than 20**, set `number_of_coins = '20'` and return only 20 coins. Be sure to include this note:\
When presenting results:\
- Start with: **“Here are {number_of_coins} coins with their descriptions:”**\
- Only include each coin's **Name** and write a short description for each coin.\
- **Do NOT include any other information** such as price, market cap, volume, or rankings.\
- Format the output as a clean, organized list using plain language in **native language**, in one or more natural paragraphs.\
- Always use the tool result to generate your response.\
- If the tool returns fewer coins than requested, just return what's available without explanation.",
model_client_stream=True, # Enable streaming tokens from the model client.
tools=[trend_data_tool],
reflect_on_tool_use=True
)
# for user role, each user has a unique chat agent instance
logger.info(f"OnchainTrendAgent initialized. agent id = {self.id}")
[docs]
@message_handler
async def my_message_handler(self, message: TextMessage, ctx: MessageContext) -> None:
"""Handles messages, performing deep research and streaming responses."""
assert ctx.message_channel is not None
logger.info(f"OnchainTrendAgent received messages. model name = {self.model_name}")
messages = [message] # Ensure we at least process the latest message
session_id = message.metadata.get("session_id")
if not session_id:
raise ValueError("Message metadata is missing `session_id` field")
await stream_chunks_and_publish(self.agent, ctx, messages, self.publish_message, self.name, session_id, self.cancel_token)