How the Embedding Module in AgentScope Works: Vector Storage and Retrieval Explained

The embedding module in AgentScope provides a plug-and-play abstraction for converting text into dense vectors, featuring provider-specific implementations like OpenAITextEmbedding and a built-in caching layer via FileEmbeddingCache to minimize API costs and reduce latency.

The embedding module in AgentScope serves as the foundation for vector-based operations within the multi-agent framework, enabling seamless integration of text embedding models with persistent storage capabilities. Located in src/agentscope/embedding/, this subsystem abstracts away provider-specific API details while offering robust caching mechanisms to avoid redundant external calls. Whether you are building retrieval-augmented generation pipelines or semantic search features, understanding this module's architecture is essential for optimizing performance and cost.

Architecture Overview

The embedding module in AgentScope follows a layered design pattern that separates concerns between model inference, data persistence, and response formatting. This architecture enables developers to swap embedding providers or cache implementations without modifying downstream application logic.

Model Abstraction Layer

At the core of the system lies EmbeddingModelBase, an abstract base class defined in _embedding_base.py that establishes the contract for all embedding providers. This class declares essential attributes including model_name, dimensions, and supported_modalities, while enforcing the implementation of an asynchronous __call__ method that must return an EmbeddingResponse object.

According to the AgentScope source code in src/agentscope/embedding/_embedding_base.py (lines 8-14), the base constructor stores the model name and dimensionality, providing a consistent interface for querying model capabilities before invocation.

Provider Implementation

Concrete implementations like OpenAITextEmbedding in _openai_embedding.py inherit from EmbeddingModelBase and handle provider-specific API interactions. This class accepts configuration parameters including api_key and an optional embedding_cache instance.

As implemented in src/agentscope/embedding/_openai_embedding.py (lines 19-27), the constructor initializes an openai.AsyncClient for non-blocking API calls. The __call__ method (lines 79-92) first normalizes input text, then checks the cache via self.embedding_cache.retrieve(). On a cache miss, it invokes self.client.embeddings.create(), measures latency, and stores the result via self.embedding_cache.store() for future reuse.

Caching Layer

The caching subsystem centers on EmbeddingCacheBase in _cache_base.py, which defines the asynchronous interface for vector storage operations. This abstract class specifies four core methods: store(identifier, embeddings, overwrite=False), retrieve(identifier), remove(identifier), and clear().

For production deployments, FileEmbeddingCache in _file_cache.py provides a concrete implementation that persists embeddings as JSON mappings on disk. As shown in lines 19-55 of src/agentscope/embedding/_file_cache.py, this class maintains a dictionary mapping identifier hashes to embedding vectors, using asyncio.Lock to handle concurrent access safely.

Response Handling

All embedding operations return an EmbeddingResponse object defined in _embedding_response.py. This dataclass encapsulates the resulting vectors along with metadata including id, created_at, type, optional usage statistics, and a critical source field indicating whether the data originated from "cache" or "api".

The EmbeddingUsage class in _embedding_usage.py complements this by tracking token counts and execution time, enabling performance monitoring and cost analysis across both cached and live API calls.

End-to-End Workflow

When invoking the embedding module in AgentScope, the following sequence occurs:

1. Request Normalization: The user passes text inputs to an embedding model instance (e.g., OpenAITextEmbedding).

2. Cache Lookup: The model generates a unique identifier from the request parameters (model name, input text, dimensions) and queries the configured EmbeddingCacheBase implementation.

3. Cache Hit: If retrieve() returns vectors, the model immediately returns an EmbeddingResponse with source="cache" and zero API latency.

