zhanggangyong/agent_management
1
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 添加链上数据分析 Agent

Browse Source 
新增两个区块链数据分析 Agent：

1. Chain Explorer Agent - 链上数据查询
   - 查询地址余额
   - 查询交易记录
   - 查询代币信息
   - 智能对话功能

2. Chain Analysis Agent - 链上数据分析
   - 地址活动分析
   - 交易模式分析
   - 资金流向分析
   - 合约交互分析
   - 智能分析对话

支持的区块链：
- Ethereum, BSC, Polygon, Arbitrum, Optimism, Base

使用 Etherscan V2 API，支持通过 Header 传递 API Key
This commit is contained in:
zhanggangyong
2026-02-05 18:29:56 +00:00
parent abe3f690ec
commit ca6a30bdab
7 changed files with 2223 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent_templates/agents/chain_analysis_agent/chain_analysis_agent.Dockerfile
+39
View File
@@ -0,0 +1,39 @@
# Chain Analysis Agent Dockerfile
# 链上数据分析 Agent - 分析地址活动、交易模式、资金流向
FROM python:3.11-slim
WORKDIR /app
# 安装系统依赖
RUN apt-get update && apt-get install -y \
curl \
&& rm -rf /var/lib/apt/lists/*
# 复制 common 模块
COPY common/ ./common/
# 复制 Agent 代码
COPY chain_analysis_agent.py .
COPY requirements.txt .
# 安装 Python 依赖
RUN pip install --no-cache-dir -r requirements.txt
# 环境变量
ENV PYTHONUNBUFFERED=1
ENV SERVICE_HOST=0.0.0.0
ENV SERVICE_PORT=8000
ENV POD_NAME=chain-analysis-agent
ENV LLM_BASE_URL=https://litellm.graystone-fb459c5d.southeastasia.azurecontainerapps.io/v1
ENV LLM_MODEL=taiji/gpt-4o-mini
# 健康检查
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
CMD curl -f http://localhost:8000/health || exit 1
# 暴露端口
EXPOSE 8000
# 运行
CMD ["python", "chain_analysis_agent.py"]
agent_templates/agents/chain_analysis_agent/chain_analysis_agent.py
+874
View File
@@ -0,0 +1,874 @@
"""
Chain Analysis Agent - 链上数据分析 Agent
分析区块链地址活动、交易模式、资金流向、合约交互等
支持 Ethereum, BSC, Polygon 等 EVM 兼容链
"""
import os
import sys
import logging
import aiohttp
from typing import Optional, List, Dict, Any
from datetime import datetime, timedelta
from collections import defaultdict
from fastapi import FastAPI, HTTPException, Query, Header, Request
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
import uvicorn
# 添加 common 模块路径
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# 导入回调工具
try:
from common.agent_callback_utils import AgentCallbackHandler, CallbackContextManager
CALLBACK_ENABLED = True
except ImportError:
CALLBACK_ENABLED = False
AgentCallbackHandler = None
CallbackContextManager = None
# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# 环境变量
SERVICE_HOST = os.getenv("SERVICE_HOST", "0.0.0.0")
SERVICE_PORT = int(os.getenv("SERVICE_PORT", "8080"))
POD_NAME = os.getenv("POD_NAME", "chain-analysis-agent")
USER_ID = os.getenv("USER_ID", "")
# LLM 配置
LLM_BASE_URL = os.getenv("LLM_BASE_URL", "https://litellm.graystone-fb459c5d.southeastasia.azurecontainerapps.io/v1")
LLM_MODEL = os.getenv("LLM_MODEL", "taiji/gpt-4o-mini")
# 支持的区块链网络配置 (Etherscan V2 API)
CHAIN_CONFIGS = {
"ethereum": {
"name": "Ethereum",
"symbol": "ETH",
"decimals": 18,
"chainid": 1,
"api_url": "https://api.etherscan.io/v2/api",
"explorer_url": "https://etherscan.io"
},
"bsc": {
"name": "BNB Smart Chain",
"symbol": "BNB",
"decimals": 18,
"chainid": 56,
"api_url": "https://api.etherscan.io/v2/api",
"explorer_url": "https://bscscan.com"
},
"polygon": {
"name": "Polygon",
"symbol": "POL",
"decimals": 18,
"chainid": 137,
"api_url": "https://api.etherscan.io/v2/api",
"explorer_url": "https://polygonscan.com"
},
"arbitrum": {
"name": "Arbitrum",
"symbol": "ETH",
"decimals": 18,
"chainid": 42161,
"api_url": "https://api.etherscan.io/v2/api",
"explorer_url": "https://arbiscan.io"
},
"optimism": {
"name": "Optimism",
"symbol": "ETH",
"decimals": 18,
"chainid": 10,
"api_url": "https://api.etherscan.io/v2/api",
"explorer_url": "https://optimistic.etherscan.io"
},
"base": {
"name": "Base",
"symbol": "ETH",
"decimals": 18,
"chainid": 8453,
"api_url": "https://api.etherscan.io/v2/api",
"explorer_url": "https://basescan.org"
}
}
# FastAPI 应用
app = FastAPI(
title="Chain Analysis Agent",
description="链上数据分析 - 分析地址活动、交易模式、资金流向",
version="1.0.0"
)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 回调处理器
callback_handler: Optional[AgentCallbackHandler] = None
# ==================== 请求/响应模型 ====================
class AddressAnalysisRequest(BaseModel):
"""地址分析请求"""
address: str = Field(..., description="钱包地址")
chain: str = Field("ethereum", description="区块链网络")
days: int = Field(30, ge=1, le=365, description="分析天数")
user_id: Optional[str] = Field(None, description="用户ID")
class TransactionPatternRequest(BaseModel):
"""交易模式分析请求"""
address: str = Field(..., description="钱包地址")
chain: str = Field("ethereum", description="区块链网络")
user_id: Optional[str] = Field(None, description="用户ID")
class FundFlowRequest(BaseModel):
"""资金流向分析请求"""
address: str = Field(..., description="钱包地址")
chain: str = Field("ethereum", description="区块链网络")
limit: int = Field(100, ge=10, le=500, description="交易数量")
user_id: Optional[str] = Field(None, description="用户ID")
class ContractInteractionRequest(BaseModel):
"""合约交互分析请求"""
address: str = Field(..., description="钱包地址")
chain: str = Field("ethereum", description="区块链网络")
user_id: Optional[str] = Field(None, description="用户ID")
class ChatRequest(BaseModel):
"""Chat 请求"""
message: str = Field(..., description="用户消息")
chain: str = Field("ethereum", description="默认区块链网络")
user_id: Optional[str] = Field(None, description="用户ID")
class ChatResponse(BaseModel):
"""Chat 响应"""
response: str
analysis: Optional[Dict[str, Any]] = None
timestamp: str
class HealthResponse(BaseModel):
"""健康检查响应"""
status: str
pod_name: str
supported_chains: List[str]
callback_enabled: bool
timestamp: str
# ==================== 生命周期 ====================
@app.on_event("startup")
async def startup_event():
"""应用启动时初始化回调处理器"""
global callback_handler
if CALLBACK_ENABLED and AgentCallbackHandler:
try:
callback_handler = AgentCallbackHandler(
agent_name=POD_NAME,
user_id=USER_ID
)
logger.info(f"回调处理器已初始化: agent={POD_NAME}, user={USER_ID}")
except Exception as e:
logger.warning(f"回调处理器初始化失败: {e}")
logger.info(f"Chain Analysis Agent 启动完成 - {POD_NAME}")
logger.info(f"支持的区块链: {list(CHAIN_CONFIGS.keys())}")
# ==================== 核心分析功能 ====================
async def fetch_all_transactions(address: str, chain: str, api_key: str, limit: int = 200) -> List[Dict]:
"""获取所有交易用于分析"""
if chain not in CHAIN_CONFIGS:
return []
config = CHAIN_CONFIGS[chain]
url = config["api_url"]
params = {
"chainid": config["chainid"],
"module": "account",
"action": "txlist",
"address": address,
"startblock": 0,
"endblock": 99999999,
"page": 1,
"offset": limit,
"sort": "desc",
"apikey": api_key
}
try:
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params, timeout=aiohttp.ClientTimeout(total=20)) as response:
if response.status == 200:
data = await response.json()
if data.get("status") == "1":
return data.get("result", [])
except Exception as e:
logger.error(f"获取交易失败: {e}")
return []
async def fetch_internal_transactions(address: str, chain: str, api_key: str) -> List[Dict]:
"""获取内部交易"""
if chain not in CHAIN_CONFIGS:
return []
config = CHAIN_CONFIGS[chain]
url = config["api_url"]
params = {
"chainid": config["chainid"],
"module": "account",
"action": "txlistinternal",
"address": address,
"startblock": 0,
"endblock": 99999999,
"page": 1,
"offset": 100,
"sort": "desc",
"apikey": api_key
}
try:
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params, timeout=aiohttp.ClientTimeout(total=15)) as response:
if response.status == 200:
data = await response.json()
if data.get("status") == "1":
return data.get("result", [])
except Exception as e:
logger.error(f"获取内部交易失败: {e}")
return []
async def fetch_balance(address: str, chain: str, api_key: str) -> float:
"""获取余额"""
if chain not in CHAIN_CONFIGS:
return 0.0
config = CHAIN_CONFIGS[chain]
url = config["api_url"]
params = {
"chainid": config["chainid"],
"module": "account",
"action": "balance",
"address": address,
"tag": "latest",
"apikey": api_key
}
try:
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params, timeout=aiohttp.ClientTimeout(total=10)) as response:
if response.status == 200:
data = await response.json()
if data.get("status") == "1":
balance_wei = int(data.get("result", 0))
return balance_wei / (10 ** config["decimals"])
except Exception as e:
logger.error(f"获取余额失败: {e}")
return 0.0
def analyze_address_activity(transactions: List[Dict], address: str, chain: str, days: int = 30) -> Dict[str, Any]:
"""分析地址活动"""
config = CHAIN_CONFIGS.get(chain, CHAIN_CONFIGS["ethereum"])
address_lower = address.lower()
now = datetime.utcnow()
cutoff = now - timedelta(days=days)
# 统计数据
total_sent = 0.0
total_received = 0.0
tx_count_in = 0
tx_count_out = 0
unique_addresses = set()
failed_tx = 0
daily_activity = defaultdict(lambda: {"in": 0, "out": 0, "count": 0})
for tx in transactions:
try:
timestamp = datetime.fromtimestamp(int(tx.get("timeStamp", 0)))
if timestamp < cutoff:
continue
value_wei = int(tx.get("value", 0))
value = value_wei / (10 ** config["decimals"])
day_key = timestamp.strftime("%Y-%m-%d")
daily_activity[day_key]["count"] += 1
if tx.get("isError") == "1":
failed_tx += 1
continue
from_addr = tx.get("from", "").lower()
to_addr = tx.get("to", "").lower()
if from_addr == address_lower:
# 发出
total_sent += value
tx_count_out += 1
daily_activity[day_key]["out"] += value
if to_addr:
unique_addresses.add(to_addr)
elif to_addr == address_lower:
# 收到
total_received += value
tx_count_in += 1
daily_activity[day_key]["in"] += value
unique_addresses.add(from_addr)
except Exception as e:
logger.error(f"解析交易失败: {e}")
# 计算活跃天数
active_days = len(daily_activity)
return {
"address": address,
"chain": chain,
"period_days": days,
"summary": {
"total_sent": round(total_sent, 6),
"total_received": round(total_received, 6),
"net_flow": round(total_received - total_sent, 6),
"tx_count_in": tx_count_in,
"tx_count_out": tx_count_out,
"total_tx": tx_count_in + tx_count_out,
"failed_tx": failed_tx,
"unique_addresses": len(unique_addresses),
"active_days": active_days
},
"symbol": config["symbol"],
"daily_activity": dict(sorted(daily_activity.items(), reverse=True)[:7]) # 最近7天
}
def analyze_transaction_patterns(transactions: List[Dict], address: str, chain: str) -> Dict[str, Any]:
"""分析交易模式"""
config = CHAIN_CONFIGS.get(chain, CHAIN_CONFIGS["ethereum"])
address_lower = address.lower()
# 时间分布
hourly_distribution = defaultdict(int)
daily_distribution = defaultdict(int)
# 金额分布
value_ranges = {
"micro": 0, # < 0.01
"small": 0, # 0.01 - 0.1
"medium": 0, # 0.1 - 1
"large": 0, # 1 - 10
"whale": 0 # > 10
}
# 交互地址频率
address_frequency = defaultdict(int)
# 交易间隔
timestamps = []
for tx in transactions:
try:
timestamp = datetime.fromtimestamp(int(tx.get("timeStamp", 0)))
timestamps.append(timestamp)
hourly_distribution[timestamp.hour] += 1
daily_distribution[timestamp.strftime("%A")] += 1
value_wei = int(tx.get("value", 0))
value = value_wei / (10 ** config["decimals"])
if value < 0.01:
value_ranges["micro"] += 1
elif value < 0.1:
value_ranges["small"] += 1
elif value < 1:
value_ranges["medium"] += 1
elif value < 10:
value_ranges["large"] += 1
else:
value_ranges["whale"] += 1
from_addr = tx.get("from", "").lower()
to_addr = tx.get("to", "").lower()
counterparty = to_addr if from_addr == address_lower else from_addr
if counterparty:
address_frequency[counterparty] += 1
except Exception as e:
logger.error(f"解析交易失败: {e}")
# 计算交易间隔
avg_interval = None
if len(timestamps) > 1:
timestamps.sort(reverse=True)
intervals = []
for i in range(len(timestamps) - 1):
interval = (timestamps[i] - timestamps[i+1]).total_seconds() / 3600 # 小时
intervals.append(interval)
avg_interval = round(sum(intervals) / len(intervals), 2)
# 前5个交互地址
top_addresses = sorted(address_frequency.items(), key=lambda x: x[1], reverse=True)[:5]
return {
"address": address,
"chain": chain,
"patterns": {
"hourly_distribution": dict(hourly_distribution),
"daily_distribution": dict(daily_distribution),
"value_distribution": value_ranges,
"avg_interval_hours": avg_interval,
"top_counterparties": [{"address": addr, "tx_count": count} for addr, count in top_addresses]
},
"behavior_summary": generate_behavior_summary(hourly_distribution, value_ranges, avg_interval)
}
def generate_behavior_summary(hourly: Dict, values: Dict, interval: Optional[float]) -> str:
"""生成行为摘要"""
summary_parts = []
# 活跃时段
if hourly:
peak_hour = max(hourly, key=hourly.get)
summary_parts.append(f"活跃高峰时段: {peak_hour}:00 UTC")
# 交易规模
total_tx = sum(values.values())
if total_tx > 0:
whale_ratio = values["whale"] / total_tx * 100
if whale_ratio > 20:
summary_parts.append("大额交易频繁（可能是机构或巨鲸）")
elif values["micro"] / total_tx > 0.5:
summary_parts.append("以小额交易为主（可能是频繁交易者或机器人）")
# 交易频率
if interval:
if interval < 1:
summary_parts.append("高频交易（可能是自动化程序）")
elif interval > 168: # 一周
summary_parts.append("低频交易（普通持有者）")
return "; ".join(summary_parts) if summary_parts else "交易模式正常"
def analyze_fund_flow(transactions: List[Dict], address: str, chain: str) -> Dict[str, Any]:
"""分析资金流向"""
config = CHAIN_CONFIGS.get(chain, CHAIN_CONFIGS["ethereum"])
address_lower = address.lower()
inflow = defaultdict(float) # 资金来源
outflow = defaultdict(float) # 资金去向
for tx in transactions:
try:
if tx.get("isError") == "1":
continue
value_wei = int(tx.get("value", 0))
value = value_wei / (10 ** config["decimals"])
if value == 0:
continue
from_addr = tx.get("from", "").lower()
to_addr = tx.get("to", "").lower()
if from_addr == address_lower and to_addr:
outflow[to_addr] += value
elif to_addr == address_lower:
inflow[from_addr] += value
except Exception as e:
logger.error(f"解析交易失败: {e}")
# 排序获取 Top 10
top_inflow = sorted(inflow.items(), key=lambda x: x[1], reverse=True)[:10]
top_outflow = sorted(outflow.items(), key=lambda x: x[1], reverse=True)[:10]
total_in = sum(inflow.values())
total_out = sum(outflow.values())
return {
"address": address,
"chain": chain,
"fund_flow": {
"total_inflow": round(total_in, 6),
"total_outflow": round(total_out, 6),
"net_flow": round(total_in - total_out, 6),
"inflow_sources": len(inflow),
"outflow_destinations": len(outflow),
"top_inflow": [
{"address": addr, "amount": round(amt, 6), "symbol": config["symbol"]}
for addr, amt in top_inflow
],
"top_outflow": [
{"address": addr, "amount": round(amt, 6), "symbol": config["symbol"]}
for addr, amt in top_outflow
]
},
"symbol": config["symbol"]
}
def analyze_contract_interactions(transactions: List[Dict], address: str, chain: str) -> Dict[str, Any]:
"""分析合约交互"""
config = CHAIN_CONFIGS.get(chain, CHAIN_CONFIGS["ethereum"])
address_lower = address.lower()
contract_interactions = defaultdict(lambda: {"count": 0, "methods": set(), "value": 0.0})
for tx in transactions:
try:
from_addr = tx.get("from", "").lower()
to_addr = tx.get("to", "").lower()
# 只分析发出的交易且有 input data 的（合约调用）
if from_addr != address_lower:
continue
input_data = tx.get("input", "")
if input_data and input_data != "0x" and len(input_data) >= 10:
method_id = input_data[:10]
value_wei = int(tx.get("value", 0))
value = value_wei / (10 ** config["decimals"])
contract_interactions[to_addr]["count"] += 1
contract_interactions[to_addr]["methods"].add(method_id)
contract_interactions[to_addr]["value"] += value
except Exception as e:
logger.error(f"解析交易失败: {e}")
# 排序
sorted_contracts = sorted(
contract_interactions.items(),
key=lambda x: x[1]["count"],
reverse=True
)[:10]
return {
"address": address,
"chain": chain,
"contract_interactions": {
"total_contracts": len(contract_interactions),
"top_contracts": [
{
"contract": addr,
"interaction_count": data["count"],
"unique_methods": len(data["methods"]),
"total_value": round(data["value"], 6),
"symbol": config["symbol"],
"explorer_url": f"{config['explorer_url']}/address/{addr}"
}
for addr, data in sorted_contracts
]
}
}
async def chat_with_llm(message: str, context: str, api_key: str) -> str:
"""调用 LLM 生成分析报告"""
try:
async with aiohttp.ClientSession() as session:
payload = {
"model": LLM_MODEL,
"messages": [
{
"role": "system",
"content": """你是一个资深的区块链数据分析师，擅长：
1. 分析钱包地址的链上行为模式
2. 识别交易特征（高频交易、巨鲸、机器人等）
3. 追踪资金流向和来源
4. 分析合约交互行为
5. 提供风险评估和投资建议
请根据链上数据提供专业、深入的分析报告，用简洁易懂的语言表达。"""
},
{
"role": "user",
"content": f"链上分析数据:\n{context}\n\n分析请求: {message}"
}
],
"max_tokens": 800,
"temperature": 0.7
}
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
async with session.post(
f"{LLM_BASE_URL}/chat/completions",
json=payload,
headers=headers,
timeout=aiohttp.ClientTimeout(total=30)
) as response:
if response.status == 200:
data = await response.json()
return data.get("choices", [{}])[0].get("message", {}).get("content", "抱歉，无法生成分析报告")
else:
error = await response.text()
logger.error(f"LLM 请求失败: {response.status} - {error}")
return f"LLM 服务错误: {response.status}"
except Exception as e:
logger.error(f"LLM 调用失败: {e}")
return f"分析失败: {str(e)}"
def extract_address_from_message(message: str) -> Optional[str]:
"""从消息中提取以太坊地址"""
import re
pattern = r'0x[a-fA-F0-9]{40}'
match = re.search(pattern, message)
return match.group(0) if match else None
# ==================== API 端点 ====================
@app.get("/", response_model=dict)
async def root():
"""服务状态"""
return {
"service": "Chain Analysis Agent",
"description": "链上数据分析 - 分析地址活动、交易模式、资金流向",
"status": "running",
"supported_chains": list(CHAIN_CONFIGS.keys()),
"tools": ["address_analysis", "transaction_patterns", "fund_flow", "contract_interactions", "chat"]
}
@app.get("/health", response_model=HealthResponse)
async def health_check():
"""健康检查"""
return HealthResponse(
status="healthy",
pod_name=POD_NAME,
supported_chains=list(CHAIN_CONFIGS.keys()),
callback_enabled=CALLBACK_ENABLED,
timestamp=datetime.utcnow().isoformat()
)
@app.post("/address-analysis")
async def address_analysis(
request: AddressAnalysisRequest,
api_key: Optional[str] = Header(None, alias="api-key"),
etherscan_key: Optional[str] = Header(None, alias="etherscan-key")
):
"""地址活动分析"""
scan_key = etherscan_key or api_key
if not scan_key:
raise HTTPException(status_code=401, detail="请在请求头中提供 etherscan-key 或 api-key")
transactions = await fetch_all_transactions(request.address, request.chain, scan_key)
if not transactions:
raise HTTPException(status_code=404, detail="未找到交易记录")
result = analyze_address_activity(transactions, request.address, request.chain, request.days)
balance = await fetch_balance(request.address, request.chain, scan_key)
result["current_balance"] = round(balance, 8)
return {
**result,
"timestamp": datetime.utcnow().isoformat()
}
@app.post("/transaction-patterns")
async def transaction_patterns(
request: TransactionPatternRequest,
api_key: Optional[str] = Header(None, alias="api-key"),
etherscan_key: Optional[str] = Header(None, alias="etherscan-key")
):
"""交易模式分析"""
scan_key = etherscan_key or api_key
if not scan_key:
raise HTTPException(status_code=401, detail="请在请求头中提供 etherscan-key 或 api-key")
transactions = await fetch_all_transactions(request.address, request.chain, scan_key)
if not transactions:
raise HTTPException(status_code=404, detail="未找到交易记录")
result = analyze_transaction_patterns(transactions, request.address, request.chain)
return {
**result,
"timestamp": datetime.utcnow().isoformat()
}
@app.post("/fund-flow")
async def fund_flow(
request: FundFlowRequest,
api_key: Optional[str] = Header(None, alias="api-key"),
etherscan_key: Optional[str] = Header(None, alias="etherscan-key")
):
"""资金流向分析"""
scan_key = etherscan_key or api_key
if not scan_key:
raise HTTPException(status_code=401, detail="请在请求头中提供 etherscan-key 或 api-key")
transactions = await fetch_all_transactions(request.address, request.chain, scan_key, request.limit)
if not transactions:
raise HTTPException(status_code=404, detail="未找到交易记录")
result = analyze_fund_flow(transactions, request.address, request.chain)
return {
**result,
"timestamp": datetime.utcnow().isoformat()
}
@app.post("/contract-interactions")
async def contract_interactions(
request: ContractInteractionRequest,
api_key: Optional[str] = Header(None, alias="api-key"),
etherscan_key: Optional[str] = Header(None, alias="etherscan-key")
):
"""合约交互分析"""
scan_key = etherscan_key or api_key
if not scan_key:
raise HTTPException(status_code=401, detail="请在请求头中提供 etherscan-key 或 api-key")
transactions = await fetch_all_transactions(request.address, request.chain, scan_key)
if not transactions:
raise HTTPException(status_code=404, detail="未找到交易记录")
result = analyze_contract_interactions(transactions, request.address, request.chain)
return {
**result,
"timestamp": datetime.utcnow().isoformat()
}
@app.post("/chat", response_model=ChatResponse)
async def chat(
request: ChatRequest,
api_key: Optional[str] = Header(None, alias="api-key"),
etherscan_key: Optional[str] = Header(None, alias="etherscan-key"),
llm_key: Optional[str] = Header(None, alias="llm-key"),
authorization: Optional[str] = Header(None)
):
"""智能对话 - 支持自然语言分析链上数据
api_key 通过请求头传递:
- api-key 或 etherscan-key: 区块链浏览器 API Key
- llm-key 或 Authorization: LLM API Key
"""
# 获取区块链 API Key
scan_key = etherscan_key or api_key
if not scan_key:
raise HTTPException(status_code=401, detail="请在请求头中提供 etherscan-key 或 api-key")
# 获取 LLM API Key
llm_api_key = llm_key
if not llm_api_key and authorization:
if authorization.startswith("Bearer "):
llm_api_key = authorization[7:]
else:
llm_api_key = authorization
if not llm_api_key:
raise HTTPException(status_code=401, detail="请在请求头中提供 llm-key 或 Authorization")
# 从消息中提取地址
address = extract_address_from_message(request.message)
analysis_data = {}
if address:
transactions = await fetch_all_transactions(address, request.chain, scan_key)
if transactions:
# 执行全面分析
analysis_data["activity"] = analyze_address_activity(transactions, address, request.chain)
analysis_data["patterns"] = analyze_transaction_patterns(transactions, address, request.chain)
analysis_data["fund_flow"] = analyze_fund_flow(transactions, address, request.chain)
analysis_data["contracts"] = analyze_contract_interactions(transactions, address, request.chain)
analysis_data["balance"] = await fetch_balance(address, request.chain, scan_key)
# 构建上下文
if analysis_data:
context_parts = []
if "activity" in analysis_data:
s = analysis_data["activity"]["summary"]
context_parts.append(f"地址: {address}")
context_parts.append(f"当前余额: {analysis_data['balance']:.6f} ETH")
context_parts.append(f"30天活动: 收入 {s['total_received']:.4f} ETH, 支出 {s['total_sent']:.4f} ETH")
context_parts.append(f"交易统计: 入账 {s['tx_count_in']} 笔, 出账 {s['tx_count_out']} 笔")
if "patterns" in analysis_data:
p = analysis_data["patterns"]
context_parts.append(f"行为特征: {p['behavior_summary']}")
if "fund_flow" in analysis_data:
f = analysis_data["fund_flow"]["fund_flow"]
context_parts.append(f"资金来源数: {f['inflow_sources']}, 去向数: {f['outflow_destinations']}")
if "contracts" in analysis_data:
c = analysis_data["contracts"]["contract_interactions"]
context_parts.append(f"交互合约数: {c['total_contracts']}")
context = "\n".join(context_parts)
else:
context = "未检测到有效的钱包地址，请提供 0x 开头的以太坊地址"
# 调用 LLM 生成分析报告
llm_response = await chat_with_llm(request.message, context, llm_api_key)
return ChatResponse(
response=llm_response,
analysis=analysis_data if analysis_data else {"detected_address": address},
timestamp=datetime.utcnow().isoformat()
)
@app.get("/chains")
async def list_chains():
"""列出支持的区块链"""
return {
"chains": [
{
"id": chain_id,
"name": config["name"],
"symbol": config["symbol"],
"explorer": config["explorer_url"]
}
for chain_id, config in CHAIN_CONFIGS.items()
]
}
# ==================== 主入口 ====================
def main():
"""主函数"""
logger.info(f"启动 Chain Analysis Agent - {POD_NAME}")
logger.info(f"支持的区块链: {list(CHAIN_CONFIGS.keys())}")
logger.info(f"回调功能: {'已启用' if CALLBACK_ENABLED else '未启用'}")
uvicorn.run(app, host=SERVICE_HOST, port=SERVICE_PORT, log_level="info")
if __name__ == "__main__":
main()
agent_templates/agents/chain_analysis_agent/requirements.txt
+6
View File
@@ -0,0 +1,6 @@
fastapi>=0.104.0
uvicorn>=0.24.0
aiohttp>=3.9.0
pydantic>=2.0.0
python-multipart>=0.0.6
httpx>=0.25.0
agent_templates/agents/chain_explorer_agent/chain_explorer_agent.Dockerfile
+39
View File
@@ -0,0 +1,39 @@
# Chain Explorer Agent Dockerfile
# 链上数据查询 Agent - 查询地址余额、交易记录、代币信息
FROM python:3.11-slim
WORKDIR /app
# 安装系统依赖
RUN apt-get update && apt-get install -y \
curl \
&& rm -rf /var/lib/apt/lists/*
# 复制 common 模块
COPY common/ ./common/
# 复制 Agent 代码
COPY chain_explorer_agent.py .
COPY requirements.txt .
# 安装 Python 依赖
RUN pip install --no-cache-dir -r requirements.txt
# 环境变量
ENV PYTHONUNBUFFERED=1
ENV SERVICE_HOST=0.0.0.0
ENV SERVICE_PORT=8000
ENV POD_NAME=chain-explorer-agent
ENV LLM_BASE_URL=https://litellm.graystone-fb459c5d.southeastasia.azurecontainerapps.io/v1
ENV LLM_MODEL=taiji/gpt-4o-mini
# 健康检查
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
CMD curl -f http://localhost:8000/health || exit 1
# 暴露端口
EXPOSE 8000
# 运行
CMD ["python", "chain_explorer_agent.py"]
agent_templates/agents/chain_explorer_agent/chain_explorer_agent.py
+613
View File
@@ -0,0 +1,613 @@
"""
Chain Explorer Agent - 链上数据查询 Agent
查询区块链地址余额、交易记录、代币信息等
支持 Ethereum, BSC, Polygon 等 EVM 兼容链
"""
import os
import sys
import logging
import aiohttp
from typing import Optional, List, Dict, Any
from datetime import datetime
from decimal import Decimal
from fastapi import FastAPI, HTTPException, Query, Header, Request
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
import uvicorn
# 添加 common 模块路径
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# 导入回调工具
try:
from common.agent_callback_utils import AgentCallbackHandler, CallbackContextManager
CALLBACK_ENABLED = True
except ImportError:
CALLBACK_ENABLED = False
AgentCallbackHandler = None
CallbackContextManager = None
# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# 环境变量
SERVICE_HOST = os.getenv("SERVICE_HOST", "0.0.0.0")
SERVICE_PORT = int(os.getenv("SERVICE_PORT", "8080"))
POD_NAME = os.getenv("POD_NAME", "chain-explorer-agent")
USER_ID = os.getenv("USER_ID", "")
# LLM 配置
LLM_BASE_URL = os.getenv("LLM_BASE_URL", "https://litellm.graystone-fb459c5d.southeastasia.azurecontainerapps.io/v1")
LLM_MODEL = os.getenv("LLM_MODEL", "taiji/gpt-4o-mini")
# 支持的区块链网络配置 (Etherscan V2 API)
CHAIN_CONFIGS = {
"ethereum": {
"name": "Ethereum",
"symbol": "ETH",
"decimals": 18,
"chainid": 1,
"api_url": "https://api.etherscan.io/v2/api",
"explorer_url": "https://etherscan.io"
},
"bsc": {
"name": "BNB Smart Chain",
"symbol": "BNB",
"decimals": 18,
"chainid": 56,
"api_url": "https://api.etherscan.io/v2/api",
"explorer_url": "https://bscscan.com"
},
"polygon": {
"name": "Polygon",
"symbol": "POL",
"decimals": 18,
"chainid": 137,
"api_url": "https://api.etherscan.io/v2/api",
"explorer_url": "https://polygonscan.com"
},
"arbitrum": {
"name": "Arbitrum",
"symbol": "ETH",
"decimals": 18,
"chainid": 42161,
"api_url": "https://api.etherscan.io/v2/api",
"explorer_url": "https://arbiscan.io"
},
"optimism": {
"name": "Optimism",
"symbol": "ETH",
"decimals": 18,
"chainid": 10,
"api_url": "https://api.etherscan.io/v2/api",
"explorer_url": "https://optimistic.etherscan.io"
},
"base": {
"name": "Base",
"symbol": "ETH",
"decimals": 18,
"chainid": 8453,
"api_url": "https://api.etherscan.io/v2/api",
"explorer_url": "https://basescan.org"
}
}
# FastAPI 应用
app = FastAPI(
title="Chain Explorer Agent",
description="链上数据查询 - 查询地址余额、交易记录、代币信息",
version="1.0.0"
)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 回调处理器
callback_handler: Optional[AgentCallbackHandler] = None
# ==================== 请求/响应模型 ====================
class BalanceRequest(BaseModel):
"""余额查询请求"""
address: str = Field(..., description="钱包地址")
chain: str = Field("ethereum", description="区块链网络: ethereum, bsc, polygon, arbitrum, optimism")
user_id: Optional[str] = Field(None, description="用户ID")
class BalanceResponse(BaseModel):
"""余额响应"""
address: str
chain: str
balance: str
balance_formatted: str
symbol: str
usd_value: Optional[float] = None
timestamp: str
class TransactionRequest(BaseModel):
"""交易查询请求"""
address: str = Field(..., description="钱包地址")
chain: str = Field("ethereum", description="区块链网络")
page: int = Field(1, ge=1, description="页码")
limit: int = Field(10, ge=1, le=100, description="每页数量")
user_id: Optional[str] = Field(None, description="用户ID")
class TokenBalanceRequest(BaseModel):
"""代币余额查询请求"""
address: str = Field(..., description="钱包地址")
chain: str = Field("ethereum", description="区块链网络")
user_id: Optional[str] = Field(None, description="用户ID")
class ChatRequest(BaseModel):
"""Chat 请求"""
message: str = Field(..., description="用户消息")
chain: str = Field("ethereum", description="默认区块链网络")
user_id: Optional[str] = Field(None, description="用户ID")
class ChatResponse(BaseModel):
"""Chat 响应"""
response: str
data: Optional[Dict[str, Any]] = None
timestamp: str
class HealthResponse(BaseModel):
"""健康检查响应"""
status: str
pod_name: str
supported_chains: List[str]
callback_enabled: bool
timestamp: str
# ==================== 生命周期 ====================
@app.on_event("startup")
async def startup_event():
"""应用启动时初始化回调处理器"""
global callback_handler
if CALLBACK_ENABLED and AgentCallbackHandler:
try:
callback_handler = AgentCallbackHandler(
agent_name=POD_NAME,
user_id=USER_ID
)
logger.info(f"回调处理器已初始化: agent={POD_NAME}, user={USER_ID}")
except Exception as e:
logger.warning(f"回调处理器初始化失败: {e}")
logger.info(f"Chain Explorer Agent 启动完成 - {POD_NAME}")
logger.info(f"支持的区块链: {list(CHAIN_CONFIGS.keys())}")
# ==================== 核心功能 ====================
async def fetch_balance(address: str, chain: str, api_key: str) -> Dict[str, Any]:
"""获取地址余额"""
if chain not in CHAIN_CONFIGS:
return {"success": False, "error": f"不支持的区块链: {chain}"}
config = CHAIN_CONFIGS[chain]
url = config["api_url"]
params = {
"chainid": config["chainid"],
"module": "account",
"action": "balance",
"address": address,
"tag": "latest",
"apikey": api_key
}
try:
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params, timeout=aiohttp.ClientTimeout(total=15)) as response:
if response.status == 200:
data = await response.json()
if data.get("status") == "1":
balance_wei = int(data.get("result", 0))
balance_eth = balance_wei / (10 ** config["decimals"])
return {
"success": True,
"address": address,
"chain": chain,
"chain_name": config["name"],
"balance_wei": str(balance_wei),
"balance": round(balance_eth, 8),
"symbol": config["symbol"],
"explorer_url": f"{config['explorer_url']}/address/{address}"
}
else:
return {"success": False, "error": data.get("message", "API 错误")}
else:
return {"success": False, "error": f"HTTP {response.status}"}
except Exception as e:
logger.error(f"获取余额失败: {e}")
return {"success": False, "error": str(e)}
async def fetch_transactions(address: str, chain: str, api_key: str, page: int = 1, limit: int = 10) -> Dict[str, Any]:
"""获取交易记录"""
if chain not in CHAIN_CONFIGS:
return {"success": False, "error": f"不支持的区块链: {chain}"}
config = CHAIN_CONFIGS[chain]
url = config["api_url"]
params = {
"chainid": config["chainid"],
"module": "account",
"action": "txlist",
"address": address,
"startblock": 0,
"endblock": 99999999,
"page": page,
"offset": limit,
"sort": "desc",
"apikey": api_key
}
try:
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params, timeout=aiohttp.ClientTimeout(total=15)) as response:
if response.status == 200:
data = await response.json()
if data.get("status") == "1":
transactions = []
for tx in data.get("result", []):
value_wei = int(tx.get("value", 0))
value_eth = value_wei / (10 ** config["decimals"])
transactions.append({
"hash": tx.get("hash"),
"block": tx.get("blockNumber"),
"timestamp": datetime.fromtimestamp(int(tx.get("timeStamp", 0))).isoformat(),
"from": tx.get("from"),
"to": tx.get("to"),
"value": round(value_eth, 8),
"symbol": config["symbol"],
"gas_used": tx.get("gasUsed"),
"gas_price": tx.get("gasPrice"),
"is_error": tx.get("isError") == "1",
"tx_url": f"{config['explorer_url']}/tx/{tx.get('hash')}"
})
return {
"success": True,
"address": address,
"chain": chain,
"transactions": transactions,
"count": len(transactions),
"page": page
}
else:
return {"success": False, "error": data.get("message", "API 错误")}
else:
return {"success": False, "error": f"HTTP {response.status}"}
except Exception as e:
logger.error(f"获取交易失败: {e}")
return {"success": False, "error": str(e)}
async def fetch_token_balances(address: str, chain: str, api_key: str) -> Dict[str, Any]:
"""获取 ERC20 代币余额"""
if chain not in CHAIN_CONFIGS:
return {"success": False, "error": f"不支持的区块链: {chain}"}
config = CHAIN_CONFIGS[chain]
url = config["api_url"]
params = {
"chainid": config["chainid"],
"module": "account",
"action": "tokentx",
"address": address,
"page": 1,
"offset": 100,
"sort": "desc",
"apikey": api_key
}
try:
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params, timeout=aiohttp.ClientTimeout(total=15)) as response:
if response.status == 200:
data = await response.json()
if data.get("status") == "1":
# 统计代币
token_map = {}
for tx in data.get("result", []):
contract = tx.get("contractAddress")
if contract not in token_map:
token_map[contract] = {
"contract": contract,
"name": tx.get("tokenName"),
"symbol": tx.get("tokenSymbol"),
"decimals": int(tx.get("tokenDecimal", 18)),
"tx_count": 0
}
token_map[contract]["tx_count"] += 1
tokens = list(token_map.values())
return {
"success": True,
"address": address,
"chain": chain,
"tokens": tokens,
"token_count": len(tokens)
}
else:
return {"success": True, "address": address, "chain": chain, "tokens": [], "token_count": 0}
else:
return {"success": False, "error": f"HTTP {response.status}"}
except Exception as e:
logger.error(f"获取代币失败: {e}")
return {"success": False, "error": str(e)}
async def chat_with_llm(message: str, context: str, api_key: str) -> str:
"""调用 LLM 生成响应"""
try:
async with aiohttp.ClientSession() as session:
payload = {
"model": LLM_MODEL,
"messages": [
{
"role": "system",
"content": """你是一个专业的区块链数据分析师。你可以：
1. 查询钱包地址的余额和交易记录
2. 分析地址的链上活动
3. 解答关于以太坊、BSC、Polygon等EVM链的问题
请根据提供的链上数据，用简洁专业的语言回答用户问题。"""
},
{
"role": "user",
"content": f"链上数据:\n{context}\n\n用户问题: {message}"
}
],
"max_tokens": 500,
"temperature": 0.7
}
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
async with session.post(
f"{LLM_BASE_URL}/chat/completions",
json=payload,
headers=headers,
timeout=aiohttp.ClientTimeout(total=30)
) as response:
if response.status == 200:
data = await response.json()
return data.get("choices", [{}])[0].get("message", {}).get("content", "抱歉，无法生成回复")
else:
error = await response.text()
logger.error(f"LLM 请求失败: {response.status} - {error}")
return f"LLM 服务错误: {response.status}"
except Exception as e:
logger.error(f"LLM 调用失败: {e}")
return f"调用失败: {str(e)}"
def extract_address_from_message(message: str) -> Optional[str]:
"""从消息中提取以太坊地址"""
import re
# 匹配以太坊地址格式 (0x开头，40个十六进制字符)
pattern = r'0x[a-fA-F0-9]{40}'
match = re.search(pattern, message)
return match.group(0) if match else None
# ==================== API 端点 ====================
@app.get("/", response_model=dict)
async def root():
"""服务状态"""
return {
"service": "Chain Explorer Agent",
"description": "链上数据查询 - 查询地址余额、交易记录、代币信息",
"status": "running",
"supported_chains": list(CHAIN_CONFIGS.keys()),
"tools": ["balance", "transactions", "tokens", "chat"]
}
@app.get("/health", response_model=HealthResponse)
async def health_check():
"""健康检查"""
return HealthResponse(
status="healthy",
pod_name=POD_NAME,
supported_chains=list(CHAIN_CONFIGS.keys()),
callback_enabled=CALLBACK_ENABLED,
timestamp=datetime.utcnow().isoformat()
)
@app.post("/balance")
async def get_balance(
request: BalanceRequest,
api_key: Optional[str] = Header(None, alias="api-key"),
etherscan_key: Optional[str] = Header(None, alias="etherscan-key")
):
"""查询地址余额"""
scan_key = etherscan_key or api_key
if not scan_key:
raise HTTPException(status_code=401, detail="请在请求头中提供 etherscan-key 或 api-key")
result = await fetch_balance(request.address, request.chain, scan_key)
if not result["success"]:
raise HTTPException(status_code=400, detail=result["error"])
return {
**result,
"timestamp": datetime.utcnow().isoformat()
}
@app.post("/transactions")
async def get_transactions(
request: TransactionRequest,
api_key: Optional[str] = Header(None, alias="api-key"),
etherscan_key: Optional[str] = Header(None, alias="etherscan-key")
):
"""查询交易记录"""
scan_key = etherscan_key or api_key
if not scan_key:
raise HTTPException(status_code=401, detail="请在请求头中提供 etherscan-key 或 api-key")
result = await fetch_transactions(request.address, request.chain, scan_key, request.page, request.limit)
if not result["success"]:
raise HTTPException(status_code=400, detail=result["error"])
return {
**result,
"timestamp": datetime.utcnow().isoformat()
}
@app.post("/tokens")
async def get_token_balances(
request: TokenBalanceRequest,
api_key: Optional[str] = Header(None, alias="api-key"),
etherscan_key: Optional[str] = Header(None, alias="etherscan-key")
):
"""查询代币余额"""
scan_key = etherscan_key or api_key
if not scan_key:
raise HTTPException(status_code=401, detail="请在请求头中提供 etherscan-key 或 api-key")
result = await fetch_token_balances(request.address, request.chain, scan_key)
if not result["success"]:
raise HTTPException(status_code=400, detail=result["error"])
return {
**result,
"timestamp": datetime.utcnow().isoformat()
}
@app.post("/chat", response_model=ChatResponse)
async def chat(
request: ChatRequest,
api_key: Optional[str] = Header(None, alias="api-key"),
etherscan_key: Optional[str] = Header(None, alias="etherscan-key"),
llm_key: Optional[str] = Header(None, alias="llm-key"),
authorization: Optional[str] = Header(None)
):
"""智能对话 - 支持自然语言查询链上数据
api_key 通过请求头传递:
- api-key: 区块链浏览器 API Key (Etherscan 等)
- etherscan-key: Etherscan API Key (优先)
- llm-key: LLM API Key (用于 AI 分析)
- Authorization: Bearer LLM-API-Key
"""
# 获取区块链 API Key
scan_key = etherscan_key or api_key
if not scan_key:
raise HTTPException(status_code=401, detail="请在请求头中提供 etherscan-key 或 api-key")
# 获取 LLM API Key
llm_api_key = llm_key
if not llm_api_key and authorization:
if authorization.startswith("Bearer "):
llm_api_key = authorization[7:]
else:
llm_api_key = authorization
if not llm_api_key:
raise HTTPException(status_code=401, detail="请在请求头中提供 llm-key 或 Authorization 用于 AI 分析")
# 从消息中提取地址
address = extract_address_from_message(request.message)
chain_data = {}
if address:
# 获取余额
balance_result = await fetch_balance(address, request.chain, scan_key)
if balance_result["success"]:
chain_data["balance"] = balance_result
# 获取最近交易
tx_result = await fetch_transactions(address, request.chain, scan_key, 1, 5)
if tx_result["success"]:
chain_data["recent_transactions"] = tx_result["transactions"][:5]
# 获取代币
token_result = await fetch_token_balances(address, request.chain, scan_key)
if token_result["success"]:
chain_data["tokens"] = token_result["tokens"][:10]
# 构建上下文
if chain_data:
context_parts = []
if "balance" in chain_data:
b = chain_data["balance"]
context_parts.append(f"地址: {b['address']}\n余额: {b['balance']} {b['symbol']} ({b['chain_name']})")
if "recent_transactions" in chain_data:
context_parts.append(f"最近交易数: {len(chain_data['recent_transactions'])}")
for tx in chain_data["recent_transactions"][:3]:
context_parts.append(f" - {tx['value']} {tx['symbol']} @ {tx['timestamp'][:10]}")
if "tokens" in chain_data:
context_parts.append(f"持有代币种类: {len(chain_data['tokens'])}")
context = "\n".join(context_parts)
else:
context = "未检测到有效的钱包地址，请提供 0x 开头的以太坊地址"
# 调用 LLM 生成回复
llm_response = await chat_with_llm(request.message, context, llm_api_key)
return ChatResponse(
response=llm_response,
data=chain_data if chain_data else {"detected_address": address},
timestamp=datetime.utcnow().isoformat()
)
@app.get("/chains")
async def list_chains():
"""列出支持的区块链"""
return {
"chains": [
{
"id": chain_id,
"name": config["name"],
"symbol": config["symbol"],
"explorer": config["explorer_url"]
}
for chain_id, config in CHAIN_CONFIGS.items()
]
}
# ==================== 主入口 ====================
def main():
"""主函数"""
logger.info(f"启动 Chain Explorer Agent - {POD_NAME}")
logger.info(f"支持的区块链: {list(CHAIN_CONFIGS.keys())}")
logger.info(f"回调功能: {'已启用' if CALLBACK_ENABLED else '未启用'}")
uvicorn.run(app, host=SERVICE_HOST, port=SERVICE_PORT, log_level="info")
if __name__ == "__main__":
main()
agent_templates/agents/chain_explorer_agent/requirements.txt
+6
View File
@@ -0,0 +1,6 @@
fastapi>=0.104.0
uvicorn>=0.24.0
aiohttp>=3.9.0
pydantic>=2.0.0
python-multipart>=0.0.6
httpx>=0.25.0
docs/CHAIN_AGENTS_DOC.md
+646
View File
@@ -0,0 +1,646 @@
# 链上数据分析 AI Agent 文档
---
本文档详细介绍了两个链上数据分析 Agent 的功能、API 接口和使用方法。
## 概述
| Agent | 功能 | 端口 |
|-------|------|------|
| Chain Explorer Agent | 链上数据查询 - 余额、交易、代币 | 8000 |
| Chain Analysis Agent | 链上数据分析 - 活动分析、交易模式、资金流向 | 8000 |
## 支持的区块链
| 网络 | Chain ID | 符号 | 说明 |
|------|----------|------|------|
| Ethereum | ethereum | ETH | 以太坊主网 |
| BSC | bsc | BNB | 币安智能链 |
| Polygon | polygon | POL | Polygon 网络 |
| Arbitrum | arbitrum | ETH | Arbitrum L2 |
| Optimism | optimism | ETH | Optimism L2 |
| Base | base | ETH | Coinbase L2 |
---
## 认证方式
所有 API 调用都需要通过请求头传递 API Key：
| Header | 说明 | 必需 |
|--------|------|------|
| `etherscan-key` | Etherscan API Key（区块链浏览器） | ✅ |
| `api-key` | 备选的区块链浏览器 API Key | ⭕ |
| `llm-key` | LLM API Key（用于 Chat 功能） | Chat 时必需 |
| `Authorization` | Bearer Token（LLM API Key） | Chat 时备选 |
### 示例
```bash
curl -X POST "http://agent-url/balance" \
-H "Content-Type: application/json" \
-H "etherscan-key: YOUR_ETHERSCAN_API_KEY" \
-d '{"address": "0x...", "chain": "ethereum"}'
```
---
# 1. Chain Explorer Agent - 链上数据查询
## 功能概览
| 端点 | 方法 | 功能 |
|------|------|------|
| `/` | GET | 服务状态 |
| `/health` | GET | 健康检查 |
| `/chains` | GET | 支持的区块链列表 |
| `/balance` | POST | 查询地址余额 |
| `/transactions` | POST | 查询交易记录 |
| `/tokens` | POST | 查询代币信息 |
| `/chat` | POST | 智能对话 |
---
## 1.1 查询地址余额
### 请求
```bash
POST /balance
Content-Type: application/json
etherscan-key: YOUR_API_KEY
```
### 参数
| 参数 | 类型 | 必需 | 默认值 | 说明 |
|------|------|------|--------|------|
| address | string | ✅ | - | 钱包地址 (0x开头) |
| chain | string | ❌ | ethereum | 区块链网络 |
### 示例
```bash
curl -X POST "http://localhost:8000/balance" \
-H "Content-Type: application/json" \
-H "etherscan-key: F71AZ6XW2WN6GK3D63HJ7AVAEDC9M42EZY" \
-d '{
"address": "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045",
"chain": "ethereum"
}'
```
### 响应
```json
{
"success": true,
"address": "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045",
"chain": "ethereum",
"chain_name": "Ethereum",
"balance_wei": "32116130289281011210",
"balance": 32.11613029,
"symbol": "ETH",
"explorer_url": "https://etherscan.io/address/0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045",
"timestamp": "2026-02-05T17:00:28.539769"
}
```
---
## 1.2 查询交易记录
### 请求
```bash
POST /transactions
Content-Type: application/json
etherscan-key: YOUR_API_KEY
```
### 参数
| 参数 | 类型 | 必需 | 默认值 | 说明 |
|------|------|------|--------|------|
| address | string | ✅ | - | 钱包地址 |
| chain | string | ❌ | ethereum | 区块链网络 |
| page | int | ❌ | 1 | 页码 |
| limit | int | ❌ | 10 | 每页数量 (1-100) |
### 示例
```bash
curl -X POST "http://localhost:8000/transactions" \
-H "Content-Type: application/json" \
-H "etherscan-key: F71AZ6XW2WN6GK3D63HJ7AVAEDC9M42EZY" \
-d '{
"address": "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045",
"chain": "ethereum",
"limit": 5
}'
```
### 响应
```json
{
"success": true,
"address": "0x...",
"chain": "ethereum",
"transactions": [
{
"hash": "0x5b0d81bab...",
"block": "21780123",
"timestamp": "2026-02-05T13:43:47",
"from": "0x...",
"to": "0x...",
"value": 0.000505,
"symbol": "ETH",
"gas_used": "21000",
"gas_price": "5000000000",
"is_error": false,
"tx_url": "https://etherscan.io/tx/0x..."
}
],
"count": 5,
"page": 1,
"timestamp": "2026-02-05T17:00:30.123456"
}
```
---
## 1.3 查询代币信息
### 请求
```bash
POST /tokens
Content-Type: application/json
etherscan-key: YOUR_API_KEY
```
### 参数
| 参数 | 类型 | 必需 | 默认值 | 说明 |
|------|------|------|--------|------|
| address | string | ✅ | - | 钱包地址 |
| chain | string | ❌ | ethereum | 区块链网络 |
### 示例
```bash
curl -X POST "http://localhost:8000/tokens" \
-H "Content-Type: application/json" \
-H "etherscan-key: F71AZ6XW2WN6GK3D63HJ7AVAEDC9M42EZY" \
-d '{
"address": "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045",
"chain": "ethereum"
}'
```
### 响应
```json
{
"success": true,
"address": "0x...",
"chain": "ethereum",
"tokens": [
{
"contract": "0x...",
"name": "Dogelon",
"symbol": "ELON",
"decimals": 18,
"tx_count": 5
}
],
"token_count": 49,
"timestamp": "2026-02-05T17:00:35.123456"
}
```
---
## 1.4 智能对话 (Chat)
### 请求
```bash
POST /chat
Content-Type: application/json
etherscan-key: YOUR_ETHERSCAN_KEY
llm-key: YOUR_LLM_API_KEY
```
### 参数
| 参数 | 类型 | 必需 | 默认值 | 说明 |
|------|------|------|--------|------|
| message | string | ✅ | - | 用户消息（包含地址） |
| chain | string | ❌ | ethereum | 默认区块链网络 |
### 示例
```bash
curl -X POST "http://localhost:8000/chat" \
-H "Content-Type: application/json" \
-H "etherscan-key: F71AZ6XW2WN6GK3D63HJ7AVAEDC9M42EZY" \
-H "llm-key: sk-xxx" \
-d '{
"message": "帮我查看 0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045 的余额和交易",
"chain": "ethereum"
}'
```
### 响应
```json
{
"response": "地址 0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045 目前的余额为 32.12 ETH...",
"data": {
"balance": { ... },
"recent_transactions": [ ... ],
"tokens": [ ... ]
},
"timestamp": "2026-02-05T17:01:00.123456"
}
```
---
# 2. Chain Analysis Agent - 链上数据分析
## 功能概览
| 端点 | 方法 | 功能 |
|------|------|------|
| `/` | GET | 服务状态 |
| `/health` | GET | 健康检查 |
| `/chains` | GET | 支持的区块链列表 |
| `/address-analysis` | POST | 地址活动分析 |
| `/transaction-patterns` | POST | 交易模式分析 |
| `/fund-flow` | POST | 资金流向分析 |
| `/contract-interactions` | POST | 合约交互分析 |
| `/chat` | POST | 智能分析对话 |
---
## 2.1 地址活动分析
分析指定时间段内的地址活动，包括收支统计、活跃度等。
### 请求
```bash
POST /address-analysis
Content-Type: application/json
etherscan-key: YOUR_API_KEY
```
### 参数
| 参数 | 类型 | 必需 | 默认值 | 说明 |
|------|------|------|--------|------|
| address | string | ✅ | - | 钱包地址 |
| chain | string | ❌ | ethereum | 区块链网络 |
| days | int | ❌ | 30 | 分析天数 (1-365) |
### 示例
```bash
curl -X POST "http://localhost:8000/address-analysis" \
-H "Content-Type: application/json" \
-H "etherscan-key: F71AZ6XW2WN6GK3D63HJ7AVAEDC9M42EZY" \
-d '{
"address": "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045",
"chain": "ethereum",
"days": 30
}'
```
### 响应
```json
{
"address": "0x...",
"chain": "ethereum",
"period_days": 30,
"summary": {
"total_sent": 1.0,
"total_received": 0.004591,
"net_flow": -0.995409,
"tx_count_in": 52,
"tx_count_out": 11,
"total_tx": 63,
"failed_tx": 2,
"unique_addresses": 31,
"active_days": 18
},
"symbol": "ETH",
"current_balance": 32.11613029,
"daily_activity": { ... },
"timestamp": "2026-02-05T17:02:00.123456"
}
```
---
## 2.2 交易模式分析
分析地址的交易行为模式，包括时间分布、金额分布、高频交互对手等。
### 请求
```bash
POST /transaction-patterns
Content-Type: application/json
etherscan-key: YOUR_API_KEY
```
### 参数
| 参数 | 类型 | 必需 | 默认值 | 说明 |
|------|------|------|--------|------|
| address | string | ✅ | - | 钱包地址 |
| chain | string | ❌ | ethereum | 区块链网络 |
### 示例
```bash
curl -X POST "http://localhost:8000/transaction-patterns" \
-H "Content-Type: application/json" \
-H "etherscan-key: F71AZ6XW2WN6GK3D63HJ7AVAEDC9M42EZY" \
-d '{
"address": "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045",
"chain": "ethereum"
}'
```
### 响应
```json
{
"address": "0x...",
"chain": "ethereum",
"patterns": {
"hourly_distribution": { "0": 5, "14": 20, ... },
"daily_distribution": { "Monday": 10, "Tuesday": 15, ... },
"value_distribution": {
"micro": 198, // < 0.01 ETH
"small": 1, // 0.01 - 0.1 ETH
"medium": 0, // 0.1 - 1 ETH
"large": 1, // 1 - 10 ETH
"whale": 0 // > 10 ETH
},
"avg_interval_hours": 12.5,
"top_counterparties": [
{ "address": "0x...", "tx_count": 15 }
]
},
"behavior_summary": "活跃高峰时段: 14:00 UTC; 以小额交易为主（可能是频繁交易者或机器人）",
"timestamp": "2026-02-05T17:02:30.123456"
}
```
---
## 2.3 资金流向分析
分析资金来源和去向，识别主要入金/出金地址。
### 请求
```bash
POST /fund-flow
Content-Type: application/json
etherscan-key: YOUR_API_KEY
```
### 参数
| 参数 | 类型 | 必需 | 默认值 | 说明 |
|------|------|------|--------|------|
| address | string | ✅ | - | 钱包地址 |
| chain | string | ❌ | ethereum | 区块链网络 |
| limit | int | ❌ | 100 | 分析交易数量 (10-500) |
### 示例
```bash
curl -X POST "http://localhost:8000/fund-flow" \
-H "Content-Type: application/json" \
-H "etherscan-key: F71AZ6XW2WN6GK3D63HJ7AVAEDC9M42EZY" \
-d '{
"address": "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045",
"chain": "ethereum",
"limit": 100
}'
```
### 响应
```json
{
"address": "0x...",
"chain": "ethereum",
"fund_flow": {
"total_inflow": 5.234,
"total_outflow": 3.156,
"net_flow": 2.078,
"inflow_sources": 42,
"outflow_destinations": 8,
"top_inflow": [
{ "address": "0x...", "amount": 2.5, "symbol": "ETH" }
],
"top_outflow": [
{ "address": "0x...", "amount": 1.0, "symbol": "ETH" }
]
},
"symbol": "ETH",
"timestamp": "2026-02-05T17:03:00.123456"
}
```
---
## 2.4 合约交互分析
分析地址与智能合约的交互情况。
### 请求
```bash
POST /contract-interactions
Content-Type: application/json
etherscan-key: YOUR_API_KEY
```
### 参数
| 参数 | 类型 | 必需 | 默认值 | 说明 |
|------|------|------|--------|------|
| address | string | ✅ | - | 钱包地址 |
| chain | string | ❌ | ethereum | 区块链网络 |
### 示例
```bash
curl -X POST "http://localhost:8000/contract-interactions" \
-H "Content-Type: application/json" \
-H "etherscan-key: F71AZ6XW2WN6GK3D63HJ7AVAEDC9M42EZY" \
-d '{
"address": "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045",
"chain": "ethereum"
}'
```
### 响应
```json
{
"address": "0x...",
"chain": "ethereum",
"contract_interactions": {
"total_contracts": 10,
"top_contracts": [
{
"contract": "0x...",
"interaction_count": 9,
"unique_methods": 1,
"total_value": 0.5,
"symbol": "ETH",
"explorer_url": "https://etherscan.io/address/0x..."
}
]
},
"timestamp": "2026-02-05T17:03:30.123456"
}
```
---
## 2.5 智能分析对话 (Chat)
### 请求
```bash
POST /chat
Content-Type: application/json
etherscan-key: YOUR_ETHERSCAN_KEY
llm-key: YOUR_LLM_API_KEY
```
### 参数
| 参数 | 类型 | 必需 | 默认值 | 说明 |
|------|------|------|--------|------|
| message | string | ✅ | - | 分析请求（包含地址） |
| chain | string | ❌ | ethereum | 默认区块链网络 |
### 示例
```bash
curl -X POST "http://localhost:8000/chat" \
-H "Content-Type: application/json" \
-H "etherscan-key: F71AZ6XW2WN6GK3D63HJ7AVAEDC9M42EZY" \
-H "llm-key: sk-xxx" \
-d '{
"message": "分析 0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045 是不是巨鲸或机器人",
"chain": "ethereum"
}'
```
### 响应
```json
{
"response": "### 地址分析报告\n\n#### 一、基本信息\n- **当前余额**: 32.12 ETH\n...",
"analysis": {
"activity": { ... },
"patterns": { ... },
"fund_flow": { ... },
"contracts": { ... },
"balance": 32.11613029
},
"timestamp": "2026-02-05T17:04:00.123456"
}
```
---
## 统一错误格式
### 成功响应
```json
{
"success": true,
"data": { ... },
"timestamp": "2026-02-05T17:00:00.000000"
}
```
### 错误响应
```json
{
"detail": "错误信息描述"
}
```
### HTTP 状态码
| 状态码 | 说明 |
|--------|------|
| 200 | 成功 |
| 400 | 请求参数错误 |
| 401 | 未提供 API Key |
| 404 | 未找到数据 |
| 500 | 服务器错误 |
---
## 部署信息
| Agent | 镜像地址 | 端口 |
|-------|----------|------|
| Chain Explorer | `agnettaiji.azurecr.io/ai-agents/chain-explorer-agent:latest` | 8000 |
| Chain Analysis | `agnettaiji.azurecr.io/ai-agents/chain-analysis-agent:latest` | 8000 |
### 环境变量
| 变量 | 默认值 | 说明 |
|------|--------|------|
| `SERVICE_HOST` | 0.0.0.0 | 服务绑定地址 |
| `SERVICE_PORT` | 8000 | 服务端口 |
| `LLM_BASE_URL` | https://litellm.xxx | LLM 服务地址 |
| `LLM_MODEL` | taiji/gpt-4o-mini | LLM 模型 |
---
## 测试用地址
| 地址 | 说明 |
|------|------|
| `0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045` | Vitalik Buterin |
| `0xBE0eB53F46cd790Cd13851d5EFf43D12404d33E8` | Binance Cold Wallet |
| `0x28C6c06298d514Db089934071355E5743bf21d60` | Binance Hot Wallet |
---
## 最佳实践
1. **API Key 管理**：Etherscan API 有速率限制，建议申请付费 API Key
2. **缓存策略**：对于不常变化的数据（如历史交易），建议本地缓存
3. **并发控制**：避免短时间内大量请求，建议间隔 200ms
4. **多链支持**：使用统一的 Etherscan V2 API，通过 chainid 区分网络
---
## 版本历史
| 版本 | 日期 | 更新内容 |
|------|------|----------|
| 1.0.0 | 2026-02-05 | 初始版本，支持 Etherscan V2 API |