Comment construire un pipeline de recherche Hermes en utilisant l'API de recherche

import os, requests, json
from datetime import datetime
from collections import defaultdict

API_KEY = os.environ['SCAVIO_API_KEY']
SH = {'x-api-key': API_KEY, 'Content-Type': 'application/json'}

def research_search(query, depth='standard'):
    """Search optimized for research. Returns structured sources."""
    data = requests.post('https://api.scavio.dev/api/v1/search',
        headers=SH, json={'query': query, 'country_code': 'us'}, timeout=10).json()
    sources = []
    for r in data.get('organic_results', []):
        source = {
            'title': r.get('title', ''),
            'url': r.get('link', ''),
            'snippet': r.get('snippet', ''),
            'domain': r.get('displayed_link', '').split('/')[0],
            'position': r.get('position', 0),
        }
        sources.append(source)
    related = [q.get('question', '') for q in data.get('people_also_ask', [])]
    featured = data.get('featured_snippet', {})
    return {
        'query': query,
        'sources': sources,
        'related_questions': related,
        'featured_answer': featured.get('snippet', ''),
        'source_count': len(sources),
    }

# Test
result = research_search('what is model context protocol mcp 2026')
print(f'Query: {result["query"]}')
print(f'Sources: {result["source_count"]}')
print(f'Featured answer: {result["featured_answer"][:80]}...' if result['featured_answer'] else 'No featured answer')
for s in result['sources'][:3]:
    print(f'  [{s["domain"]:20}] {s["title"][:50]}')

def research_pipeline(topic, angles=None):
    """Multi-step research pipeline for Hermes agent."""
    if not angles:
        angles = [
            f'{topic}',
            f'{topic} comparison alternatives',
            f'{topic} tutorial getting started',
        ]
    all_sources = []
    all_facts = []
    print(f'\n=== Researching: {topic} ===')
    for i, angle in enumerate(angles, 1):
        print(f'\n  Step {i}: "{angle[:50]}"')
        result = research_search(angle)
        all_sources.extend(result['sources'])
        if result['featured_answer']:
            all_facts.append({'fact': result['featured_answer'][:150], 'query': angle})
        print(f'    Sources: {result["source_count"]} | Featured: {"yes" if result["featured_answer"] else "no"}')
        # Use related questions for deeper research
        if result['related_questions'] and i == 1:
            for rq in result['related_questions'][:2]:
                angles.append(rq)
                print(f'    Added follow-up: {rq[:50]}')
    # Deduplicate sources by domain
    seen = set()
    unique_sources = []
    for s in all_sources:
        if s['domain'] not in seen:
            seen.add(s['domain'])
            unique_sources.append(s)
    return {
        'topic': topic,
        'queries': len(angles),
        'unique_sources': unique_sources,
        'facts': all_facts,
        'cost': len(angles) * 0.005,
    }

research = research_pipeline('model context protocol MCP')
print(f'\n  Queries: {research["queries"]} | Unique sources: {len(research["unique_sources"])}')
print(f'  Facts found: {len(research["facts"])}')
print(f'  Cost: ${research["cost"]:.3f}')

def generate_brief(research):
    print(f'\n{"=" * 60}')
    print(f'  RESEARCH BRIEF: {research["topic"]}')
    print(f'  Date: {datetime.now().strftime("%Y-%m-%d")}')
    print(f'  Queries: {research["queries"]} | Sources: {len(research["unique_sources"])}')
    print(f'{"=" * 60}')
    # Key facts
    if research['facts']:
        print(f'\n  Key Findings:')
        for i, f in enumerate(research['facts'], 1):
            print(f'    {i}. {f["fact"][:80]}')
            print(f'       Source query: "{f["query"][:40]}"')
    # Source overview
    print(f'\n  Sources ({len(research["unique_sources"])} unique):')
    for s in research['unique_sources'][:10]:
        print(f'    [{s["domain"]:25}] {s["title"][:45]}')
    # Authority breakdown
    domains = [s['domain'] for s in research['unique_sources']]
    authority = {
        'official': sum(1 for d in domains if any(w in d for w in ['github.com', '.dev', '.io'])),
        'news': sum(1 for d in domains if any(w in d for w in ['techcrunch', 'verge', 'arstechnica'])),
        'community': sum(1 for d in domains if any(w in d for w in ['reddit', 'stackoverflow', 'hackernews'])),
    }
    print(f'\n  Source Authority:')
    for cat, count in authority.items():
        print(f'    {cat:15} {count} sources')
    print(f'\n  Research cost: ${research["cost"]:.3f}')
    print(f'  Time: ~{research["queries"]}s (vs 30-60s with browser search)')

generate_brief(research)