​
Overview

Valyu provides seamless integration with the OpenAI API through function calling, enabling your OpenAI models to access proprietary data sources, real-time web search, academic data sources, and financial data. This integration allows your AI applications to provide more informed and up-to-date responses without changing your core OpenAI workflow.

​
Installation

Install the required packages:

pip install openai requests

You’ll also need to set your API keys:

export OPENAI_API_KEY="your-openai-api-key"
export VALYU_API_KEY="your-valyu-api-key"

Free Credits

Get your API key with $10 credit from the Valyu Platform.

​
Basic Integration

​
Function Definition

First, define the Valyu search function for OpenAI to use:

import openai
import requests
import json
from typing import Literal

# Initialize OpenAI client
client = openai.OpenAI()

def valyu_search(
    query: str,
    search_type: Literal["all", "web", "proprietary"] = "all",
    max_num_results: int = 5,
    relevance_threshold: float = 0.5,
    max_price: float = 30.0,
    category: str = None
) -> str:
    """
    Search for information using Valyu's comprehensive knowledge base.

    Args:
        query: Natural language search query
        search_type: Type of search - "all", "web", or "proprietary"
        max_num_results: Number of results to return (1-20)
        relevance_threshold: Minimum relevance score (0.0-1.0)
        max_price: Maximum cost in dollars
        category: Natural language category to guide search

    Returns:
        JSON string with search results
    """
    url = "https://api.valyu.network/v1/deepsearch"

    payload = {
        "query": query,
        "search_type": search_type,
        "max_num_results": max_num_results,
        "relevance_threshold": relevance_threshold,
        "max_price": max_price,
        "is_tool_call": True
    }

    if category:
        payload["category"] = category

    headers = {
        "Authorization": f"Bearer {os.environ['VALYU_API_KEY']}",
        "Content-Type": "application/json"
    }

    try:
        response = requests.post(url, json=payload, headers=headers)
        response.raise_for_status()
        return json.dumps(response.json(), indent=2)
    except Exception as e:
        return f"Search error: {str(e)}"

# Define the function schema for OpenAI
valyu_function = {
    "type": "function",
    "function": {
        "name": "valyu_search",
        "description": "Search for real-time information, academic papers, and comprehensive knowledge using Valyu's database",
        "parameters": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "Natural language search query"
                },
                "search_type": {
                    "type": "string",
                    "enum": ["all", "web", "proprietary"],
                    "description": "Type of search: 'all' for comprehensive, 'web' for current events, 'proprietary' for academic"
                },
                "max_num_results": {
                    "type": "integer",
                    "minimum": 1,
                    "maximum": 20,
                    "description": "Number of results to return"
                },
                "relevance_threshold": {
                    "type": "number",
                    "minimum": 0.0,
                    "maximum": 1.0,
                    "description": "Minimum relevance score for results"
                },
                "max_price": {
                    "type": "number",
                    "description": "Maximum cost in dollars for this search"
                },
                "category": {
                    "type": "string",
                    "description": "Natural language category to guide search context"
                }
            },
            "required": ["query"]
        }
    }
}

​
Basic Usage

Use the function with OpenAI’s function calling:

import os

def chat_with_search(user_message: str):
    messages = [
        {
            "role": "system",
            "content": "You are a helpful assistant with access to real-time search. Use the valyu_search function to find current information when needed."
        },
        {
            "role": "user",
            "content": user_message
        }
    ]

    # Initial completion with function calling
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=[valyu_function],
        tool_choice="auto"
    )

    # Check if the model wants to call a function
    if response.choices[0].message.tool_calls:
        # Add the assistant's response to messages
        messages.append(response.choices[0].message)

        # Process each tool call
        for tool_call in response.choices[0].message.tool_calls:
            if tool_call.function.name == "valyu_search":
                # Parse function arguments
                function_args = json.loads(tool_call.function.arguments)

                # Call the function
                search_results = valyu_search(**function_args)

                # Add function result to messages
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": search_results
                })

        # Get final response with search results
        final_response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages
        )

        return final_response.choices[0].message.content
    else:
        return response.choices[0].message.content

