feat(responses): add file_search tool interception for non-native providers

- Add FileSearchResponsesAPIUtils for intercepting file_search tools
- Search vector stores and inject context for Claude, Gemini, etc.
- Add supports_native_file_search to provider configs
- Add blog post and docs for Responses API file_search
- Add create_vector_store script and config
- Add unit tests for file_search utils
- Restructure proxy UI out from .html to /index.html

Made-with: Cursor
This commit is contained in:
Sameer Kankute 2026-03-16 18:54:41 +05:30
parent 58e74a631c
commit e4fd0383f3
44 changed files with 974 additions and 228 deletions

17
config_vector_store.yaml Normal file
View file

@ -0,0 +1,17 @@
model_list:
- model_name: gpt-4o
litellm_params:
model: openai/gpt-4o
api_key: os.environ/OPENAI_API_KEY
general_settings:
set_verbose: true
master_key: sk-vector-store-demo
# Add your vector store ID after running: python scripts/create_vector_store.py
vector_store_registry:
- vector_store_name: "litellm-docs-openai"
litellm_params:
vector_store_id: "REPLACE_WITH_VS_ID" # e.g. vs_687ae3b2439881918b433cb99d10662e
custom_llm_provider: "openai"
vector_store_description: "LiteLLM docs (CLAUDE.md) for RAG"

View file

