如何将多平台实时数据添加到任何 AI 代理

import os, requests

SCAVIO_KEY = os.environ['SCAVIO_API_KEY']
SEARCH_URL = 'https://api.scavio.dev/api/v1/search'
TT_URL = 'https://api.scavio.dev/api/v1/tiktok'
H = {'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json'}
TT_H = {'Authorization': f'Bearer {SCAVIO_KEY}', 'Content-Type': 'application/json'}

def search_platform(query: str, platform: str = 'google', num: int = 5) -> list:
    """Search any platform. Returns normalized results."""
    if platform == 'tiktok':
        resp = requests.post(f'{TT_URL}/search/videos', headers=TT_H,
            json={'keyword': query, 'count': num, 'cursor': 0})
        videos = resp.json().get('data', {}).get('videos', [])
        return [{'title': v.get('desc', '')[:100], 'url': f'tiktok.com/@{v.get("author",{}).get("uniqueId","")}',
                 'snippet': f'Plays: {v.get("stats",{}).get("playCount",0):,}', 'platform': 'tiktok'} for v in videos]
    # All other platforms use the search endpoint with site: prefix
    site_map = {'amazon': 'amazon.com', 'youtube': 'youtube.com',
                'walmart': 'walmart.com', 'reddit': 'reddit.com'}
    q = f'site:{site_map[platform]} {query}' if platform in site_map else query
    resp = requests.post(SEARCH_URL, headers=H,
        json={'query': q, 'country_code': 'us', 'num_results': num})
    return [{'title': r['title'], 'url': r['link'], 'snippet': r.get('snippet', ''),
             'platform': platform} for r in resp.json().get('organic_results', [])]

for p in ['google', 'amazon', 'reddit']:
    results = search_platform('wireless earbuds', platform=p, num=2)
    print(f'{p}: {len(results)} results')
    for r in results:
        print(f'  {r["title"][:60]}')

TOOL_DEFINITION = {
    'type': 'function',
    'function': {
        'name': 'search',
        'description': ('Search across Google, Amazon, YouTube, Walmart, Reddit, or TikTok. '
                       'Use platform parameter to target a specific source.'),
        'parameters': {
            'type': 'object',
            'properties': {
                'query': {'type': 'string', 'description': 'The search query'},
                'platform': {
                    'type': 'string',
                    'enum': ['google', 'amazon', 'youtube', 'walmart', 'reddit', 'tiktok'],
                    'description': 'Platform to search. Default: google'
                },
                'num_results': {'type': 'integer', 'description': 'Number of results (1-10)', 'default': 5}
            },
            'required': ['query']
        }
    }
}

def handle_tool_call(name: str, args: dict) -> str:
    """Execute agent tool calls."""
    if name == 'search':
        results = search_platform(
            query=args['query'],
            platform=args.get('platform', 'google'),
            num=args.get('num_results', 5)
        )
        return '\n'.join(f'[{i+1}] {r["title"]} ({r["platform"]})\n    {r["snippet"]}'
                         for i, r in enumerate(results))
    return 'Unknown tool'

print(handle_tool_call('search', {'query': 'best earbuds', 'platform': 'reddit'}))

import json

LLM_URL = os.environ.get('LLM_URL', 'http://localhost:11434/v1/chat/completions')
LLM_KEY = os.environ.get('LLM_API_KEY', 'ollama')

def agent_loop(user_query: str, max_turns: int = 5) -> str:
    messages = [
        {'role': 'system', 'content': (
            'You are a research assistant with access to 6 search platforms: '
            'Google, Amazon, YouTube, Walmart, Reddit, and TikTok. '
            'Use the search tool to find real-time data. Search multiple platforms '
            'when the user needs a comprehensive answer.'
        )},
        {'role': 'user', 'content': user_query}
    ]
    for turn in range(max_turns):
        resp = requests.post(LLM_URL, headers={
            'Authorization': f'Bearer {LLM_KEY}', 'Content-Type': 'application/json'
        }, json={'model': 'llama3', 'messages': messages,
                'tools': [TOOL_DEFINITION], 'max_tokens': 1024})
        choice = resp.json()['choices'][0]
        msg = choice['message']
        messages.append(msg)
        if not msg.get('tool_calls'):
            return msg['content']
        for tc in msg['tool_calls']:
            args = json.loads(tc['function']['arguments']) if isinstance(tc['function']['arguments'], str) else tc['function']['arguments']
            result = handle_tool_call(tc['function']['name'], args)
            messages.append({'role': 'tool', 'tool_call_id': tc['id'], 'content': result})
    return messages[-1].get('content', 'Max turns reached')

answer = agent_loop('Compare wireless earbuds prices on Amazon vs Walmart')
print(answer)

from concurrent.futures import ThreadPoolExecutor, as_completed

def multi_platform_search(query: str, platforms: list[str] = None) -> dict:
    """Search multiple platforms in parallel."""
    if platforms is None:
        platforms = ['google', 'amazon', 'reddit']
    results = {}
    with ThreadPoolExecutor(max_workers=len(platforms)) as executor:
        futures = {executor.submit(search_platform, query, p, 3): p
                   for p in platforms}
        for future in as_completed(futures):
            platform = futures[future]
            try:
                results[platform] = future.result()
            except Exception as e:
                results[platform] = [{'title': f'Error: {e}', 'url': '', 'snippet': '', 'platform': platform}]
    return results

def format_multi_results(results: dict) -> str:
    lines = []
    idx = 1
    for platform, items in results.items():
        lines.append(f'\n--- {platform.upper()} ---')
        for r in items:
            lines.append(f'[{idx}] {r["title"]}')
            if r['snippet']:
                lines.append(f'    {r["snippet"][:150]}')
            idx += 1
    return '\n'.join(lines)

results = multi_platform_search('noise cancelling headphones')
print(format_multi_results(results))
print(f'\nTotal cost: ${sum(len(v) for v in results.values()) * 0.005:.3f}')