OpenAI Integration
Enhance OpenAI completions with real-time search and academic context
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 = 20.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="")
Structured Outputs with Search
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 = 10.0):
# Adjust search parameters based on budget
if max_budget < 5.0:
search_params = {
"max_num_results": 3,
"search_type": "web",
"max_price": max_budget
}
elif max_budget < 15.0:
search_params = {
"max_num_results": 5,
"search_type": "all",
"max_price": max_budget
}
else:
search_params = {
"max_num_results": 10,
"search_type": "all",
"max_price": max_budget,
"relevance_threshold": 0.7
}
# 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 querysearch_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
- Overview
- Installation
- Basic Integration
- Function Definition
- Basic Usage
- Advanced Patterns
- Streaming with Function Calls
- Structured Outputs with Search
- Specialized Use Cases
- Financial Analysis Assistant
- Academic Research Assistant
- Best Practices
- 1. Error Handling and Fallbacks
- 2. Cost Management
- 3. Async Support
- API Reference
- Function Parameters
- Response Format
- Additional Resources