# Example usage
result = chat_with_search("What are the latest developments in quantum computing?")
print(result)

​
Advanced Patterns

​
Streaming with Function Calls

Handle streaming responses with function calling:

def stream_chat_with_search(user_message: str):
    messages = [
        {
            "role": "system",
            "content": "You are a helpful assistant with access to real-time search."
        },
        {
            "role": "user",
            "content": user_message
        }
    ]

    # Stream the initial response
    stream = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=[valyu_function],
        tool_choice="auto",
        stream=True
    )

    tool_calls = []
    current_tool_call = None

    for chunk in stream:
        if chunk.choices[0].delta.tool_calls:
            for tool_call_delta in chunk.choices[0].delta.tool_calls:
                if tool_call_delta.index is not None:
                    # New tool call
                    if len(tool_calls) <= tool_call_delta.index:
                        tool_calls.append({
                            "id": "",
                            "function": {"name": "", "arguments": ""}
                        })
                    current_tool_call = tool_calls[tool_call_delta.index]

                if tool_call_delta.id:
                    current_tool_call["id"] = tool_call_delta.id
                if tool_call_delta.function.name:
                    current_tool_call["function"]["name"] = tool_call_delta.function.name
                if tool_call_delta.function.arguments:
                    current_tool_call["function"]["arguments"] += tool_call_delta.function.arguments

        elif chunk.choices[0].delta.content:
            print(chunk.choices[0].delta.content, end="")

    # Process tool calls if any
    if tool_calls:
        for tool_call in tool_calls:
            if tool_call["function"]["name"] == "valyu_search":
                function_args = json.loads(tool_call["function"]["arguments"])
                search_results = valyu_search(**function_args)

                messages.append({
                    "role": "assistant",
                    "tool_calls": [tool_call]
                })
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call["id"],
                    "content": search_results
                })

        # Stream final response
        final_stream = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            stream=True
        )

        for chunk in final_stream:
            if chunk.choices[0].delta.content:
                print(chunk.choices[0].delta.content, end="")

Combine Valyu search with OpenAI’s structured outputs:

from pydantic import BaseModel
from typing import List

class SearchResult(BaseModel):
    title: str
    summary: str
    source_type: str
    relevance_score: float

class ResearchReport(BaseModel):
    topic: str
    key_findings: List[str]
    sources: List[SearchResult]
    conclusion: str

def generate_research_report(topic: str) -> ResearchReport:
    messages = [
        {
            "role": "system",
            "content": "You are a research assistant. Use valyu_search to gather information, then create a structured research report."
        },
        {
            "role": "user",
            "content": f"Create a comprehensive research report on: {topic}"
        }
    ]

    # First, get search results
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=[valyu_function],
        tool_choice="auto"
    )

    # Process tool calls
    if response.choices[0].message.tool_calls:
        messages.append(response.choices[0].message)

        for tool_call in response.choices[0].message.tool_calls:
            if tool_call.function.name == "valyu_search":
                function_args = json.loads(tool_call.function.arguments)
                search_results = valyu_search(**function_args)

                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": search_results
                })

    # Generate structured report
    messages.append({
        "role": "user",
        "content": "Now create a structured research report based on the search results."
    })

    completion = client.beta.chat.completions.parse(
        model="gpt-4o",
        messages=messages,
        response_format=ResearchReport
    )

    return completion.choices[0].parsed

# Example usage
report = generate_research_report("artificial intelligence in healthcare")
print(f"Topic: {report.topic}")
print(f"Key Findings: {report.key_findings}")

​
Specialized Use Cases

​
Financial Analysis Assistant

def financial_analysis(query: str):
    messages = [
        {
            "role": "system",
            "content": """You are a financial analyst with access to real-time market data.
            Use valyu_search with search_type='web' for current market news and
            search_type='proprietary' for academic financial research."""
        },
        {
            "role": "user",
            "content": query
        }
    ]

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=[valyu_function],
        tool_choice="auto"
    )

    # Process function calls and return analysis
    # (Implementation similar to basic usage)
    return process_response_with_tools(response, messages)