@ -0,0 +1,99 @@
---
slug: responses_api_file_search
title: "Generic file_search in the Responses API: RAG for Any Model"
date: 2026-03-16T10:00:00
authors:
- name: Sameer Kankute
title: SWE @ LiteLLM (LLM Translation)
url: https://www.linkedin.com/in/sameer-kankute/
image_url: https://pbs.twimg.com/profile_images/2001352686994907136/ONgNuSk5_400x400.jpg
description: "Use the standard OpenAI Responses API file_search tool with Claude, Gemini, GPT-4, and any LiteLLM-supported model. LiteLLM intercepts for non-native providers, searches your vector stores, and injects context automatically."
tags: [responses api, file_search, vector stores, rag, openai]
hide_table_of_contents: false
---
# Generic file_search in the Responses API
The **`file_search` tool** in the OpenAI Responses API lets models query vector stores for retrieval-augmented generation (RAG). Until now, only OpenAI and Azure natively supported it. LiteLLM now extends this to **all providers**—Claude, Gemini, Vertex AI, Bedrock, and more—via a single, unified contract.
Pass `tools=[{"type": "file_search", "vector_store_ids": ["vs_abc123"]}]` and LiteLLM handles the rest.
## Architecture
```mermaid
flowchart TD
Request["tools=[{type: file_search, vector_store_ids: [...]}]"]
Request --> CheckProvider["check supports_native_file_search(provider)"]
CheckProvider --> HasFileSearch{"has file_search tools?"}
HasFileSearch -->|no| Handler["existing handler"]
HasFileSearch -->|yes| Native{"native provider?"}
Native -->|"YES: OpenAI / Azure"| PassThrough["pass file_search through unchanged"]
PassThrough --> Handler
Native -->|"NO: Claude / Gemini / etc."| Intercept["FileSearchResponsesAPIUtils"]
Intercept --> ExtractQuery["extract query from input"]
ExtractQuery --> Asearch["litellm.vector_stores.asearch per vector_store_id"]
Asearch --> InjectContext["inject context into input"]
InjectContext --> StripTool["strip file_search from tools"]
StripTool --> StoreResults["store results for post-call hook"]
StoreResults --> Handler
Handler --> Route["base_llm_http_handler OR litellm_completion_transformation_handler"]
```
**Native path (OpenAI, Azure):** The `file_search` tool is forwarded to the provider as-is. The provider performs retrieval and returns citations in its response.
**Non-native path (Claude, Gemini, etc.):** LiteLLM intercepts the request before routing:
1. Extracts the query from the last user message in `input`
2. Calls `litellm.vector_stores.asearch()` for each `vector_store_id`
3. Builds a context string from the retrieved chunks
4. Injects the context as a user message before the original input
5. Strips the `file_search` tool from `tools` (other tools are preserved)
6. Forwards the enriched request to the model
7. Stores search results in `model_call_details["search_results"]` for logging and citations
## Usage
```python
import litellm
# Works with any provider—Claude, Gemini, GPT-4, etc.
response = litellm.responses(
model="anthropic/claude-opus-4-5",
input="What does our docs say about testing?",
tools=[
{
"type": "file_search",
"vector_store_ids": ["vs_abc123"],
}
],
)
print(response)
```
Via the LiteLLM Proxy with the OpenAI SDK:
```python
from openai import OpenAI
client = OpenAI(
base_url="http://localhost:4000",
api_key="your-proxy-api-key",
)
response = client.responses.create(
model="anthropic/claude-opus-4-5",
input="Summarise the company handbook.",
tools=[{"type": "file_search", "vector_store_ids": ["vs_handbook_abc"]}],
)
```
## Prerequisites
- **Vector stores** must be created and populated (e.g. via [Create a Vector Store](/docs/vector_stores/create)).
- **`vector_store_registry`** must be configured in the proxy `config.yaml` or via the Python SDK so LiteLLM can resolve each `vector_store_id` to the correct provider and credentials.
## Learn more
- [Responses API file_search docs](/docs/response_api#file_search-vector-store-rag)
- [Vector Store Create](/docs/vector_stores/create)
- [Vector Store Search](/docs/vector_stores/search)

View file

@ -1556,6 +1556,129 @@ curl -X POST "http://localhost:4000/v1/responses" \
}'
```
## file_search (Vector Store RAG)
The **`file_search` tool** lets any model use LiteLLM-managed vector stores for retrieval-augmented generation (RAG) via the standard OpenAI Responses API contract.
| Provider | Behaviour |
|----------|-----------|
| `openai`, `azure` | `file_search` is forwarded to the provider **unchanged** — the provider handles retrieval natively. |
| All other providers (`anthropic`, `gemini`, `vertex_ai`, `bedrock`, etc.) | LiteLLM **intercepts** the request: searches each `vector_store_id` using `litellm.vector_stores`, injects the retrieved context into `input`, strips the `file_search` tool, then forwards the enriched request to the model. |
### How to use
Pass `tools=[{"type": "file_search", "vector_store_ids": [...]}]` in your request. The vector stores must already exist and be accessible via [LiteLLM vector stores](./vector_stores/create.md).
#### LiteLLM Python SDK
```python showLineNumbers title="file_search with LiteLLM Python SDK (any provider)"
import litellm
# Works with OpenAI, Claude, Gemini, and any other LiteLLM provider
response = litellm.responses(
model="anthropic/claude-opus-4-5", # non-native provider example
input="What are the key points about RAG systems?",
tools=[
{
"type": "file_search",
"vector_store_ids": ["vs_abc123", "vs_def456"],
}
],
)
print(response)
```
For native providers (OpenAI/Azure) the tool is passed through as-is:
```python showLineNumbers title="file_search with OpenAI (native passthrough)"
import litellm
response = litellm.responses(
model="openai/gpt-4.1",
input="Summarise the company handbook.",
tools=[
{
"type": "file_search",
"vector_store_ids": ["vs_handbook_abc"],
}
],
)
```
#### LiteLLM Proxy (AI Gateway)
**OpenAI Python SDK (proxy as `base_url`):**
```python showLineNumbers title="file_search via LiteLLM Proxy"
from openai import OpenAI
client = OpenAI(
base_url="http://localhost:4000",
api_key="your-proxy-api-key",
)
response = client.responses.create(
model="anthropic/claude-opus-4-5",
input="What does our refund policy say?",
tools=[
{
"type": "file_search",
"vector_store_ids": ["vs_policies_xyz"],
}
],
)
```
**curl:**
```bash title="file_search via curl to LiteLLM Proxy"
curl -X POST "http://localhost:4000/v1/responses" \
-H "Content-Type: application/json" \
-H "Authorization: Bearer your-proxy-api-key" \
-d '{
"model": "anthropic/claude-opus-4-5",
"input": "What does our refund policy say?",
"tools": [
{
"type": "file_search",
"vector_store_ids": ["vs_policies_xyz"]
}
]
}'
```
### How the non-native path works
```
tools=[{type: file_search, vector_store_ids: [...]}]
check supports_native_file_search(provider)
┌────┴──────────────────┐
│ native? (OpenAI/Azure)│──► pass file_search tool through unchanged
└───────────────────────┘
│ non-native? │──► 1. extract query from input
│ (Claude, Gemini, …) │ 2. litellm.vector_stores.asearch() per vector_store_id
│ │ 3. inject retrieved context into input
│ │ 4. strip file_search from tools
└───────────────────────┘
call model with enriched input (no file_search tool)
```
1. LiteLLM extracts the query text from the last user message in `input`.
2. It calls [`litellm.vector_stores.asearch`](./vector_stores/search.md) for every `vector_store_id` listed in the tool.
3. Retrieved text chunks are prepended to `input` as a context user-message (`"Context:\n\n..."`).
4. The `file_search` entry is removed from `tools` before the request is sent to the model (other tools are preserved).
5. Search results are stored in `litellm_logging_obj.model_call_details["search_results"]` for downstream logging and observability hooks.
### Prerequisites
- Vector stores must be created and populated beforehand. See [Create a Vector Store](./vector_stores/create.md).
- The `litellm.vector_store_registry` (or the proxy's `vector_stores` config block) must be configured so LiteLLM knows where to route each `vector_store_id`.
## Session Management
LiteLLM Proxy supports session management for all supported models. This allows you to store and fetch conversation history (state) in LiteLLM Proxy.

View file

@ -230,6 +230,19 @@ class BaseResponsesAPIConfig(ABC):
"""
return False
def supports_native_file_search(self) -> bool:
"""
Returns True if the provider natively handles file_search tools in the Responses API.
When True (OpenAI, Azure), file_search tools with vector_store_ids are forwarded to
the provider unchanged. When False (Claude, Gemini, etc.), LiteLLM intercepts the
request, searches the specified vector stores, injects the retrieved context into
the input, and strips the file_search tool before forwarding to the provider.
Default: False
"""
return False
#########################################################
########## CANCEL RESPONSE API TRANSFORMATION ##########
#########################################################

View file

@ -60,6 +60,9 @@ class OpenAIResponsesAPIConfig(BaseResponsesAPIConfig):
"""No mapping applied since inputs are in OpenAI spec already"""
return dict(response_api_optional_params)
def supports_native_file_search(self) -> bool:
return True
def transform_responses_api_request(
self,
model: str,

View file

@ -0,0 +1,227 @@
"""
Utilities for handling file_search tools in the Responses API for non-native providers.
For providers that do not natively support file_search (e.g. Claude, Gemini), LiteLLM
intercepts the request, searches the specified vector stores via litellm.vector_stores,
injects the retrieved context into the input, and strips the file_search tool before
forwarding to the provider.
"""
from typing import Any, Dict, List, Optional, Tuple, Union
import litellm
import litellm.vector_stores
from litellm._logging import verbose_logger
from litellm.types.llms.openai import ResponseInputParam
from litellm.types.vector_stores import (
VectorStoreResultContent,
VectorStoreSearchResponse,
VectorStoreSearchResult,
)
CONTEXT_PREFIX = "Context:\n\n"
class FileSearchResponsesAPIUtils:
"""Utilities for file_search tool interception in the Responses API."""
@staticmethod
def has_file_search_tools(tools: List[Dict[str, Any]]) -> bool:
"""Return True if any tool has type 'file_search'."""
return any(
isinstance(t, dict) and t.get("type") == "file_search" for t in tools
)
@staticmethod
def split_file_search_tools(
tools: List[Dict[str, Any]],
) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
"""
Split tools into file_search tools and all other tools.
Returns:
(file_search_tools, other_tools)
"""
file_search_tools: List[Dict[str, Any]] = []
other_tools: List[Dict[str, Any]] = []
for tool in tools:
if isinstance(tool, dict) and tool.get("type") == "file_search":
file_search_tools.append(tool)
else:
other_tools.append(tool)
return file_search_tools, other_tools
@staticmethod
def extract_query_from_responses_input(
input: Union[str, ResponseInputParam],
) -> Optional[str]:
"""
Extract the query text from a Responses API input value.
Handles both plain string inputs and list-of-input-item inputs.
For list inputs the text of the last user message is returned.
"""
if isinstance(input, str):
return input
if not isinstance(input, list) or len(input) == 0:
return None
# Walk backwards to find the last user message with text content
for item in reversed(input):
if not isinstance(item, dict):
continue
role = item.get("role")
if role != "user":
continue
content = item.get("content")
if isinstance(content, str):
return content
if isinstance(content, list):
for part in content:
if (
isinstance(part, dict)
and part.get("type") == "text"
and part.get("text")
):
return str(part["text"])
return None
@staticmethod
def inject_context_into_responses_input(
input: Union[str, ResponseInputParam],
context: str,
) -> Union[str, ResponseInputParam]:
"""
Inject a context message into a Responses API input.
The context is inserted as a user message immediately before the last item
in the input list. If input is a plain string it is first wrapped in a
single-element list so that injection is consistent.
"""
context_item: Dict[str, Any] = {
"role": "user",
"content": context,
}
if isinstance(input, str):
return [context_item, {"role": "user", "content": input}]
if not isinstance(input, list):
return input
modified = list(input)
if len(modified) == 0:
modified.append(context_item)
else:
modified.insert(len(modified) - 1, context_item)
return modified
@staticmethod
def _build_context_from_search_results(
results: List[VectorStoreSearchResponse],
) -> str:
"""Build a plain-text context string from a list of vector store search responses."""
context = CONTEXT_PREFIX
for search_response in results:
data: Optional[List[VectorStoreSearchResult]] = search_response.get("data")
if not data:
continue
for result in data:
content_items: Optional[List[VectorStoreResultContent]] = result.get("content")
if not content_items:
continue
for content_item in content_items:
text: Optional[str] = content_item.get("text")
if text:
context += text + "\n\n"
return context
@staticmethod
async def asearch_and_inject_context(
input: Union[str, ResponseInputParam],
tools: List[Dict[str, Any]],
litellm_logging_obj: Optional[Any] = None,
) -> Tuple[Union[str, ResponseInputParam], List[Dict[str, Any]]]:
"""
Intercept file_search tools for a non-native provider.
Steps:
1. Split file_search tools from remaining tools.
2. Extract the query from the input.
3. For every vector_store_id found across all file_search tools, run
litellm.vector_stores.asearch().
4. Build a context string from all results.
5. Inject the context into the input.
6. Return (modified_input, other_tools) file_search tools are dropped.
If no query can be extracted, or no vector_store_ids are found, the
original (input, tools) is returned unchanged.
"""
file_search_tools, other_tools = FileSearchResponsesAPIUtils.split_file_search_tools(tools)
if not file_search_tools:
return input, tools
query = FileSearchResponsesAPIUtils.extract_query_from_responses_input(input)
if not query:
verbose_logger.debug(
"FileSearchResponsesAPIUtils: no query found in input; skipping vector store search"
)
return input, other_tools
# Collect all vector_store_ids from all file_search tools
vector_store_ids: List[str] = []
for fs_tool in file_search_tools:
ids = fs_tool.get("vector_store_ids") or []
vector_store_ids.extend(ids)
if not vector_store_ids:
verbose_logger.debug(
"FileSearchResponsesAPIUtils: file_search tool has no vector_store_ids; stripping tool only"
)
return input, other_tools
all_search_results: List[VectorStoreSearchResponse] = []
for vector_store_id in vector_store_ids:
try:
search_response: VectorStoreSearchResponse = (
await litellm.vector_stores.asearch(
vector_store_id=vector_store_id,
query=query,
)
)
verbose_logger.debug(
"FileSearchResponsesAPIUtils: searched vector_store_id=%s, got %d results",
vector_store_id,
len(search_response.get("data") or []),
)
all_search_results.append(search_response)
except Exception as e:
verbose_logger.exception(
"FileSearchResponsesAPIUtils: error searching vector_store_id=%s: %s",
vector_store_id,
str(e),
)
if not all_search_results:
return input, other_tools
context = FileSearchResponsesAPIUtils._build_context_from_search_results(all_search_results)
# Only inject if we found actual content beyond the prefix
if context == CONTEXT_PREFIX:
return input, other_tools
modified_input = FileSearchResponsesAPIUtils.inject_context_into_responses_input(
input=input, context=context
)
# Store results in logging object for downstream hooks / observability
if litellm_logging_obj is not None:
try:
litellm_logging_obj.model_call_details["search_results"] = all_search_results
except Exception:
pass
return modified_input, other_tools

View file

@ -32,6 +32,7 @@ from litellm.litellm_core_utils.prompt_templates.common_utils import (
)
from litellm.llms.base_llm.responses.transformation import BaseResponsesAPIConfig
from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler
from litellm.responses.file_search_utils import FileSearchResponsesAPIUtils
from litellm.responses.litellm_completion_transformation.handler import (
LiteLLMCompletionTransformationHandler,
)
@ -699,6 +700,26 @@ def responses(
)
)
#########################################################
# file_search tool interception for non-native providers
#########################################################
tools_list = list(tools) if tools else []
if tools_list and FileSearchResponsesAPIUtils.has_file_search_tools(tools_list):
native = (
responses_api_provider_config is not None
and responses_api_provider_config.supports_native_file_search()
)
if not native:
input, tools_list = run_async_function(
FileSearchResponsesAPIUtils.asearch_and_inject_context,
input=input,
tools=tools_list,
litellm_logging_obj=litellm_logging_obj,
)
tools = cast(Optional[Iterable[ToolParam]], tools_list)
local_vars["input"] = input
local_vars["tools"] = tools
local_vars.update(kwargs)
# Map reasoning_effort (from litellm_params/proxy config) to reasoning when not set
if reasoning is None and "reasoning_effort" in local_vars:

View file

@ -1,231 +1,12 @@
model_list:
- model_name: gpt-3.5-turbo-end-user-test
litellm_params:
model: gpt-3.5-turbo
region_name: "eu"
model_info:
id: "1"
- model_name: gpt-3.5-turbo-end-user-test
litellm_params:
model: openai/gpt-4.1-mini
api_key: os.environ/OPENAI_API_KEY # The `os.environ/` prefix tells litellm to read this from the env. See https://docs.litellm.ai/docs/simple_proxy#load-api-keys-from-vault
- model_name: gpt-3.5-turbo
litellm_params:
model: openai/gpt-4.1-mini
api_key: os.environ/OPENAI_API_KEY # The `os.environ/` prefix tells litellm to read this from the env. See https://docs.litellm.ai/docs/simple_proxy#load-api-keys-from-vault
- model_name: gpt-3.5-turbo-large
litellm_params:
model: "gpt-3.5-turbo-1106"
api_key: os.environ/OPENAI_API_KEY
rpm: 480
timeout: 300
stream_timeout: 60
- model_name: gpt-4
litellm_params:
model: openai/gpt-4.1-mini
api_key: os.environ/OPENAI_API_KEY # The `os.environ/` prefix tells litellm to read this from the env. See https://docs.litellm.ai/docs/simple_proxy#load-api-keys-from-vault
rpm: 480
timeout: 300
stream_timeout: 60
- model_name: sagemaker-completion-model
litellm_params:
model: sagemaker/berri-benchmarking-Llama-2-70b-chat-hf-4
input_cost_per_second: 0.000420
- model_name: text-embedding-ada-002
litellm_params:
model: openai/text-embedding-ada-002
api_key: os.environ/OPENAI_API_KEY
model_info:
mode: embedding
base_model: text-embedding-ada-002
- model_name: dall-e-2 # some tests use dall-e-2 which is now deprecated, alias to dall-e-3
litellm_params:
model: openai/dall-e-3
- model_name: openai-dall-e-3
litellm_params:
model: dall-e-3
- model_name: fake-openai-endpoint
litellm_params:
model: openai/gpt-3.5-turbo
api_key: fake-key
api_base: https://exampleopenaiendpoint-production.up.railway.app/
- model_name: fake-openai-endpoint-2
litellm_params:
model: openai/my-fake-model
api_key: my-fake-key
api_base: https://exampleopenaiendpoint-production.up.railway.app/
stream_timeout: 0.001
rpm: 1
- model_name: fake-openai-endpoint-3
litellm_params:
model: openai/my-fake-model
api_key: my-fake-key
api_base: https://exampleopenaiendpoint-production.up.railway.app/
stream_timeout: 0.001
rpm: 1000
- model_name: fake-openai-endpoint-4
litellm_params:
model: openai/my-fake-model
api_key: my-fake-key
api_base: https://exampleopenaiendpoint-production.up.railway.app/
num_retries: 50
- model_name: fake-openai-endpoint-3
litellm_params:
model: openai/my-fake-model-2
api_key: my-fake-key
api_base: https://exampleopenaiendpoint-production.up.railway.app/
stream_timeout: 0.001
rpm: 1000
- model_name: bad-model
litellm_params:
model: openai/bad-model
api_key: os.environ/OPENAI_API_KEY
api_base: https://exampleopenaiendpoint-production.up.railway.app/
mock_timeout: True
timeout: 60
rpm: 1000
model_info:
health_check_timeout: 1
- model_name: good-model
litellm_params:
model: openai/bad-model
api_key: os.environ/OPENAI_API_KEY
api_base: https://exampleopenaiendpoint-production.up.railway.app/
rpm: 1000
model_info:
health_check_timeout: 1
- model_name: "*"
litellm_params:
model: openai/*
api_key: os.environ/OPENAI_API_KEY
- model_name: realtime-v1
litellm_params:
model: azure/gpt-realtime-20250828-standard
api_version: "2025-08-28"
realtime_protocol: GA # Possible values: "GA"/ "v1", "beta"
- model_name: realtime-beta
litellm_params:
model: azure/gpt-realtime-20250828-standard
api_version: 2025-04-01-preview
- model_name: gpt-4o
litellm_params:
model: openai/gpt-4o
- model_name: claude-sonnet-4-5
litellm_params:
model: anthropic/claude-sonnet-4-5
api_key: os.environ/ANTHROPIC_API_KEY
# provider specific wildcard routing
- model_name: "anthropic/*"
litellm_params:
model: "anthropic/*"
api_key: os.environ/ANTHROPIC_API_KEY
- model_name: "bedrock/*"
litellm_params:
model: "bedrock/*"
- model_name: "groq/*"
litellm_params:
model: "groq/*"
api_key: os.environ/GROQ_API_KEY
- model_name: mistral-embed
litellm_params:
model: mistral/mistral-embed
- model_name: gpt-instruct # [PROD TEST] - tests if `/health` automatically infers this to be a text completion model
litellm_params:
model: text-completion-openai/gpt-3.5-turbo-instruct
- model_name: fake-openai-endpoint-5
litellm_params:
model: openai/my-fake-model
api_key: my-fake-key
api_base: https://exampleopenaiendpoint-production.up.railway.app/
timeout: 1
- model_name: badly-configured-openai-endpoint
litellm_params:
model: openai/my-fake-model
api_key: my-fake-key
api_base: https://exampleopenaiendpoint-production.up.railway.appxxxx/
- model_name: gemini-1.5-flash
litellm_params:
model: gemini/gemini-1.5-flash
api_key: os.environ/GOOGLE_API_KEY
- model_name: gpt-4o
litellm_params:
model: gpt-4o
api_key: os.environ/OPENAI_API_KEY
litellm_settings:
# set_verbose: True # Uncomment this if you want to see verbose logs; not recommended in production
drop_params: True
success_callback: ["prometheus"]
# max_budget: 100
# budget_duration: 30d
num_retries: 5
request_timeout: 600
telemetry: False
context_window_fallbacks: [{"gpt-3.5-turbo": ["gpt-3.5-turbo-large"]}]
default_team_settings:
- team_id: team-1
success_callback: ["langfuse"]
failure_callback: ["langfuse"]
langfuse_public_key: os.environ/LANGFUSE_PROJECT1_PUBLIC # Project 1
langfuse_secret: os.environ/LANGFUSE_PROJECT1_SECRET # Project 1
- team_id: team-2
success_callback: ["langfuse"]
failure_callback: ["langfuse"]
langfuse_public_key: os.environ/LANGFUSE_PROJECT2_PUBLIC # Project 2
langfuse_secret: os.environ/LANGFUSE_PROJECT2_SECRET # Project 2
langfuse_host: https://us.cloud.langfuse.com
# cache: true # [OPTIONAL] use for caching responses
# enable_caching_on_provider_specific_optional_params: True # Include provider-specific params in cache keys
# cache_params: # And for shared health check
# type: redis
# host: localhost
# port: 6379
# For /fine_tuning/jobs endpoints
finetune_settings:
- custom_llm_provider: azure
api_base: os.environ/AZURE_API_BASE
api_key: os.environ/AZURE_API_KEY
api_version: "2023-03-15-preview"
- custom_llm_provider: openai
api_key: os.environ/OPENAI_API_KEY
# for /files endpoints
files_settings:
- custom_llm_provider: azure
api_base: os.environ/AZURE_API_BASE
api_key: os.environ/AZURE_API_KEY
api_version: "2023-03-15-preview"
- custom_llm_provider: openai
api_key: os.environ/OPENAI_API_KEY
router_settings:
routing_strategy: usage-based-routing-v2
redis_host: os.environ/REDIS_HOST
redis_password: os.environ/REDIS_PASSWORD
redis_port: os.environ/REDIS_PORT
enable_pre_call_checks: true
model_group_alias: {"my-special-fake-model-alias-name": "fake-openai-endpoint-3"}
general_settings:
master_key: sk-1234 # [OPTIONAL] Use to enforce auth on proxy. See - https://docs.litellm.ai/docs/proxy/virtual_keys
store_model_in_db: True
proxy_budget_rescheduler_min_time: 60
proxy_budget_rescheduler_max_time: 64
proxy_batch_write_at: 1
database_connection_pool_limit: 10
# background_health_checks: true
# use_shared_health_check: true
# health_check_interval: 30
# database_url: "postgresql://<user>:<password>@<host>:<port>/<dbname>" # [OPTIONAL] use for token-based auth to proxy
pass_through_endpoints:
- path: "/v1/rerank" # route you want to add to LiteLLM Proxy Server
target: "https://api.cohere.com/v1/rerank" # URL this route should forward requests to
headers: # headers to forward to this URL
content-type: application/json # (Optional) Extra Headers to pass to this endpoint
accept: application/json
forward_headers: True
# environment_variables:
# settings for using redis caching
# REDIS_HOST: redis-16337.c322.us-east-1-2.ec2.cloud.redislabs.com
# REDIS_PORT: "16337"
# REDIS_PASSWORD:
# litellm_settings:
# forward_client_headers_to_llm_api: true

View file

@ -0,0 +1,57 @@
#!/usr/bin/env python3
"""
Create a LiteLLM-managed OpenAI vector store from CLAUDE.md.
Requires OPENAI_API_KEY in environment or .env.
Usage:
python scripts/create_vector_store.py
Output: vector store ID (vs_xxx) to use in file_search tools.
"""
import os
import sys
from pathlib import Path
# Load .env if present
env_path = Path(__file__).resolve().parent.parent / ".env"
if env_path.exists():
for line in env_path.read_text().splitlines():
line = line.strip()
if line and not line.startswith("#") and "=" in line:
k, v = line.split("=", 1)
v = v.strip().strip('"').strip("'")
if " #" in v:
v = v.split(" #")[0].strip()
os.environ[k.strip()] = v
from openai import OpenAI
api_key = os.environ.get("OPENAI_API_KEY")
if not api_key:
print("ERROR: OPENAI_API_KEY not set. Add it to .env or export it.", file=sys.stderr)
sys.exit(1)
client = OpenAI(api_key=api_key)
repo_root = Path(__file__).resolve().parent.parent
doc_path = repo_root / "CLAUDE.md"
if not doc_path.exists():
print(f"ERROR: {doc_path} not found", file=sys.stderr)
sys.exit(1)
print("Uploading CLAUDE.md...")
with open(doc_path, "rb") as f:
file = client.files.create(file=f, purpose="assistants")
file_id = file.id
print(f"File ID: {file_id}")
print("Creating vector store...")
vs = client.vector_stores.create(
name="LiteLLM Docs Store",
file_ids=[file_id],
)
print(f"Vector Store ID: {vs.id}")
print()
print("=== Use this ID in file_search ===")
print(vs.id)

View file

@ -0,0 +1,405 @@
"""
Unit tests for FileSearchResponsesAPIUtils file_search tool interception in the
Responses API for non-native providers.
"""
import os
import sys
from typing import Any, Dict, List, Optional
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
sys.path.insert(
0, os.path.abspath("../../..")
) # Adds the parent directory to the system path
from litellm.llms.base_llm.responses.transformation import BaseResponsesAPIConfig
from litellm.llms.openai.responses.transformation import OpenAIResponsesAPIConfig
from litellm.responses.file_search_utils import (
CONTEXT_PREFIX,
FileSearchResponsesAPIUtils,
)
# ---------------------------------------------------------------------------
# helpers
# ---------------------------------------------------------------------------
def _make_file_search_tool(vector_store_ids: Optional[List[str]] = None) -> Dict[str, Any]:
tool: Dict[str, Any] = {"type": "file_search"}
if vector_store_ids is not None:
tool["vector_store_ids"] = vector_store_ids
return tool
def _make_function_tool(name: str = "my_func") -> Dict[str, Any]:
return {"type": "function", "name": name}
def _make_search_response(texts: List[str]) -> Dict[str, Any]:
return {
"data": [
{
"id": f"result_{i}",
"score": 0.9,
"content": [{"type": "text", "text": text}],
}
for i, text in enumerate(texts)
]
}
# ---------------------------------------------------------------------------
# has_file_search_tools
# ---------------------------------------------------------------------------
class TestHasFileSearchTools:
def test_detects_file_search_tool(self):
tools = [_make_file_search_tool(["vs_abc"])]
assert FileSearchResponsesAPIUtils.has_file_search_tools(tools) is True
def test_ignores_non_file_search_tool(self):
tools = [_make_function_tool()]
assert FileSearchResponsesAPIUtils.has_file_search_tools(tools) is False
def test_mixed_tools_detected(self):
tools = [_make_function_tool(), _make_file_search_tool(["vs_abc"])]
assert FileSearchResponsesAPIUtils.has_file_search_tools(tools) is True
def test_empty_list(self):
assert FileSearchResponsesAPIUtils.has_file_search_tools([]) is False
# ---------------------------------------------------------------------------
# split_file_search_tools
# ---------------------------------------------------------------------------
class TestSplitFileSearchTools:
def test_splits_correctly(self):
fs = _make_file_search_tool(["vs_1"])
fn = _make_function_tool()
file_search_tools, other_tools = FileSearchResponsesAPIUtils.split_file_search_tools([fs, fn])
assert file_search_tools == [fs]
assert other_tools == [fn]
def test_all_file_search(self):
tools = [_make_file_search_tool(["vs_1"]), _make_file_search_tool(["vs_2"])]
fs, others = FileSearchResponsesAPIUtils.split_file_search_tools(tools)
assert len(fs) == 2
assert others == []
def test_no_file_search(self):
tools = [_make_function_tool("a"), _make_function_tool("b")]
fs, others = FileSearchResponsesAPIUtils.split_file_search_tools(tools)
assert fs == []
assert others == tools
def test_empty(self):
fs, others = FileSearchResponsesAPIUtils.split_file_search_tools([])
assert fs == []
assert others == []
# ---------------------------------------------------------------------------
# extract_query_from_responses_input
# ---------------------------------------------------------------------------
class TestExtractQueryFromResponsesInput:
def test_string_input(self):
assert (
FileSearchResponsesAPIUtils.extract_query_from_responses_input("what is rag?")
== "what is rag?"
)
def test_list_input_last_user_message(self):
input_items = [
{"role": "assistant", "content": "Hello"},
{"role": "user", "content": "what is rag?"},
]
assert (
FileSearchResponsesAPIUtils.extract_query_from_responses_input(input_items)
== "what is rag?"
)
def test_list_input_structured_content(self):
input_items = [
{
"role": "user",
"content": [{"type": "text", "text": "explain vector search"}],
}
]
assert (
FileSearchResponsesAPIUtils.extract_query_from_responses_input(input_items)
== "explain vector search"
)
def test_empty_list_returns_none(self):
assert (
FileSearchResponsesAPIUtils.extract_query_from_responses_input([]) is None
)
def test_no_user_message_returns_none(self):
input_items = [{"role": "assistant", "content": "hi"}]
assert (
FileSearchResponsesAPIUtils.extract_query_from_responses_input(input_items)
is None
)
def test_non_list_non_str_returns_none(self):
assert (
FileSearchResponsesAPIUtils.extract_query_from_responses_input(None) is None # type: ignore
)
# ---------------------------------------------------------------------------
# inject_context_into_responses_input
# ---------------------------------------------------------------------------
class TestInjectContextIntoResponsesInput:
def test_string_input_becomes_list(self):
result = FileSearchResponsesAPIUtils.inject_context_into_responses_input(
"user question", "some context"
)
assert isinstance(result, list)
assert len(result) == 2
assert result[0]["content"] == "some context"
assert result[1]["content"] == "user question"
def test_list_input_context_before_last(self):
input_items = [
{"role": "user", "content": "first message"},
{"role": "user", "content": "last message"},
]
result = FileSearchResponsesAPIUtils.inject_context_into_responses_input(
input_items, "ctx"
)
assert isinstance(result, list)
assert len(result) == 3
assert result[1]["content"] == "ctx"
assert result[2]["content"] == "last message"
def test_single_item_list_context_prepended(self):
input_items = [{"role": "user", "content": "question"}]
result = FileSearchResponsesAPIUtils.inject_context_into_responses_input(
input_items, "ctx"
)
assert isinstance(result, list)
assert len(result) == 2
assert result[0]["content"] == "ctx"
assert result[1]["content"] == "question"
def test_empty_list_context_appended(self):
result = FileSearchResponsesAPIUtils.inject_context_into_responses_input([], "ctx")
assert isinstance(result, list)
assert len(result) == 1
assert result[0]["content"] == "ctx"
def test_original_input_not_mutated(self):
original = [{"role": "user", "content": "q"}]
FileSearchResponsesAPIUtils.inject_context_into_responses_input(original, "ctx")
assert len(original) == 1
# ---------------------------------------------------------------------------
# asearch_and_inject_context — unit / integration (mocked)
# ---------------------------------------------------------------------------
class TestAsearchAndInjectContext:
@pytest.mark.asyncio
async def test_searches_and_injects_context(self):
search_response = _make_search_response(["result text one", "result text two"])
with patch(
"litellm.vector_stores.asearch",
new=AsyncMock(return_value=search_response),
):
modified_input, remaining_tools = (
await FileSearchResponsesAPIUtils.asearch_and_inject_context(
input="what is rag?",
tools=[_make_file_search_tool(["vs_abc"])],
)
)
assert isinstance(modified_input, list)
assert remaining_tools == []
context_item = modified_input[0]
assert context_item["role"] == "user"
assert "result text one" in context_item["content"]
assert "result text two" in context_item["content"]
@pytest.mark.asyncio
async def test_file_search_tool_stripped_from_tools(self):
search_response = _make_search_response(["some context"])
fn_tool = _make_function_tool("other_fn")
with patch(
"litellm.vector_stores.asearch",
new=AsyncMock(return_value=search_response),
):
_, remaining_tools = await FileSearchResponsesAPIUtils.asearch_and_inject_context(
input="query",
tools=[_make_file_search_tool(["vs_1"]), fn_tool],
)
assert fn_tool in remaining_tools
assert not any(
t.get("type") == "file_search" for t in remaining_tools
)
@pytest.mark.asyncio
async def test_multiple_vector_stores_searched(self):
search_response = _make_search_response(["doc"])
mock_search = AsyncMock(return_value=search_response)
with patch("litellm.vector_stores.asearch", new=mock_search):
await FileSearchResponsesAPIUtils.asearch_and_inject_context(
input="query",
tools=[_make_file_search_tool(["vs_1", "vs_2"])],
)
assert mock_search.call_count == 2
called_ids = {call.kwargs["vector_store_id"] for call in mock_search.call_args_list}
assert called_ids == {"vs_1", "vs_2"}
@pytest.mark.asyncio
async def test_no_query_returns_unchanged(self):
original_input: List[Dict[str, Any]] = []
tools = [_make_file_search_tool(["vs_1"])]
modified_input, remaining_tools = (
await FileSearchResponsesAPIUtils.asearch_and_inject_context(
input=original_input,
tools=tools,
)
)
assert modified_input == original_input
# file_search tools stripped even when no query
assert not any(t.get("type") == "file_search" for t in remaining_tools)
@pytest.mark.asyncio
async def test_no_vector_store_ids_strips_tool_only(self):
tools = [_make_file_search_tool()] # no vector_store_ids key
modified_input, remaining_tools = (
await FileSearchResponsesAPIUtils.asearch_and_inject_context(
input="query",
tools=tools,
)
)
assert modified_input == "query"
assert remaining_tools == []
@pytest.mark.asyncio
async def test_search_error_does_not_raise(self):
with patch(
"litellm.vector_stores.asearch",
new=AsyncMock(side_effect=Exception("timeout")),
):
modified_input, remaining_tools = (
await FileSearchResponsesAPIUtils.asearch_and_inject_context(
input="query",
tools=[_make_file_search_tool(["vs_err"])],
)
)
# Falls back gracefully: input unchanged, file_search stripped
assert modified_input == "query"
assert remaining_tools == []
@pytest.mark.asyncio
async def test_stores_results_in_logging_obj(self):
search_response = _make_search_response(["relevant doc"])
logging_obj = MagicMock()
logging_obj.model_call_details = {}
with patch(
"litellm.vector_stores.asearch",
new=AsyncMock(return_value=search_response),
):
await FileSearchResponsesAPIUtils.asearch_and_inject_context(
input="query",
tools=[_make_file_search_tool(["vs_abc"])],
litellm_logging_obj=logging_obj,
)
assert "search_results" in logging_obj.model_call_details
assert logging_obj.model_call_details["search_results"] == [search_response]
# ---------------------------------------------------------------------------
# supports_native_file_search — capability flag tests
# ---------------------------------------------------------------------------
class TestSupportsNativeFileSearch:
def test_base_config_returns_false(self):
"""All non-overriding providers default to False."""
class _MinimalConfig(BaseResponsesAPIConfig):
@property
def custom_llm_provider(self):
return "test_provider"
def get_supported_openai_params(self, model):
return []
def map_openai_params(self, response_api_optional_params, model, drop_params):
return {}
def validate_environment(self, headers, model, litellm_params):
return {}
def get_complete_url(self, api_base, litellm_params):
return api_base or ""
def transform_responses_api_request(self, model, input, response_api_optional_request_params, litellm_params, headers):
return {}
def transform_response_api_response(self, model, raw_response, logging_obj):
return MagicMock()
def transform_streaming_response(self, model, parsed_chunk, logging_obj):
return MagicMock()
def transform_delete_response_api_request(self, response_id, api_base, litellm_params, headers):
return "", {}
def transform_delete_response_api_response(self, raw_response, logging_obj):
return MagicMock()
def transform_get_response_api_request(self, response_id, api_base, litellm_params, headers):
return "", {}
def transform_get_response_api_response(self, raw_response, logging_obj):
return MagicMock()
def transform_list_input_items_request(self, response_id, api_base, litellm_params, headers, after=None, before=None, include=None, limit=20, order="desc"):
return "", {}
def transform_list_input_items_response(self, raw_response, logging_obj):
return {}
def transform_cancel_response_api_request(self, response_id, api_base, litellm_params, headers):
return "", {}
def transform_cancel_response_api_response(self, raw_response, logging_obj):
return MagicMock()
def transform_compact_response_api_request(self, model, input, response_api_optional_request_params, api_base, litellm_params, headers):
return "", {}
def transform_compact_response_api_response(self, raw_response, logging_obj):
return MagicMock()
assert _MinimalConfig().supports_native_file_search() is False
def test_openai_config_returns_true(self):
assert OpenAIResponsesAPIConfig().supports_native_file_search() is True
def test_azure_config_inherits_true(self):
from litellm.llms.azure.responses.transformation import (
AzureOpenAIResponsesAPIConfig,
)
assert AzureOpenAIResponsesAPIConfig().supports_native_file_search() is True