4. Cache Miss: If no cached entry exists, the model calls the external API (e.g., OpenAI's embeddings.create), constructs an EmbeddingResponse with source="api", and persists the vectors via cache.store() for subsequent requests.

This flow ensures that identical embedding requests are answered instantly from local storage while maintaining API freshness for novel inputs.

Practical Implementation Examples

Basic Usage with OpenAI and File Cache

The most common pattern involves initializing a FileEmbeddingCache alongside an OpenAITextEmbedding model:

from agentscope.embedding import OpenAITextEmbedding, FileEmbeddingCache

# Initialize persistent file cache

cache = FileEmbeddingCache(cache_file="my_embeddings.json")

# Create embedding model with caching enabled

embedder = OpenAITextEmbedding(
    api_key="sk-*****",  # Replace with your API key

    model_name="text-embedding-3-large",
    dimensions=1536,
    embedding_cache=cache,
)

# First call hits the API; second call retrieves from cache

texts = ["Hello world!", "AgentScope enables multi-agent systems."]
response1 = await embedder(texts)  # source="api"

response2 = await embedder(texts)  # source="cache"

print(f"Vector dimensions: {len(response2.embeddings[0])}")
print(f"Data source: {response2.source}")

Direct Cache Operations

For advanced use cases requiring manual cache management, interact directly with FileEmbeddingCache:

from agentscope.embedding import FileEmbeddingCache, EmbeddingResponse, EmbeddingUsage

cache = FileEmbeddingCache(cache_file="demo.json")

# Manually store pre-computed vectors

identifier = {
    "model": "custom-model-v1",
    "input": ["query one", "query two"],
    "dimensions": 3,
}
vectors = [[0.1, 0.2, 0.3], [0.4, 0.5, 0.6]]
await cache.store(identifier=identifier, embeddings=vectors)

# Retrieve and wrap in response object

cached_vectors = await cache.retrieve(identifier=identifier)
response = EmbeddingResponse(
    embeddings=cached_vectors,
    usage=EmbeddingUsage(tokens=0, time=0),
    source="cache",
)

Provider Swapping

The abstraction layer allows seamless substitution of embedding providers without changing application logic:

from agentscope.embedding import OpenAITextEmbedding, FileEmbeddingCache

# Hypothetical alternative provider

# from agentscope.embedding import GeminiEmbedding

cache = FileEmbeddingCache()

# Initialize different providers with the same cache interface

openai_embedder = OpenAITextEmbedding(
    api_key="sk-...",
    embedding_cache=cache,
)

# gemini_embedder = GeminiEmbedding(api_key="...", embedding_cache=cache)

# Both implement the same interface: async __call__(texts) -> EmbeddingResponse

response = await openai_embedder(["Sample text for embedding"])

Key Source Files

The embedding module in AgentScope is organized under src/agentscope/embedding/ with the following critical components:

• _embedding_base.py – Defines EmbeddingModelBase, the abstract contract for all embedding models including attribute declarations for model_name, dimensions, and supported_modalities.
• _openai_embedding.py – Implements OpenAITextEmbedding, the concrete OpenAI provider with async client initialization and cache-aware request handling.
• _cache_base.py – Specifies EmbeddingCacheBase, the abstract interface for vector storage operations including store(), retrieve(), remove(), and clear().
• _file_cache.py – Provides FileEmbeddingCache, the JSON-backed persistent cache implementation using asyncio.Lock for thread safety.
• _embedding_response.py – Contains EmbeddingResponse, the dataclass wrapping embedding vectors with metadata including the critical source field distinguishing cache from API origins.
• _embedding_usage.py – Houses EmbeddingUsage, tracking token counts and execution latency for performance monitoring.
• __init__.py – Public API exports for the embedding subsystem.

Summary

The embedding module in AgentScope delivers a robust, extensible framework for vector storage and retrieval through these key design principles:

• Abstraction Layers – EmbeddingModelBase and EmbeddingCacheBase decouple provider-specific implementations from application logic, enabling seamless swapping of embedding models or storage backends.
• Intelligent Caching – FileEmbeddingCache persists vectors to JSON with automatic identifier hashing, eliminating redundant API calls and reducing costs while maintaining asyncio.Lock for concurrency safety.
• Transparent Provenance – Every EmbeddingResponse includes a source field indicating "cache" or "api", allowing applications to track data lineage and optimize cache hit rates.
• Async-First Design – All cache and model operations use async/await patterns, ensuring non-blocking I/O for high-throughput multi-agent workflows.

Frequently Asked Questions

How does the embedding module in AgentScope handle cache misses?

When a cache miss occurs, the concrete embedding model (such as OpenAITextEmbedding) calls the external API using the provider's async client, measures the request latency, constructs an EmbeddingResponse with source="api", and then persists the resulting vectors to the cache via EmbeddingCacheBase.store() for future retrieval.

Can I use the caching layer without calling external APIs?

Yes, the EmbeddingCacheBase abstraction allows direct storage and retrieval of pre-computed vectors. You can instantiate FileEmbeddingCache and manually call await cache.store(identifier, embeddings) to populate the cache, then retrieve them later with await cache.retrieve(identifier) without ever invoking an external embedding provider.

What identifier format ensures cache hits across identical requests?

The embedding module generates identifiers from the complete request payload including model name, input text, and dimensions parameters. This ensures that only truly identical requests (same model, same text, same output dimensions) return cached results, preventing collisions between different embedding configurations.

How does the module ensure thread safety when accessing the file cache?

The FileEmbeddingCache implementation uses asyncio.Lock to synchronize concurrent access to the underlying JSON file. This ensures that multiple async tasks can safely read from and write to the cache simultaneously without corrupting the persistent storage, making it suitable for high-concurrency multi-agent applications.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}