# Example
analysis = financial_analysis("Analyze the recent news and the historical prices of Microsoft stock")

​
Academic Research Assistant

def academic_research(research_question: str):
    # Custom function for academic searches
    academic_function = {
        "type": "function",
        "function": {
            "name": "valyu_search",
            "description": "Search academic databases for research papers and scholarly articles",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string"},
                    "search_type": {"type": "string", "enum": ["proprietary"]},
                    "max_num_results": {"type": "integer", "minimum": 5, "maximum": 15},
                    "relevance_threshold": {"type": "number", "minimum": 0.6},
                    "category": {"type": "string"}
                },
                "required": ["query"]
            }
        }
    }

    messages = [
        {
            "role": "system",
            "content": "You are an academic research assistant. Focus on peer-reviewed sources and provide proper citations."
        },
        {
            "role": "user",
            "content": research_question
        }
    ]

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=[academic_function],
        tool_choice="auto"
    )

    return process_response_with_tools(response, messages)

​
Best Practices

​
1. Error Handling and Fallbacks

def robust_search_chat(user_message: str):
    try:
        return chat_with_search(user_message)
    except requests.RequestException as e:
        print(f"Search API error: {e}")
        # Fallback to standard OpenAI without search
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": "You are a helpful assistant."},
                {"role": "user", "content": user_message}
            ]
        )
        return response.choices[0].message.content
    except Exception as e:
        print(f"Unexpected error: {e}")
        return "I apologize, but I encountered an error processing your request."

​
2. Cost Management

def cost_controlled_search(query: str, max_budget: float = 20.0):
    # Adjust search parameters based on budget
    if max_budget < 20.0:
        search_params = {
            "max_num_results": 3,
            "max_price": max_budget
        }
    elif max_budget < 30.0:
        search_params = {
            "max_num_results": 5,
            "max_price": max_budget
        }
    else:
        search_params = {
            "max_num_results": 10,
            "max_price": max_budget,
            "relevance_threshold": 0.6
        }

    # Use in function call
    return chat_with_search_params(query, search_params)

​
3. Async Support

import asyncio
import aiohttp

async def async_valyu_search(query: str, **kwargs):
    """Async version of Valyu search"""
    url = "https://api.valyu.network/v1/deepsearch"

    payload = {
        "query": query,
        "is_tool_call": True,
        **kwargs
    }

    headers = {
        "Authorization": f"Bearer {os.environ['VALYU_API_KEY']}",
        "Content-Type": "application/json"
    }

    async with aiohttp.ClientSession() as session:
        async with session.post(url, json=payload, headers=headers) as response:
            result = await response.json()
            return json.dumps(result, indent=2)

async def async_chat_with_search(user_message: str):
    """Async chat with search capabilities"""
    # Implementation using async OpenAI client
    # and async_valyu_search function
    pass

​
API Reference

​
Function Parameters

The valyu_search function supports all v2 API parameters:

  • query (required): Natural language search query
  • search_type: "all", "web", or "proprietary" (default: "all")
  • max_num_results: 1-20 results (default: 5)
  • relevance_threshold: 0.0-1.0 relevance filter (default: 0.5)
  • max_price: Maximum cost in dollars (default: 20.0)
  • category: Natural language context guide (optional)
  • included_sources: List of specific datasets/URLs (optional)
  • start_date/end_date: Time filtering (YYYY-MM-DD format, optional)

​
Response Format

Search results are returned as JSON with the following structure:

{
  "results": [
    {
      "title": "Result title",
      "content": "Result content/snippet",
      "url": "Source URL",
      "relevance_score": 0.85,
      "source_type": "web|academic|financial",
      "published_date": "2024-01-15"
    }
  ],
  "total_results": 5,
  "search_metadata": {
    "query": "original query",
    "search_type": "all",
    "cost": 2.5
  }
}

​
Additional Resources

OpenAI Function Calling

Official OpenAI function calling documentation

Valyu API Reference

Complete Valyu v2 API documentation

OpenAI Structured Outputs

Learn about structured outputs with Pydantic

Get API Key

Sign up for free $10 credit