
Alibabacloud Pds Multimodal Search
- 148 installs
- 208 repo stars
- Updated August 4, 2026
- aliyun/alibabacloud-aiops-skills
Wire Alibaba Cloud PDS multimodal search into agents so users can query documents, images, and other stored assets with natural language across a drive or content library.
About
Agent skill for Alibaba Cloud PDS multimodal search that helps developers integrate semantic retrieval over files and media in Personal Drive Service. It supports building AI assistants that find documents, images, and mixed content via cloud APIs rather than local indexing alone.
- PDS multimodal retrieval
- Natural-language asset search
- Cross-modal file discovery
- Agent-callable cloud APIs
- Enterprise drive integration
Alibabacloud Pds Multimodal Search by the numbers
- 148 all-time installs (skills.sh)
- Ranked #3,363 of 16,546 AI & Agent Building skills by installs in the Skillselion catalog
- Data as of Aug 5, 2026 (Skillselion catalog sync)
npx skills add https://github.com/aliyun/alibabacloud-aiops-skills --skill alibabacloud-pds-multimodal-searchAdd your badge
Show developers this skill is listed on Skillselion. Paste this into your README.
| Installs | 148 |
|---|---|
| repo stars | ★ 208 |
| Last updated | August 4, 2026 |
| Repository | aliyun/alibabacloud-aiops-skills ↗ |
What it does
Wire Alibaba Cloud PDS multimodal search into agents so users can query documents, images, and other stored assets with natural language across a drive or content library.
Files
PDS Multimodal Search
Please read this entire skill document carefully
Features
- For getting drive/drive_id, querying enterprise space, team space, personal space -> read
references/drive.md - For uploading local files to enterprise space, team space, personal space → read
references/upload-file.md - For downloading files from enterprise space, team space, personal space to local → read
references/download-file.md - For searching or finding files → read
references/search-file.md - For document/audio/video analysis, quick view, summarization on cloud drive → read
references/multianalysis-file.md - For image search, similar image search, image-text hybrid retrieval → read
references/visual-similar-search.md
Agent Execution Guidelines
- Must execute steps in order: Do not skip any step, do not proceed to the next step before the previous one is completed.
- Must follow documentation: The aliyun pds cli commands and parameters must follow this document's guidance, do not fabricate commands.
- [MUST] CLI User-Agent — Every
aliyunCLI command invocation must include:
--user-agent AlibabaCloud-Agent-Skills/alibabacloud-pds-multimodal-search
- Must determine the target space before file operations: Before search, upload, download, or analysis, first decide whether the user explicitly means enterprise space, team space, personal space, or all spaces.
- Space scope must not be broadened silently: If the user explicitly says "enterprise space", only use the enterprise space drive_id. If the user explicitly says "team space", only use the matching team space drive_id. If the user explicitly says "personal space", only use the personal space drive_id. Only search across multiple spaces when the user did not restrict the scope.
- Enterprise space and team space are not interchangeable: Even though both are returned by
list-my-group-drive,root_group_driveis the enterprise space anditemsare team spaces. Never substitute one for the other. - If the requested space is missing, stop and explain: For example, if the user asks for enterprise space but
root_group_driveis empty, do not fall back to a team space search.
Core Concepts:
- Domain: PDS instance with a unique domain_id, data is completely isolated between domains
- User: End user under a domain, has user_id
- Group: Team organization under a domain, divided into enterprise group and team group
- Drive: Storage space, can belong to a user (personal space) or group (enterprise space or team space)
- File: File or folder under a space, has file_id
- Mountapp: PDS mount app plugin, used to mount PDS space to local, allowing users to access and manage files in PDS space conveniently
Space Selection Rules
Apply the following rules before choosing a drive_id:
| User wording | Allowed drive source | Forbidden fallback |
|---|---|---|
| "企业空间" / "company space" / "enterprise space" | root_group_drive only | Any drive from items |
| "团队空间" / "某个团队空间" / "team space" | items only | root_group_drive |
| "个人空间" / "我的空间" / "personal space" | list-my-drives.items only | group drives |
| "网盘里" / "我的网盘" / no space specified | all relevant spaces | none |
Before continuing, perform a brief self-check: 1. Did the user explicitly name the target space type? 2. Does the selected drive_id come from the correct response field for that space type? 3. If multiple team spaces exist and the user only said "team space", do I need to disambiguate which team space?
If any answer is uncertain, do not guess.
---
Installation Requirements
Prerequisites: Requires Aliyun CLI >= 3.3.3
>
Verify CLI version:
```bash
aliyun version # requires >= 3.3.3
```
>
If not installed or version too low,
run curl -fsSL https://aliyuncli.alicdn.com/setup.sh | bash to update,or see references/cli-installation-guide.md for installation instructions.>
Verify PDS plugin version:
```bash
aliyun pds version # requires >= 0.1.4
```
>
After installation, must enable auto plugin installation:
```bash
aliyun configure set --auto-plugin-install true
```
>
Then [MUST] run aliyun plugin update to ensure that any existing plugins on your local machine are always up-to-date.---
CLI Initialization (MUST run before Core Workflow)
At the start of the Core Workflow (before any CLI invocation):
[MUST] Enable AI-Mode — AI-mode is required for Agent Skill execution. Run the following commands before any CLI invocation:
aliyun configure ai-mode enable
aliyun configure ai-mode set-user-agent --user-agent "AlibabaCloud-Agent-Skills/alibabacloud-pds-multimodal-search"[MUST] Disable AI-Mode at EVERY exit point — Before delivering the final response for ANY reason, always disable AI-mode first. This applies to ALL exit paths: workflow success, workflow failure, error/exception, user cancellation, session end, or any other scenario where no further CLI commands will be executed. AI-mode is only used for Agent Skill invocation scenarios and MUST NOT remain enabled after the skill stops running.
aliyun configure ai-mode disable---
Authentication Configuration
Prerequisites: Alibaba Cloud credentials must be configured
>
Security Rules:
- Forbidden to read, output, or print AK/SK values (e.g., echo $ALIBABA_CLOUD_ACCESS_KEY_ID is forbidden)- Forbidden to ask users to input AK/SK directly in conversation or command line
- Forbidden to use aliyun configure set to set plaintext credentials- Only allowed to use aliyun configure list to check credential status>
Check credential configuration:
```bash
aliyun configure list
```
>
Confirm the output shows a valid profile (AK, STS, or OAuth identity).
>
If no valid configuration exists, stop first.
1. Obtain credentials from Alibaba Cloud Console
2. Configure credentials outside this session (run aliyun configure in terminal or set environment variables)3. Run aliyun configure list to verify after configuration is complete# Install Aliyun CLI (if not installed)
curl -fsSL --max-time 10 https://aliyuncli.alicdn.com/install.sh | bash
aliyun version # confirm >= 3.3.3
# Enable auto plugin installation
aliyun configure set --auto-plugin-install true
# Install Python dependencies (for multipart upload script)
pip3 install requestsPDS-Specific Configuration
Before executing any PDS operations, you must first configure domain_id, user_id, and authentication type -> read references/config.md
[MUST] CLI User-Agent — EveryaliyunCLI command invocation must include--user-agent AlibabaCloud-Agent-Skills/alibabacloud-pds-multimodal-search
>
Examples:
```bash
aliyun pds get-user --user-agent AlibabaCloud-Agent-Skills/alibabacloud-pds-multimodal-search
aliyun pds list-my-drives --user-agent AlibabaCloud-Agent-Skills/alibabacloud-pds-multimodal-search
aliyun pds upload-file --drive-id <id> --local-path <path> --user-agent AlibabaCloud-Agent-Skills/alibabacloud-pds-multimodal-search
```
References
| Reference Document | Path |
|---|---|
| CLI Installation Guide | references/cli-installation-guide.md |
| RAM Permission Policies | references/ram-policies.md |
Error Handling
1. If file search fails, please read references/search-file.md and strictly follow the documented process to re-execute file search.
Aliyun CLI Installation & Configuration Guide
Complete guide for installing and configuring Aliyun CLI.
Aliyun CLI 3.3.3+: Supports installing and using all published Alibaba Cloud product plugins. Make sure to upgrade to 3.3.3 or later for full plugin ecosystem coverage.
Installation
macOS
Using Homebrew (Recommended)
brew install aliyun-cli
# Upgrade to latest
brew upgrade aliyun-cli
# Verify version (>= 3.3.3)
aliyun versionUsing Binary
# Download
wget --timeout=600 https://aliyuncli.alicdn.com/aliyun-cli-macosx-latest-amd64.tgz
# Extract
tar -xzf aliyun-cli-macosx-latest-amd64.tgz
# Move to PATH
sudo mv aliyun /usr/local/bin/
# Verify
aliyun versionLinux
Debian/Ubuntu
# Download
wget --timeout=600 https://aliyuncli.alicdn.com/aliyun-cli-linux-latest-amd64.tgz
# Extract and install
tar -xzf aliyun-cli-linux-latest-amd64.tgz
sudo mv aliyun /usr/local/bin/
# Verify
aliyun versionCentOS/RHEL
# Download
wget --timeout=600 https://aliyuncli.alicdn.com/aliyun-cli-linux-latest-amd64.tgz
# Extract and install
tar -xzf aliyun-cli-linux-latest-amd64.tgz
sudo mv aliyun /usr/local/bin/
# Verify
aliyun versionARM64 Architecture
# Download ARM64 version
wget --timeout=600 https://aliyuncli.alicdn.com/aliyun-cli-linux-latest-arm64.tgz
# Extract and install
tar -xzf aliyun-cli-linux-latest-arm64.tgz
sudo mv aliyun /usr/local/bin/Windows
Using Binary 1. Download from: https://aliyuncli.alicdn.com/aliyun-cli-windows-latest-amd64.zip 2. Extract the ZIP file 3. Add the directory to your PATH environment variable 4. Open new Command Prompt or PowerShell 5. Verify: aliyun version
Using PowerShell
# Download
Invoke-WebRequest -Uri "https://aliyuncli.alicdn.com/aliyun-cli-windows-latest-amd64.zip" -OutFile "aliyun-cli.zip"
# Extract
Expand-Archive -Path aliyun-cli.zip -DestinationPath C:\aliyun-cli
# Add to PATH (requires admin privileges)
$env:Path += ";C:\aliyun-cli"
[Environment]::SetEnvironmentVariable("Path", $env:Path, [System.EnvironmentVariableTarget]::Machine)
# Verify
aliyun versionConfiguration
Quick Start
aliyun configure set \
--mode AK \
--access-key-id <your-access-key-id> \
--access-key-secret <your-access-key-secret> \
--region cn-hangzhouAll aliyun configure commands support non-interactive flags, which is the recommended approach — it works in scripts, CI/CD pipelines, and agent-driven automation without hanging on stdin prompts.
Where to Get Access Keys
1. Log in to Aliyun Console: https://ram.console.aliyun.com/ 2. Navigate to: AccessKey Management 3. Create a new AccessKey pair 4. Save the secret immediately — it's only shown once
Configuration Modes
Aliyun CLI supports 6 authentication modes. All examples below use non-interactive flags.
1. AK Mode (Access Key)
Most common mode for personal accounts and scripts.
aliyun configure set \
--mode AK \
--access-key-id LTAI5tXXXXXXXX \
--access-key-secret 8dXXXXXXXXXXXXXXXXXXXXXXXX \
--region cn-hangzhouConfiguration is stored in ~/.aliyun/config.json:
{
"current": "default",
"profiles": [
{
"name": "default",
"mode": "AK",
"access_key_id": "LTAI5tXXXXXXXX",
"access_key_secret": "8dXXXXXXXXXXXXXXXXXXXXXXXX",
"region_id": "cn-hangzhou",
"output_format": "json",
"language": "en"
}
]
}2. StsToken Mode (Temporary Credentials)
For short-lived access (tokens expire in 1-12 hours).
aliyun configure set \
--mode StsToken \
--access-key-id LTAI5tXXXXXXXX \
--access-key-secret 8dXXXXXXXXXXXXXXXXXXXXXXXX \
--sts-token v1.0:XXXXXXXXXXXXXXXX \
--region cn-hangzhouUse cases: CI/CD pipelines, temporary access for external contractors, cross-account access.
3. RamRoleArn Mode (Assume RAM Role)
Assume a RAM role for elevated or cross-account access.
aliyun configure set \
--mode RamRoleArn \
--access-key-id LTAI5tXXXXXXXX \
--access-key-secret 8dXXXXXXXXXXXXXXXXXXXXXXXX \
--ram-role-arn acs:ram::123456789012:role/AdminRole \
--role-session-name my-session \
--region cn-hangzhouUse cases: cross-account resource access, temporary elevated privileges, role-based access control.
4. EcsRamRole Mode (ECS Instance RAM Role)
Use the RAM role attached to an ECS instance — no credentials needed.
aliyun configure set \
--mode EcsRamRole \
--ram-role-name MyEcsRole \
--region cn-hangzhouRequirements: must be running on an ECS instance with a RAM role attached.
Use cases: scripts and automation running on ECS instances.
5. RsaKeyPair Mode (RSA Key Pair)
Use RSA key pair for authentication (generate key pair in Aliyun Console first).
aliyun configure set \
--mode RsaKeyPair \
--private-key /path/to/private-key.pem \
--key-pair-name my-key-pair \
--region cn-hangzhou6. RamRoleArnWithEcs Mode (ECS + RAM Role)
Combine ECS instance role with RAM role assumption for cross-account access from ECS.
aliyun configure set \
--mode RamRoleArnWithEcs \
--ram-role-name MyEcsRole \
--ram-role-arn acs:ram::123456789012:role/TargetRole \
--role-session-name my-session \
--region cn-hangzhouEnvironment Variables
Highest priority - overrides config file
Access Key Mode
export ALIBABA_CLOUD_ACCESS_KEY_ID=your_access_key_id
export ALIBABA_CLOUD_ACCESS_KEY_SECRET=your_access_key_secret
export ALIBABA_CLOUD_REGION_ID=cn-hangzhouSTS Token Mode
export ALIBABA_CLOUD_ACCESS_KEY_ID=your_access_key_id
export ALIBABA_CLOUD_ACCESS_KEY_SECRET=your_access_key_secret
export ALIBABA_CLOUD_SECURITY_TOKEN=your_sts_token
export ALIBABA_CLOUD_REGION_ID=cn-hangzhouECS RAM Role Mode
export ALIBABA_CLOUD_ECS_METADATA=role_nameUse Case:
- CI/CD pipelines
- Docker containers
- Temporary credential override
Managing Multiple Profiles
Create Named Profiles
aliyun configure set --profile projectA \
--mode AK \
--access-key-id LTAI5tAAAAAAAA \
--access-key-secret 8dAAAAAAAAAAAAAAAAAAAAAAAA \
--region cn-hangzhou
aliyun configure set --profile projectB \
--mode AK \
--access-key-id LTAI5tBBBBBBBB \
--access-key-secret 8dBBBBBBBBBBBBBBBBBBBBBBBB \
--region cn-shanghaiUse Specific Profile
aliyun ecs describe-instances --profile projectA
export ALIBABA_CLOUD_PROFILE=projectA
aliyun ecs describe-instances # Uses projectAList and Switch Profiles
aliyun configure list # List all profiles
aliyun configure set --current projectA # Switch default profileCredential Priority
Credentials are loaded in this order (first found wins):
1. Command-line flag: --profile <name> 2. Environment variable: ALIBABA_CLOUD_PROFILE 3. Environment credentials: ALIBABA_CLOUD_ACCESS_KEY_ID, etc. 4. Configuration file: ~/.aliyun/config.json (current profile) 5. ECS Instance RAM Role: If running on ECS with attached role
Verification
Test Authentication
# Basic test - list regions
aliyun ecs describe-regions
# Expected output: JSON array of regionsIf successful, you'll see:
{
"Regions": {
"Region": [
{
"RegionId": "cn-hangzhou",
"RegionEndpoint": "ecs.cn-hangzhou.aliyuncs.com",
"LocalName": "华东 1(杭州)"
},
...
]
},
"RequestId": "..."
}If failed, you'll see error messages:
InvalidAccessKeyId.NotFound- Wrong Access Key IDSignatureDoesNotMatch- Wrong Access Key SecretInvalidSecurityToken.Expired- STS token expired (for StsToken mode)Forbidden.RAM- Insufficient permissions
Debug Configuration
# Show current configuration
aliyun configure get
# Test with debug logging
aliyun ecs describe-regions --log-level=debug
# Check credential provider
aliyun configure get modeSecurity Best Practices
1. Use RAM Users (Not Root Account)
❌ Don't: Use Aliyun root account credentials ✅ Do: Create RAM users with specific permissions
# Create RAM user in console
# Attach only necessary policies
# Use RAM user's access keys2. Principle of Least Privilege
Grant only the minimum permissions needed:
# Example: Read-only ECS access
# Attach policy: AliyunECSReadOnlyAccess3. Rotate Access Keys Regularly
# Create new access key in RAM Console, then update configuration
aliyun configure set --access-key-id NEW_KEY --access-key-secret NEW_SECRET
# Delete old access key from console4. Use STS Tokens for Temporary Access
aliyun configure set --mode StsToken \
--access-key-id XXXX --access-key-secret XXXX \
--sts-token XXXX --region cn-hangzhou5. Use ECS RAM Roles When Possible
aliyun configure set --mode EcsRamRole --ram-role-name MyRole --region cn-hangzhou6. Never Commit Credentials
# Add to .gitignore
echo "~/.aliyun/config.json" >> .gitignore
# Use environment variables in CI/CD instead7. Secure Config File
# Restrict permissions
chmod 600 ~/.aliyun/config.jsonTroubleshooting
Issue: Command Not Found
# Check installation
which aliyun
# Check PATH
echo $PATH
# Reinstall or add to PATHIssue: Authentication Failed
# Verify configuration
aliyun configure get
# Test with debug
aliyun ecs describe-regions --log-level=debug
# Check credentials in console
# Verify access key is activeIssue: Permission Denied
# Error: Forbidden.RAM
# Check RAM user permissions
# Attach necessary policies in RAM console
# Example: AliyunECSFullAccess for ECS operationsIssue: STS Token Expired
# Error: InvalidSecurityToken.Expired
# Reconfigure with new token
aliyun configure set --mode StsToken \
--access-key-id XXXX --access-key-secret XXXX \
--sts-token NEW_TOKEN --region cn-hangzhouIssue: Wrong Region
# Some resources may not exist in the specified region
# Check available regions
aliyun ecs describe-regions
# Update default region
aliyun configure set region cn-shanghaiAdvanced Configuration
Custom Endpoint
# Use custom or private endpoint
export ALIBABA_CLOUD_ECS_ENDPOINT=ecs-vpc.cn-hangzhou.aliyuncs.comProxy Settings
# HTTP proxy
export HTTP_PROXY=http://proxy.example.com:8080
export HTTPS_PROXY=http://proxy.example.com:8080
# No proxy for specific domains
export NO_PROXY=localhost,127.0.0.1,.aliyuncs.comTimeout Settings
# Connection timeout (default: 10s)
export ALIBABA_CLOUD_CONNECT_TIMEOUT=30
# Read timeout (default: 10s)
export ALIBABA_CLOUD_READ_TIMEOUT=30Next Steps
After installation and configuration:
1. Install plugins for services you need (v3.3.3+ supports all published product plugins):
aliyun plugin install --names ecs vpc rds
# List all available plugins
aliyun plugin list-remote2. Explore commands:
aliyun ecs --help
aliyun fc --help3. Read documentation:
- Command Syntax Guide
- Global Flags Reference
- Common Scenarios
References
- Official Documentation: https://help.aliyun.com/zh/cli/
- RAM Console: https://ram.console.aliyun.com/
- Access Key Management: https://ram.console.aliyun.com/manage/ak
- Plugin Repository: https://github.com/aliyun/aliyun-cli
PDS Aliyun CLI Configuration Guide (Important)
Scenario: Required configuration when using aliyun pds cli for the first time Purpose: Configure domain_id, user_id, and authentication type for aliyun pds cli
---
Before executing any PDS operations, you must first configure domain_id, user_id, and authentication type:
Step 1: Verify if configuration already exists (only needs to be configured once during initialization)
aliyun pds get-user --user-agent AlibabaCloud-Agent-Skills/alibabacloud-pds-multimodal-searchIf already configured successfully, it will return the current logged-in user information, and you can skip the subsequent steps.
Step 2: Query domain list using aliyun pds list-domains (skip this step if you already have the domain_id to configure)
aliyun pds list-domains --service-code edm --limit 100 --region cn-beijing --user-agent AlibabaCloud-Agent-Skills/alibabacloud-pds-multimodal-searchThe returned JSON structure is as follows. Extract the domain list from the response and display it to the user in a table format with columns domain_id and domain_name, prompting the user to select one domain. (If there is only one domain, use it directly without asking)
{
"items": [{
"domain_id": "bj322",
"domain_name": "beijing-31216",
"region_id": "cn-beijing",
"service_code": "edm"
}],
"next_marker": ""
}This step requires obtaining the selected domain_id before proceeding to the next step.
Step 3: Query user list under the domain using aliyun pds list-user (skip this step if you already have the user_id to configure)
# First configure domain_id with ak authentication type
aliyun pds config --domain-id <domain_id> --authentication-type ak --user-agent AlibabaCloud-Agent-Skills/alibabacloud-pds-multimodal-search
# Then list users under this domain
aliyun pds list-user --limit 100 --user-agent AlibabaCloud-Agent-Skills/alibabacloud-pds-multimodal-searchThe returned JSON structure is as follows. Extract the user list from the response and display it to the user in a table format with columns user_id, nick_name, phone, email, and role, prompting the user to select one user. (If there is only one user, use it directly without asking)
{
"items": [
{
"nick_name": "SuperAdmin",
"role": "superadmin",
"status": "enabled",
"updated_at": 1774159173066,
"phone": "123",
"email": "test@example.com",
"user_id": "a34527bd247e48b6b7e48d5c381b23f3"
}
],
"next_marker": ""
}This step requires obtaining the selected user_id before proceeding to the next step.
Step 4: Configure domain_id, user_id, and authentication type to aliyun pds cli using aliyun pds config
aliyun pds config \
--domain-id <domain_id> \
--user-id <user_id> \
--authentication-type token \
--user-agent AlibabaCloud-Agent-Skills/alibabacloud-pds-multimodal-searchParameter Description:
--domain-id: PDS domain ID (e.g.,bj31216), provided by PDS user, check if included in the prompt--user-id: PDS user ID (e.g.,a34527bd247e48b6b7e48d5c381b23f3), provided by PDS user, check if included in the prompt--authentication-type: Must be set to `token` if user_id parameter is provided, indicating access with user identity
Effect After Configuration:
- No need to pass
--domain-idparameter for subsequent PDS API calls - CLI will automatically use the configured domain_id and user_id
Verify Configuration:
# Test if configuration is effective, get-user API without parameters returns current logged-in user information in token scenario
aliyun pds get-user --user-agent AlibabaCloud-Agent-Skills/alibabacloud-pds-multimodal-searchExtract the current logged-in user information from the returned JSON: domain_id: domain_id, user_id: user_id, nick_name: nick_name.
After successful configuration, notify the user: Current PDS DomainID: <domain_id>, logged-in user: <nick_name>(<user_id>)
Notes:
- Domain_id and user_id will be preset in CLI configuration
- User's token will be preset in Aliyun CLI configuration file
- After configuring once, no need to repeat configuration for subsequent operations
---
PDS File Download Guide
Scenario: When you have obtained the drive_id and file_id of the file to download and need to download that file Purpose: Download file to local
---
Get File ID from File Path
If you want to download a file from a PDS drive but only have the file path (e.g., /Photos/2026/04/vacation.jpg), you need to traverse each level of the path to find the corresponding file's file_id. The steps are as follows: For example, to download the file /Photos/2026/04/vacation.jpg from a personal space:
1. First, use the aliyun pds list-file --drive-id <drive_id> --type folder --parent-file-id root --user-agent AlibabaCloud-Agent-Skills/alibabacloud-pds-multimodal-search command to list all directories under the root directory (parent-file-id=root) and find the file_id of the Photos directory: a. If the Photos directory exists, note down its file_id b. If the Photos directory does not exist, the file path is invalid 2. Use the aliyun pds list-file --drive-id <drive_id> --type folder --parent-file-id <parent_file_id> --user-agent AlibabaCloud-Agent-Skills/alibabacloud-pds-multimodal-search command to list all directories under the parent directory (parent-file-id=<Photos directory's file_id>) and find the file_id of the 2026 directory: a. If the 2026 directory exists, note down its file_id b. If the 2026 directory does not exist, the file path is invalid 3. Use the aliyun pds list-file --drive-id <drive_id> --type folder --parent-file-id <2026 directory's file_id> --user-agent AlibabaCloud-Agent-Skills/alibabacloud-pds-multimodal-search command to list all directories under the parent directory (parent-file-id=<2026 directory's file_id>) and find the file_id of the 04 directory: a. If the 04 directory exists, note down its file_id b. If the 04 directory does not exist, the file path is invalid 4. Use the aliyun pds list-file --drive-id <drive_id> --type file --parent-file-id <04 directory's file_id> --user-agent AlibabaCloud-Agent-Skills/alibabacloud-pds-multimodal-search command to list all files under the parent directory (parent-file-id=<04 directory's file_id>) and find the file_id of the vacation.jpg file: a. If the vacation.jpg file exists, note down its file_id b. If the vacation.jpg file does not exist, the file path is invalid 5. After obtaining the file_id of vacation.jpg, you can use this file_id to download the file
Note: When executing the aliyun pds list-file command, if there are no valid items returned and the next_marker is not empty, it means that the query is not complete and the next_marker needs to be used as the --marker parameter for the next list query until next_marker is empty.
---
Download File
Step 1: Get Download URL
Get the download link for the file:
aliyun pds get-download-url \
--drive-id <drive_id> \
--file-id <file_id> \
--expire-sec 3600 \
--user-agent AlibabaCloud-Agent-Skills/alibabacloud-pds-multimodal-searchParameter Description:
--drive-id: The drive_id of the space where the file is located (obtained from search results)--file-id: The file_id of the file to download (obtained from search results)--expire-sec: Download link validity period (seconds), default 900, maximum 115200 (32 hours)
Output: Returns a JSON object containing url (download link), expiration, method, size, and other information.
Example Output:
{
"url": "https://pds-data.aliyuncs.com/...",
"expiration": "2024-01-15T11:30:00Z",
"method": "GET",
"size": 1048576
}---
Step 2: Download File
Use the obtained download URL to download the file:
curl -L --max-time 3600 --max-redirs 10 -o <output_filename> '<download_URL>'Parameter Description:
-L: Follow redirects automatically (when download_URL returns a redirect URL, curl will continue downloading from the new location)--max-redirs 10: Maximum number of redirects to follow (prevents infinite redirect loops)--max-time 3600: Maximum time for the entire download operation (seconds)
Note: The -L parameter is critical because PDS download URLs often return a redirect to the actual OSS storage URL. Without this parameter, curl will fail with a 3xx redirect response.
Or use wget:
wget --timeout=3600 --max-redirect=10 -O <output_filename> '<download_URL>'Parameter Description:
--max-redirect=10: Maximum number of redirects to follow--timeout=3600: Timeout for the download operation (seconds)
---
Step 3: Verify Local File Exists
PDS Drive Concepts and API Reference
Scenario: Used when querying user's drive list (including personal space, enterprise space, team space, all spaces) Purpose: Get drive_id for user's personal space, team space, and enterprise space
---
Drive Concept Introduction
A PDS drive is a cloud storage space that can store files. A drive must have an owner, which can be either a user or a group.
- When a drive belongs to a user, it is that user's personal space.
- When a drive belongs to an enterprise group, it is an enterprise space.
- When a drive belongs to a team group, it is a team space.
Users have three types of spaces in a domain:
- Enterprise space
- Team space
- Personal space
When referring to "my PDS drive" without specifying which type of space, it should be understood as all spaces: including enterprise space, team space, and personal space
Mandatory Space Mapping Rules
This section is critical when the user explicitly specifies a space type.
root_group_driverepresents the enterprise space. There is at most one.itemsreturned bylist-my-group-driverepresent team spaces only. There may be multiple.list-my-drives.itemsrepresent personal spaces.- Enterprise space and team space are both group-owned drives, but they are different scopes and must never be mixed.
- If the user says "enterprise space", you must read
drive_idfromroot_group_driveonly. - If the user says "team space", you must read
drive_idfromitemsonly. - If the user says "personal space", you must read
drive_idfromlist-my-drives.itemsonly. - If the user does not specify any space, then and only then can you consider all spaces together.
- If the requested space does not exist in the corresponding field, stop and report that the requested space is unavailable. Do not silently switch to another space type.
- If the user asks for "team space" and multiple team spaces exist, do not arbitrarily choose one unless the request already identifies which team space to use.
Quick Decision Table
| Requested scope | API field to inspect | Allowed behavior |
|---|---|---|
| Enterprise space | root_group_drive | Use it if present; otherwise stop |
| Team space | items | Use the specified team space; ask/clarify if multiple candidates |
| Personal space | list-my-drives.items | Use the user's personal drive |
| Unspecified / all spaces | all of the above | Search one or more spaces as needed |
Drive Query API Reference
Query Method for Enterprise Space and Team Space
You can query using the list group drives API. The items field in the response contains the user's team space list, and the root_group_drive field contains the enterprise space object.
aliyun pds list-my-group-drive --limit 100 --marker "" --user-agent AlibabaCloud-Agent-Skills/alibabacloud-pds-multimodal-searchOutput: Returns JSON containing enterprise space and team space, including items, root_group_drive, next_marker, etc. Detailed explanation:
- items: Contains team space list. There may be multiple team spaces. If not all displayed on one page, next_marker will be returned. If there are no team spaces, this field returns empty.
- root_group_drive: Contains enterprise space object. There is at most one enterprise space. If none exists, this field returns empty.
- next_marker: Used for pagination, indicates the marker for next page. Pass the returned next_marker to the marker parameter to query the next page. If no next page, this field returns empty.
The JSON objects returned in items and root_group_drive are Drive objects. Important attributes of Drive objects include:
- drive_id: Unique space ID, commonly used in API parameters to identify a drive (important parameter for identifying a space, other APIs may require this field as input)
- drive_name: Space name, commonly used for display
- total_size: Total space size in bytes
- used_size: Used space size in bytes
- owner_type: Owner type, either user or group
- owner: Owner ID
Example Output:
{
"items": [
{
"category": "",
"created_at": "2026-03-22T06:00:12.951Z",
"creator": "a34527b***c381b23f3",
"description": "",
"domain_id": "bj12",
"drive_id": "100",
"drive_name": "Test Team Space 1",
"drive_type": "normal",
"encrypt_data_access": false,
"encrypt_mode": "none",
"owner": "e71ce9***c5862d5",
"owner_type": "group",
"permission": null,
"relative_path": "",
"status": "enabled",
"store_id": "fb651***943990a",
"total_size": 107374182400,
"updated_at": "2026-03-22T06:00:12.952Z",
"used_size": 138194
},
{
"category": "",
"created_at": "2026-03-22T06:00:12.951Z",
"creator": "a34527***81b23f3",
"description": "",
"domain_id": "bj12",
"drive_id": "101",
"drive_name": "Test Team Space 2",
"drive_type": "normal",
"encrypt_data_access": false,
"encrypt_mode": "none",
"owner": "e71ce9***b7fc5862d5",
"owner_type": "group",
"permission": null,
"relative_path": "",
"status": "enabled",
"store_id": "fb6516****45c943990a",
"total_size": 107374182400,
"updated_at": "2026-03-22T06:00:12.952Z",
"used_size": 138194
}
],
"next_marker": "",
"root_group_drive": {
"category": "",
"created_at": "2026-03-22T05:55:03.280Z",
"creator": "system",
"description": "",
"domain_id": "bj12",
"drive_id": "103",
"drive_name": "Test Space",
"drive_type": "normal",
"encrypt_data_access": false,
"encrypt_mode": "none",
"owner": "9c251e****b9f952f",
"owner_type": "group",
"permission": null,
"relative_path": "",
"status": "enabled",
"store_id": "fb651****43990a",
"total_size": 107374182400,
"updated_at": "2026-03-23T07:08:40.098Z",
"used_size": 240062520
}
}In the above example output, team space drive_ids are: 100 and 101, enterprise space drive_id is: 103
Important interpretation rule for the example above:
- If the user asked for enterprise space, only
103is eligible. - If the user asked for team space, only
100or101are eligible. - It is incorrect to use
100or101as enterprise space, and incorrect to use103as a team space.
Query API for Personal Space
You can query using the list my drives API. The items field in the response contains the user's personal space list.
aliyun pds list-my-drives --limit 100 --marker "" --user-agent AlibabaCloud-Agent-Skills/alibabacloud-pds-multimodal-searchThe JSON array in the items field returned by the personal space query API contains personal space Drive objects. Important attributes of Drive objects include:
- drive_id: Unique space ID, commonly used in API parameters to identify a drive (important parameter for identifying a space, other APIs may require this field as input)
- drive_name: Space name, commonly used for display
- total_size: Total space size in bytes
- used_size: Used space size in bytes
- owner_type: Owner type, either user or group
- owner: Owner ID
{
"items": [
{
"category": "",
"created_at": "2026-03-22T05:59:33.037Z",
"creator": "a34527b***81b23f3",
"description": "",
"domain_id": "bj31216",
"drive_id": "108",
"drive_name": "SuperAdmin (Test)",
"drive_type": "normal",
"encrypt_data_access": false,
"encrypt_mode": "none",
"owner": "a34527b***81b23f3",
"owner_type": "user",
"permission": null,
"relative_path": "",
"status": "enabled",
"store_id": "fb6516***c943990a",
"total_size": 107374182400,
"updated_at": "2026-03-23T08:45:35.541Z",
"used_size": 950709133
}
],
"next_marker": ""
}In the above example output, personal space drive_id is 108
PDS Document and Audio/Video Analysis
Scenario: When you have obtained the drive_id, file_id, and revision_id of the file to analyze and need to perform analysis on that file Purpose: Perform analysis on files and get structured analysis results ---
Core Workflow
Flow 1: Submit Analysis Task and Poll for Results
Use Python script to automatically submit analysis task and poll until processing is complete.
# Document analysis polling
python scripts/pds_poll_processor.py \
--drive-id "1" \
--file-id "66e7e860a2360204b9414d5c866dd3a20af1974e" \
--revision-id "123" \
--x-pds-process "doc/analysis" \
-o doc_result.json
# Audio/Video analysis polling
python scripts/pds_poll_processor.py \
--drive-id "1" \
--file-id "66e7e860a2360204b9414d5c866dd3a20af1974e" \
--revision-id "123" \
--x-pds-process "video/analysis" \
-o video_result.jsonParameter Description:
--drive-id: The spacedrive_idwhere the analysis file is located--file-id: Thefile_idof the file to analyze--revision-id: Therevision_idof the file to analyze--x-pds-process: Processing type,doc/analysis(document) orvideo/analysis(audio/video). Since analysis is a synchronous API, x-pds-process must be used, not x-pds-async-process-o: Save raw JSON result to file (contains signed URLs)
Document Analysis Result Structure
{
"summary": ["https://bucket/summary.json?sign=xxx"],
"chapter_summaries": ["https://bucket/chapter_summaries.json?sign=xxx"],
"keywords": ["https://bucket/keywords.json?sign=xxx"],
"guiding_questions": ["https://bucket/guiding_questions.json?sign=xxx"],
"method_description": ["https://bucket/method_description.json?sign=xxx"],
"experiment_description": ["https://bucket/experiment_description.json?sign=xxx"],
"conclusion_description": ["https://bucket/conclusion_description.json?sign=xxx"],
"images": {
"imgs/page_0_img_image_box_770_540_1367_860.png": {
"Url": "https://bucket/imgs/page_0_img.png?sign=xxx",
"Thumbnail": "https://bucket/imgs/page_0_img_thumbnail.png?sign=xxx"
}
}
}Audio/Video Analysis Result Structure
{
"markdown": "https://bucket/markdown.md?sign=xxx",
"summary": ["https://bucket/summary.json?sign=xxx"],
"chapter_summaries": ["https://bucket/chapter_summary.json?sign=xxx"],
"keywords": ["https://bucket/keywords.json?sign=xxx"],
"questions": ["https://bucket/questions.json?sign=xxx"],
"transcript": ["https://bucket/transcript.json?sign=xxx"],
"transcript_summaries": ["https://bucket/transcript_summary.json?sign=xxx"],
"transcript_chapter_summaries": ["https://bucket/transcript_chapter_summary.json?sign=xxx"],
"ppt_details": ["https://bucket/ppt_details.json?sign=xxx"],
"images": {
"ppts/video_snapshots_0.jpg": {
"Url": "https://bucket/ppts/video_snapshots_0.jpg?sign=xxx",
"Thumbnail": "https://bucket/ppts/video_snapshots_0_thumbnail.jpg?sign=xxx"
}
}
}Flow 2: Use Formatter to Get Formatted Results
Analysis results contain multiple signed URLs pointing to different types of analysis files. Use formatting scripts to parse these files and generate readable output.
# Format document results
python scripts/doc_analysis_formatter.py doc_result.json -o formatted_output.txt
# Format audio/video results
python scripts/video_analysis_formatter.py video_result.json -o formatted_output.txtParameter Description:
input_file: JSON result file path from analysis API (output from Flow 1)-o: Formatted output file path (optional, outputs to console if not specified)
Formatted Output Example
The formatting script automatically downloads all files pointed to by signed URLs and generates readable output according to preset templates:
````
================================================== 📄 【Full Summary】 ==================================================
{Summary text content}
🖼️ Image: {ImagePath} (Page {PageNumber})
================================================== 🏷️ 【Keywords】 ================================================== #{Keyword 1} | #{Keyword 2} | #{Keyword 3} | ...
================================================== 📚 【Chapter Summaries】 ==================================================
▶️ {Chapter Title} ---------------------------------------- {Chapter Content}
🖼️ Image: {ImagePath}
▶️ {Next Chapter Title} ---------------------------------------- ...
================================================== ❓ 【Guiding Questions】 ==================================================
Q1: {Question 1} A1: {Answer 1}
Q2: {Question 2} A2: {Answer 2} ````
Audio/video will also include dialogue transcripts and PPT extraction information.
---
Flow 3: Extract PPT from Video
If the analyzed video contains PPT, you can extract PPT from the results and generate a PPTX file.
Prerequisites
1. Video contains PPT content 2. Analysis results contain ppt_details field 3. Install Python PPT processing library
pip install python-pptx requestsUsage
Extract PPT from video analysis results and generate PPTX file:
python scripts/ppt_extraction.py video_result.json -o extracted_ppt.pptxParameter Description:
input_file: JSON result file path from video analysis API-o: Output PPTX file path (default: extracted_ppt.pptx)--keep-aspect-ratio: Maintain image aspect ratio (default fills entire slide)--validate: Validate PPTX file after generation
Checklist
- [ ] PPTX file can be opened with PowerPoint/WPS/LibreOffice
- [ ] Slide count matches page count in
ppt_details - [ ] Each page image is clear, no stretching or distortion
- [ ] Page order matches appearance order in video
- [ ] (Optional) Notes contain timestamp information
Auto Validation
python scripts/ppt_extraction.py video_result.json --validateCommon Issues
1. Feature Not Enabled
{
"code": "OperationNotSupport",
"message": "This operation is not supported."
}Solution: Contact PDS technical support to enable analysis feature.
2. Signed URL Expired
Cause: Download took too long, signed URL has expired.
Solution: Re-request analysis results, or download all images immediately after getting results.
RAM Permission Requirements
RAM Policy
If minimum required permissions principle is needed:
metadata:
required_permissions:
- "pds:ListDomains" — List domains: list-domains
- "pds:ListUser" — List users: list-user
- "pds:GetDomain" — Get domain info: get-domain
- "pds:ListFile" — List or search files: list-file
- "pds:GetUser" — Get user info: get-user
- "pds:DownloadFile" — Download file: download-file
- "pds:AssumeUser" — Access via user identity token: user upload (upload-file) / download (get-download-url) / process file (file-process) / get user personal space list (list-my-drive) / get user team/enterprise space list (list-my-group-drive) / user mount (mountapp)API and Permission Reference Table (authentication_type: token, non-RAM authentication)
AssumeUser operation uses user identity access. In token scenario, except for domain management APIs and list user API, all other APIs operate after obtaining user token via AssumeUser, so the Required Permission for these operations is AssumeRole.
| API Action | Required Permission | Resource | |---------------------|---------------------|------------------------------------|| | list-domains | pds:ListDomains | "acs:pds:::domain/", | | get-domain | `pds:GetDomain` | "acs:pds:::domain/<domain_id>" | | list-user | `pds:ListUser` | "acs:pds:::domain/<domain_id>/" | | search-file | pds:AssumeUser | "acs:pds:::domain/<domain_id>/" | | get-user | `pds:AssumeUser` | "acs:pds:::domain/<domain_id>/" | | get-download-url | pds:AssumeUser | "acs:pds:::domain/<domain_id>/" | | process | `pds:AssumeUser` | "acs:pds:::domain/<domain_id>/" | | list-my-group-drive | pds:AssumeUser | "acs:pds:::domain/<domain_id>/" | | list-my-drives | `pds:AssumeUser` | "acs:pds:::domain/<domain_id>/" | | upload-file | pds:AssumeUser | "acs:pds:::domain/<domain_id>/" | | mountapp | `pds:AssumeUser` | "acs:pds:::domain/<domain_id>/" |
API and Permission Reference Table (authentication_type: ak, RAM authentication)
Using ak authentication method without user identity, only the following APIs are supported:
| API Action | Required Permission | Resource | |---------------------|---------------------|------------------------------------|| | list-domains | pds:ListDomains | "acs:pds:::domain/", | | get-domain | `pds:GetDomain` | "acs:pds:::domain/<domain_id>" | | search-file | `pds:ListFile` | "acs:pds:::domain/<domain_id>/" | | get-user | pds:GetUser | "acs:pds:::domain/<domain_id>/" | | list-user | `pds:ListUser` | "acs:pds:::domain/<domain_id>/" | | get-download-url | pds:DownloadFile | "acs:pds:::domain/<domain_id>/*" |
Notes
1. In addition to RAM permissions, PDS also requires assigning corresponding Drive space access permissions to users in the PDS Console 2. When calling with AK/SK method, ensure the RAM user has the above permissions 3. When calling with Bearer Token (OAuth) method, permissions are determined by the user role within PDS
PDS File Search
Scenario: When you already have the drive_id to search in and need to search for files under that drive Purpose: Find the target files and retrieve attributes such as file_id. Supports scalar search based on metadata such as filename, type, size, and time, as well as multimodal semantic search based on content understanding.
Core Workflow
Step 1: Semantic Query Analysis
Run the script python scripts/get_semantic_query_prompt.py, get the prompt from standard output (stdout), then use this prompt as the system prompt and the user's natural language query as user input, spawn a sub-agent to think and output JSON result, and report back to the main agent.
Step 2: Scalar Query Analysis
Run the script python scripts/get_scalar_query_prompt.py, get the prompt from standard output (stdout), then use this prompt as the system prompt and the user's natural language query as user input, spawn a sub-agent to think and output JSON result, and report back to the main agent.
Important: You need to prepend current time information UserQueryDatetime: {current time in ISO format} to the user input, because the scalar query prompt contains time-related examples that need to reference the current time.
Step 3: Build Query String
Pass the JSON outputs from Step 1 and Step 2 to scripts/build_query.py:
python scripts/build_query.py \
--scalar-json '{JSON output from Step 2}' \
--semantic-json '{JSON output from Step 1}'The script will: 1. Recursively parse the Query object from scalar query into a query string 2. Convert semantic query to semantic_text = "..." format 3. Merge the modality from semantic query and the category conditions from scalar query according to the retrieval mode 4. Connect all parts with correct logical operators
Modality merge rules (important)
1. Pure scalar retrieval supports multi-modal filtering, for example images or videos. 2. Pure semantic retrieval supports only a single modality and must converge to exactly one of document, image, video, or audio. 3. Mixed retrieval must converge to the single modality selected by semantic retrieval.
- If the scalar
categoryincludes that semantic modality, use the semantic modality as the final modality. - If the scalar
categoryconflicts with the semantic modality, do not continue the search. Instead, tell the user to adjust the conditions and try again.
Important: If the script execution fails, it is strictly forbidden to construct query and order_by on your own understanding for the next step, as this will very easily produce syntax errors. You should go back to step one and restart the query process from the beginning.
If the output has_query is false, do not execute the search, and kindly inform the user of the message content.
If has_query is true, use the output query and order_by for the next step.
Step 4: Execute Search
Use the query and order_by output from build_query.py to call the aliyun CLI tool:
aliyun pds search-file \
--drive-id "drive_id" \
--query "{query from build_query output}" \
--order-by "{order_by from build_query output}" \
--limit 50 \
--recursive true \
--return-total-count true \
--user-agent AlibabaCloud-Agent-Skills/alibabacloud-pds-multimodal-searchPagination: If the response containsnext_marker, you can pass it via--markerparameter in subsequent requests to get the next page. Add--return-total-countto get the total count of matches.
Step 5: Display Search Results
Parse the JSON output returned by the CLI tool and format the search results for display. The response structure contains an items array and optional next_marker, total_count fields. If there is a next_marker, it means there are more results available for pagination.
Output error messages to stderr on failure.
Best Practices
1. Prefer semantic search: When users describe file content or scenarios, semantic search is more accurate than keyword matching
2. Combine conditions appropriately: Semantic search can be combined with scalar conditions, for example "beach photos from this year" can use both a time range and a semantic description
3. Distinguish pure scalar multi-modal filtering from mixed-query single-modality convergence:
- Pure scalar example:
images or videos larger than 10 MB - Mixed, convergent example:
beach photos taken this year - Mixed, conflicting example:
find sunset photos inside video files
4. Note pagination limits: limit has a maximum value of 100, and large result sets require pagination
5. Time format specification: Time conditions use UTC format YYYY-MM-DDTHH:mm:ss
6. Language consistency in semantic search: The semantic query text must stay in the same language as the user's input. Do not translate it.
PDS File Upload Guide
Scenario: When you have obtained the target drive_id and directory file_id and need to upload files to PDS drive Purpose: Upload local files to PDS drive (supports enterprise space, team space, personal space)
---
File Upload Command
Use the aliyun pds upload-file command to directly upload local files to PDS. This command automatically completes the three steps: create file, upload content, and complete upload.
aliyun pds upload-file \
--drive-id <drive_id> \
--local-path <local_file_path> \
--parent-file-id <parent_file_id> \
--name <cloud_file_name> \
--check-name-mode <auto_rename|ignore|refuse> \
--enable-rapid-upload <true|false> \
--part-size <part_size> \
--user-agent AlibabaCloud-Agent-Skills/alibabacloud-pds-multimodal-search---
Parameter Description
| Parameter | Type | Required | Description |
|---|---|---|---|
--drive-id | string | Yes | Target space ID (obtained from space list) |
--local-path | string | Yes | Full path to local file |
--parent-file-id | string | No | Parent directory ID, default is root |
--name | string | No | Cloud file name, defaults to local file name |
--check-name-mode | string | No | Name conflict handling mode: ignore (overwrite), auto_rename (auto rename), refuse (reject), default is ignore |
--enable-rapid-upload | bool | No | Calculate file SHA-1 for rapid upload attempt, default is false |
--part-size | int | No | Size of each part (bytes), default is 5242880 (5MB) |
---
Common Examples
Basic Upload
Upload to root directory using local file name:
aliyun pds upload-file \
--drive-id "100" \
--local-path "/path/to/file.jpg" \
--user-agent AlibabaCloud-Agent-Skills/alibabacloud-pds-multimodal-searchSpecify Directory and File Name
Upload to specified directory with custom cloud file name:
aliyun pds upload-file \
--drive-id "100" \
--local-path "/path/to/file.jpg" \
--parent-file-id "root" \
--name "my-photo.jpg" \
--check-name-mode "auto_rename" \
--user-agent AlibabaCloud-Agent-Skills/alibabacloud-pds-multimodal-searchEnable Rapid Upload
Calculate file SHA-1 for rapid upload attempt (completes instantly if identical file exists in cloud):
aliyun pds upload-file \
--drive-id "100" \
--local-path "/path/to/file.jpg" \
--enable-rapid-upload \
--user-agent AlibabaCloud-Agent-Skills/alibabacloud-pds-multimodal-searchLarge File Multipart Upload
Custom part size (suitable for large file uploads):
aliyun pds upload-file \
--drive-id "100" \
--local-path "/path/to/large-file.zip" \
--part-size 10485760 \
--user-agent AlibabaCloud-Agent-Skills/alibabacloud-pds-multimodal-searchUpload File to Specified Directory
If you want to upload a file to a specified directory in a PDS drive, you need to convert the cloud directory name to the cloud directory file_id, and use this file_id as the value of the --parent-file-id parameter.
For example, to upload a file to the /Photos/2026/04 directory in a personal space, you need to traverse each level of the path to find the corresponding directory's file_id. If a directory does not exist in the cloud, you need to create it first.
Step 1: Find or Create Photos Directory
List all directories under the root directory to find the Photos directory:
aliyun pds list-file \
--drive-id <drive_id> \
--type folder \
--parent-file-id root \
--user-agent AlibabaCloud-Agent-Skills/alibabacloud-pds-multimodal-search- If Photos directory exists: Note down its
file_id - If Photos directory does not exist: Create it first and get the
file_idfrom the response:
aliyun pds create-file \
--drive-id <drive_id> \
--parent-file-id root \
--name Photos \
--check-name-mode refuse \
--type folder \
--user-agent AlibabaCloud-Agent-Skills/alibabacloud-pds-multimodal-searchStep 2: Find or Create 2026 Directory
List all directories under the Photos directory to find the 2026 directory:
aliyun pds list-file \
--drive-id <drive_id> \
--type folder \
--parent-file-id <Photos_directory_file_id> \
--user-agent AlibabaCloud-Agent-Skills/alibabacloud-pds-multimodal-search- If 2026 directory exists: Note down its
file_id - If 2026 directory does not exist: Create it first:
aliyun pds create-file \
--drive-id <drive_id> \
--parent-file-id <Photos_directory_file_id> \
--name 2026 \
--check-name-mode refuse \
--type folder \
--user-agent AlibabaCloud-Agent-Skills/alibabacloud-pds-multimodal-searchStep 3: Find or Create 04 Directory
List all directories under the 2026 directory to find the 04 directory:
aliyun pds list-file \
--drive-id <drive_id> \
--type folder \
--parent-file-id <2026_directory_file_id> \
--user-agent AlibabaCloud-Agent-Skills/alibabacloud-pds-multimodal-search- If 04 directory exists: Note down its
file_id - If 04 directory does not exist: Create it first:
aliyun pds create-file \
--drive-id <drive_id> \
--parent-file-id <2026_directory_file_id> \
--name 04 \
--check-name-mode refuse \
--type folder \
--user-agent AlibabaCloud-Agent-Skills/alibabacloud-pds-multimodal-searchStep 4: Upload File
After obtaining the file_id of the 04 directory, use it as the --parent-file-id parameter value to upload the file:
aliyun pds upload-file \
--drive-id <drive_id> \
--local-path "/path/to/file.jpg" \
--parent-file-id <04_directory_file_id> \
--user-agent AlibabaCloud-Agent-Skills/alibabacloud-pds-multimodal-searchNote: When executing thealiyun pds list-filecommand, if there are no valid items returned and thenext_markeris not empty, it means that the query is not complete. Use thenext_markeras the--markerparameter for the next list query untilnext_markeris empty.
Upload File to Specified Parent File ID
When uploading a file to a specified parent file ID, first verify whether the parent directory with the specified ID exists using the Get File command:
aliyun pds get-file \
--drive-id "100" \
--file-id "1000" \
--user-agent AlibabaCloud-Agent-Skills/alibabacloud-pds-multimodal-searchVerification Logic:
- If the specified Parent File ID does not exist: The system will prompt that the parent directory does not exist. Ask the user to confirm again.
- If the directory exists: Take the response file's
parent_file_idas the new Parent File ID and continue to query through Get File until theparent_file_idisroot, indicating that the top-level directory has been found.
After finding all levels, concatenate them to get the full path of the file in this PDS drive space after upload.
Note: Before uploading, you must query the full path relative to the root directory. Only after that can you proceed with the subsequent upload operations.
Upload Command:
After completing the path query, use the following command to upload the file:
aliyun pds upload-file \
--drive-id "100" \
--local-path "/path/to/file.jpg" \
--parent-file-id "1000" \
--user-agent AlibabaCloud-Agent-Skills/alibabacloud-pds-multimodal-searchPost-Upload:
After the upload is completed, inform the user that the file upload was successful and display the full path relative to the root directory. For example:
The file has been uploaded to thepersonal space(orteam space) with the full path:/Photos/2026/04/01/file.jpg
---
Output Description
After successful command execution, returns a JSON object with complete file information, main fields include:
file_id: Unique file IDname: Cloud file namesize: File sizecreated_at: Creation timeupdated_at: Update timeparent_file_id: Parent directory ID
---
Notes
1. Same name file handling: Recommend using --check-name-mode auto_rename to avoid overwriting existing files 2. Rapid upload feature: Enable --enable-rapid-upload to complete upload instantly when identical file exists in cloud 3. Multipart upload: Large files are automatically uploaded in parts, adjust part size via --part-size 4. Network stability: Ensure stable network when uploading large files to avoid interruptions
Alibaba Cloud PDS Visual Similar Search Guide
Scenario: When you have prepared a local image file or have obtained the drive_id, file_id, revision_id of an image file, and want to perform image search, similar image search, visual similarity search, or multimodal image retrieval Purpose: Search for similar images in the cloud drive based on user-provided image
Step 1 [Optional]: Upload Local Image File to Drive System Space
Prerequisites
If the user has already provided the image file's drive_id, file_id, revision_id, skip this step
Step 1.1 Get System Space
Execute the following command to get the domain's system space configuration:
aliyun pds get-domain --domain-id <domain-id> --user-agent AlibabaCloud-Agent-Skills/alibabacloud-pds-multimodal-searchResponse example:
{
"domain_id": "bj1093",
"system_drive_config": {
"enable": true,
"drive_id": 1,
"resource_parent_file_id_map": {
"value-add": "68d2348822056f5eea514146b4ad7183cdb94d2f"
}
}
}Extract the system space ID and upload parent file ID from the response. In the above example, system space ID is 1, and upload parent file ID is 68d2348822056f5eea514146b4ad7183cdb94d2f (the value-add item).
If enable in system_drive_config is false, or drive_id is empty, or resource_parent_file_id_map does not contain value-add, there is an issue with system space configuration. Please contact PDS technical support for assistance.
Step 1.2 Upload Local File to Drive System Space
Upload the local file to the drive's system space, where drive_id is set to the system space ID obtained in the previous step, parent_file_id is set to the upload parent file ID obtained in the previous step, and record the file's file_id and revision_id.
Step 2: Construct x-pds-process
If the user searches using a local file, the source file information comes from the file uploaded to the drive in Step 1.2; otherwise, the source file information comes from the drive file information provided by the user.
Must call the existing Python script scripts/render_visual_similar_search_process.py to generate x-pds-process. The script will output x-pds-process to the terminal.
Parameter Description
source_domain_id: Domain where the source image is locatedsource_file_id: File ID of the source imagesource_drive_id: Drive ID of the source imagesource_revision_id: Revision ID of the source imagequery: Search semantic text, not required if nonelimit: Maximum number of similar images to return, not required if none
python scripts/render_visual_similar_search_process.py \
--source_domain_id <SOURCE_DOMAIN_ID> \
--source_file_id <SOURCE_FILE_ID> \
--source_drive_id <SOURCE_DRIVE_ID> \
--source_revision_id <SOURCE_REVISION_ID> \
--query <QUERY> \
--limit <LIMIT>---
Step 3: Perform Image Search
Parameter Description
search_drive_id: Drive ID to search insearch_folder_id: File ID of the folder to search in
Search Entire Drive
aliyun pds process \
--resource-type drive \
--drive-id ${SEARCH_DRIVE_ID} \
--x-pds-process ${X_PDS_PROCESS} \
--user-agent AlibabaCloud-Agent-Skills/alibabacloud-pds-multimodal-searchSearch Specific Folder
aliyun pds process \
--resource-type file \
--drive-id ${SEARCH_DRIVE_ID} \
--file-id ${SEARCH_FOLDER_ID} \
--x-pds-process ${X_PDS_PROCESS} \
--user-agent AlibabaCloud-Agent-Skills/alibabacloud-pds-multimodal-search---
Response
Success Response:
{
"similar_files": [
{
"similarity": 0.95,
"domain_id": "bj1093",
"drive_id": "2",
"file_id": "5d79206586bb5dd69fb34c349282718146c55da7",
"name": "similar_image1.jpg",
"type": "file",
"category": "image",
"size": 102400,
"created_at": "2019-08-20T06:51:27.292Z",
"thumbnail": "https://..."
},
{
"similarity": 0.84,
"domain_id": "bj1093",
"drive_id": "2",
"file_id": "69c0e5c9432208927ca14d1f8af5e897486c6337",
"name": "similar_image2.jpg",
"type": "file",
"category": "image",
"size": 102400,
"created_at": "2023-08-20T06:51:27.292Z",
"thumbnail": "https://..."
}
]
}Field Description:
similarity: Similarity score, range [0, 1], closer to 1 means more similardrive_id: Drive where the result file is located (the search scope drive)file_id: Similar file IDname: File namethumbnail: Thumbnail URL
---
Error Handling
| HTTP Status | Error Code | Description | Solution | |------------|--------|------|---------|| | 400 | InvalidParameter.xxx | Invalid parameter | Check parameter format and encoding | | 400 | OperationNotSupport | Feature not enabled | Contact PDS technical support to enable feature | | 403 | ForbiddenNoPermission.xxx | No permission | Check AccessToken permissions |
Common Errors:
1. Feature Not Enabled
{
"code": "OperationNotSupport",
"message": "This operation is not supported."
}Solution: Contact PDS technical support to enable image search feature.
2. Insufficient Permissions
{
"code": "ForbiddenNoPermission.file",
"message": "No Permission to access resource file"
}Solution:
- Ensure current user has
FILE.LISTpermission on the search space or folder - Ensure current user has
FILE.PREVIEWpermission on the source file
---
Best Practices
1. Set Appropriate limit Parameter
- Quick preview:
l_10orl_20 - Regular search:
l_50 - Comprehensive search:
l_100(maximum)
2. Prefer Image-Only Retrieval
Unless the user explicitly requests image-text hybrid retrieval, prefer using image-only retrieval for better accuracy
---
FAQ
Q: Why are fewer results returned than expected? limit only indicates the maximum number of results, it does not guarantee that limit images will be returned. A: Possible reasons: 1. Actual number of similar images is less than limit 2. Some files were filtered due to insufficient permissions 3. Total number of images in search scope is small
Q: Can I search for videos or documents? A: Not supported, only similar image search is supported.
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
build_query.py 全分支覆盖测试
运行方式: python test_build_query.py
"""
import json
import sys
# 导入被测试的模块
from build_query import (
_escape_value,
_format_value,
_parse_query_recursive,
_modality_to_category,
build_query
)
class TestResults:
"""测试结果统计"""
def __init__(self):
self.passed = 0
self.failed = 0
self.failures = []
def record(self, test_name: str, passed: bool, message: str = ""):
if passed:
self.passed += 1
print(f" ✓ {test_name}")
else:
self.failed += 1
self.failures.append((test_name, message))
print(f" ✗ {test_name}")
if message:
print(f" {message}")
def summary(self):
print("\n" + "=" * 60)
print(f"测试结果: {self.passed} 通过, {self.failed} 失败")
if self.failures:
print("\n失败的测试:")
for name, msg in self.failures:
print(f" - {name}: {msg}")
print("=" * 60)
return self.failed == 0
results = TestResults()
def test_escape_value():
"""测试 _escape_value 函数"""
print("\n[测试 _escape_value]")
# 普通字符串
results.record(
"普通字符串不变",
_escape_value("hello") == "hello",
f"期望 'hello', 得到 '{_escape_value('hello')}'"
)
# 包含双引号
expected_quote = 'say \\"hello\\"'
actual_quote = _escape_value('say "hello"')
results.record(
"转义双引号",
actual_quote == expected_quote,
f"期望 {repr(expected_quote)}, 得到 {repr(actual_quote)}"
)
# 包含反斜杠
expected_slash = "path\\\\to\\\\file"
actual_slash = _escape_value("path\\to\\file")
results.record(
"转义反斜杠",
actual_slash == expected_slash,
f"期望 {repr(expected_slash)}, 得到 {repr(actual_slash)}"
)
# 同时包含双引号和反斜杠
expected = 'a\\\\\\"b'
actual = _escape_value('a\\"b')
results.record(
"转义双引号和反斜杠",
actual == expected,
f"期望 {repr(expected)}, 得到 {repr(actual)}"
)
def test_format_value():
"""测试 _format_value 函数"""
print("\n[测试 _format_value]")
# string 类型字段 - 加引号
results.record(
"string类型字段加引号 (name)",
_format_value("name", "test.pdf") == '"test.pdf"',
f"期望 '\"test.pdf\"', 得到 '{_format_value('name', 'test.pdf')}'"
)
# long 类型字段 - 不加引号
results.record(
"long类型字段不加引号 (size)",
_format_value("size", "1000") == "1000",
f"期望 '1000', 得到 '{_format_value('size', '1000')}'"
)
# boolean 类型字段 - 不加引号
results.record(
"boolean类型字段不加引号 (hidden)",
_format_value("hidden", "false") == "false",
f"期望 'false', 得到 '{_format_value('hidden', 'false')}'"
)
results.record(
"boolean类型字段不加引号 (starred)",
_format_value("starred", "true") == "true",
f"期望 'true', 得到 '{_format_value('starred', 'true')}'"
)
# date 类型字段 - 加引号
results.record(
"date类型字段加引号 (created_at)",
_format_value("created_at", "2025-01-01T00:00:00") == '"2025-01-01T00:00:00"',
f"期望 '\"2025-01-01T00:00:00\"', 得到 '{_format_value('created_at', '2025-01-01T00:00:00')}'"
)
# 未知字段 - 默认加引号
results.record(
"未知字段默认加引号",
_format_value("unknown_field", "value") == '"value"',
f"期望 '\"value\"', 得到 '{_format_value('unknown_field', 'value')}'"
)
# string 类型带转义
expected_escaped = '"file\\"name"'
actual_escaped = _format_value("name", 'file"name')
results.record(
"string类型带转义",
actual_escaped == expected_escaped,
f"期望 {repr(expected_escaped)}, 得到 {repr(actual_escaped)}"
)
def test_basic_operations():
"""测试基础操作符"""
print("\n[测试基础操作符]")
# 简单 eq 查询 - string 类型
query = {"Operation": "eq", "Field": "name", "Value": "test.pdf"}
result, cats = _parse_query_recursive(query)
results.record(
"eq 查询 string 类型带引号",
result == '(name = "test.pdf")',
f"期望 '(name = \"test.pdf\")', 得到 '{result}'"
)
# 简单 eq 查询 - long 类型
query = {"Operation": "eq", "Field": "size", "Value": "1000"}
result, cats = _parse_query_recursive(query)
results.record(
"eq 查询 long 类型不带引号",
result == '(size = 1000)',
f"期望 '(size = 1000)', 得到 '{result}'"
)
# 简单 eq 查询 - boolean 类型
query = {"Operation": "eq", "Field": "hidden", "Value": "false"}
result, cats = _parse_query_recursive(query)
results.record(
"eq 查询 boolean 类型不带引号",
result == '(hidden = false)',
f"期望 '(hidden = false)', 得到 '{result}'"
)
# 简单 gte 查询 - date 类型
query = {"Operation": "gte", "Field": "created_at", "Value": "2025-01-01T00:00:00"}
result, cats = _parse_query_recursive(query)
results.record(
"gte 查询 date 类型带引号",
result == '(created_at >= "2025-01-01T00:00:00")',
f"期望 '(created_at >= \"2025-01-01T00:00:00\")', 得到 '{result}'"
)
# match 操作符
query = {"Operation": "match", "Field": "name", "Value": "报告"}
result, cats = _parse_query_recursive(query)
results.record(
"match 操作符",
result == '(name match "报告")',
f"期望 '(name match \"报告\")', 得到 '{result}'"
)
# prefix 操作符
query = {"Operation": "prefix", "Field": "address", "Value": "Hang"}
result, cats = _parse_query_recursive(query)
results.record(
"prefix 操作符",
result == '(address prefix "Hang")',
f"期望 '(address prefix \"Hang\")', 得到 '{result}'"
)
def test_comparison_operators():
"""测试比较操作符"""
print("\n[测试比较操作符]")
# lt - long 类型不带引号
query = {"Operation": "lt", "Field": "size", "Value": "1000"}
result, cats = _parse_query_recursive(query)
results.record(
"lt 操作符 (long)",
result == '(size < 1000)',
f"期望 '(size < 1000)', 得到 '{result}'"
)
# lte
query = {"Operation": "lte", "Field": "size", "Value": "1000"}
result, cats = _parse_query_recursive(query)
results.record(
"lte 操作符",
result == '(size <= 1000)',
f"期望 '(size <= 1000)', 得到 '{result}'"
)
# gt
query = {"Operation": "gt", "Field": "size", "Value": "1000"}
result, cats = _parse_query_recursive(query)
results.record(
"gt 操作符",
result == '(size > 1000)',
f"期望 '(size > 1000)', 得到 '{result}'"
)
# gte - date 类型带引号
query = {"Operation": "gte", "Field": "created_at", "Value": "2025-01-01T00:00:00"}
result, cats = _parse_query_recursive(query)
results.record(
"gte 操作符 (date)",
result == '(created_at >= "2025-01-01T00:00:00")',
f"期望 '(created_at >= \"2025-01-01T00:00:00\")', 得到 '{result}'"
)
def test_logical_operators():
"""测试逻辑操作符"""
print("\n[测试逻辑操作符]")
# and - 多个子查询
query = {
"Operation": "and",
"SubQueries": [
{"Operation": "eq", "Field": "name", "Value": "test.pdf"},
{"Operation": "gt", "Field": "size", "Value": "1000"}
]
}
result, cats = _parse_query_recursive(query)
results.record(
"and 多个子查询",
result == '((name = "test.pdf") and (size > 1000))',
f"期望 '((name = \"test.pdf\") and (size > 1000))', 得到 '{result}'"
)
# and - 单个子查询(因 category 被移除)
query = {
"Operation": "and",
"SubQueries": [
{"Operation": "eq", "Field": "category", "Value": "image"},
{"Operation": "gt", "Field": "size", "Value": "1000"}
]
}
result, cats = _parse_query_recursive(query)
results.record(
"and 单个子查询(category被移除)",
result == '(size > 1000)',
f"期望 '(size > 1000)', 得到 '{result}'"
)
results.record(
"and 单个子查询 category 被收集",
cats == {"image"},
f"期望 {{'image'}}, 得到 {cats}"
)
# and - 所有子查询都是 category
query = {
"Operation": "and",
"SubQueries": [
{"Operation": "eq", "Field": "category", "Value": "image"},
{"Operation": "eq", "Field": "category", "Value": "video"}
]
}
result, cats = _parse_query_recursive(query)
results.record(
"and 所有子查询都是category返回空",
result == "",
f"期望 '', 得到 '{result}'"
)
results.record(
"and 所有category都被收集",
cats == {"image", "video"},
f"期望 {{'image', 'video'}}, 得到 {cats}"
)
# or - 多个子查询
query = {
"Operation": "or",
"SubQueries": [
{"Operation": "eq", "Field": "name", "Value": "a.pdf"},
{"Operation": "eq", "Field": "name", "Value": "b.pdf"}
]
}
result, cats = _parse_query_recursive(query)
results.record(
"or 多个子查询",
result == '((name = "a.pdf") or (name = "b.pdf"))',
f"期望 '((name = \"a.pdf\") or (name = \"b.pdf\"))', 得到 '{result}'"
)
# or - 单个子查询
query = {
"Operation": "or",
"SubQueries": [
{"Operation": "eq", "Field": "category", "Value": "doc"},
{"Operation": "eq", "Field": "name", "Value": "test.pdf"}
]
}
result, cats = _parse_query_recursive(query)
results.record(
"or 单个子查询(category被移除)",
result == '(name = "test.pdf")',
f"期望 '(name = \"test.pdf\")', 得到 '{result}'"
)
# not 操作符
query = {
"Operation": "not",
"SubQueries": [
{"Operation": "eq", "Field": "hidden", "Value": "true"}
]
}
result, cats = _parse_query_recursive(query)
results.record(
"not 操作符",
result == 'not ((hidden = true))',
f"期望 'not ((hidden = true))', 得到 '{result}'"
)
# not - 子查询为 category
query = {
"Operation": "not",
"SubQueries": [
{"Operation": "eq", "Field": "category", "Value": "image"}
]
}
result, cats = _parse_query_recursive(query)
results.record(
"not 子查询为category返回空",
result == "",
f"期望 '', 得到 '{result}'"
)
# 空 SubQueries
query = {
"Operation": "and",
"SubQueries": []
}
result, cats = _parse_query_recursive(query)
results.record(
"空SubQueries返回空",
result == "",
f"期望 '', 得到 '{result}'"
)
def test_category_extraction():
"""测试 Category 提取"""
print("\n[测试 Category 提取]")
# 单个 category eq
query = {"Operation": "eq", "Field": "category", "Value": "image"}
result, cats = _parse_query_recursive(query)
results.record(
"单个category eq - 返回空查询",
result == "",
f"期望 '', 得到 '{result}'"
)
results.record(
"单个category eq - 收集category",
cats == {"image"},
f"期望 {{'image'}}, 得到 {cats}"
)
# 多个 category 在 or 中
query = {
"Operation": "or",
"SubQueries": [
{"Operation": "eq", "Field": "category", "Value": "image"},
{"Operation": "eq", "Field": "category", "Value": "video"}
]
}
result, cats = _parse_query_recursive(query)
results.record(
"多个category在or中 - 返回空",
result == "",
f"期望 '', 得到 '{result}'"
)
results.record(
"多个category在or中 - 全部收集",
cats == {"image", "video"},
f"期望 {{'image', 'video'}}, 得到 {cats}"
)
# category 混合其他条件
query = {
"Operation": "and",
"SubQueries": [
{"Operation": "eq", "Field": "category", "Value": "doc"},
{"Operation": "gt", "Field": "size", "Value": "1000"}
]
}
result, cats = _parse_query_recursive(query)
results.record(
"category混合其他条件 - 只返回其他条件",
result == "(size > 1000)",
f"期望 '(size > 1000)', 得到 '{result}'"
)
results.record(
"category混合其他条件 - category被提取",
cats == {"doc"},
f"期望 {{'doc'}}, 得到 {cats}"
)
def test_nested_queries():
"""测试嵌套查询"""
print("\n[测试嵌套查询]")
# 两层嵌套: and[or[A, B], C]
query = {
"Operation": "and",
"SubQueries": [
{
"Operation": "or",
"SubQueries": [
{"Operation": "eq", "Field": "name", "Value": "a.pdf"},
{"Operation": "eq", "Field": "name", "Value": "b.pdf"}
]
},
{"Operation": "gt", "Field": "size", "Value": "1000"}
]
}
result, cats = _parse_query_recursive(query)
expected = '(((name = "a.pdf") or (name = "b.pdf")) and (size > 1000))'
results.record(
"两层嵌套 and[or[A,B], C]",
result == expected,
f"期望 '{expected}', 得到 '{result}'"
)
# 三层嵌套
query = {
"Operation": "or",
"SubQueries": [
{
"Operation": "and",
"SubQueries": [
{"Operation": "eq", "Field": "type", "Value": "file"},
{
"Operation": "or",
"SubQueries": [
{"Operation": "eq", "Field": "file_extension", "Value": "pdf"},
{"Operation": "eq", "Field": "file_extension", "Value": "docx"}
]
}
]
},
{"Operation": "eq", "Field": "hidden", "Value": "false"}
]
}
result, cats = _parse_query_recursive(query)
# 验证结果包含正确的结构
results.record(
"三层嵌套查询",
"type" in result and "file_extension" in result and "hidden" in result,
f"得到 '{result}'"
)
def test_semantic_queries():
"""测试语义查询"""
print("\n[测试语义查询]")
# 纯语义查询
semantic_json = json.dumps({
"valid": True,
"result": {"query": "海边日落", "modality": ["image"]}
})
result = build_query(None, semantic_json)
results.record(
"纯语义查询",
result["has_query"] == True and 'semantic_text = "海边日落"' in result["query"],
f"得到 query: {result.get('query')}"
)
results.record(
"纯语义查询含category",
"category" in result["query"],
f"得到 query: {result.get('query')}"
)
# 语义查询中的特殊字符转义
semantic_json = json.dumps({
"valid": True,
"result": {"query": '说"你好"的照片', "modality": ["image"]}
})
result = build_query(None, semantic_json)
results.record(
"语义查询特殊字符转义",
result["has_query"] == True and '\\"' in result["query"],
f"得到 query: {result.get('query')}"
)
def test_category_modality_merge():
"""测试 Category 和 Modality 合并"""
print("\n[测试 Category/Modality 合并]")
# 标量 category + 语义 modality 冲突
scalar_json = json.dumps({
"valid": True,
"result": {"Query": {"Operation": "eq", "Field": "category", "Value": "doc"}}
})
semantic_json = json.dumps({
"valid": True,
"result": {"query": "合同", "modality": ["video"]}
})
result = build_query(scalar_json, semantic_json)
results.record(
"category + modality 冲突时报错",
result["has_query"] == False and "conflicts with the scalar filters" in result["message"],
f"得到结果: has_query={result.get('has_query')}, message={result.get('message')}"
)
# 标量多模态 + 语义单模态 收敛到语义模态
scalar_json = json.dumps({
"valid": True,
"result": {
"Query": {
"Operation": "or",
"SubQueries": [
{"Operation": "eq", "Field": "category", "Value": "image"},
{"Operation": "eq", "Field": "category", "Value": "video"}
]
}
}
})
semantic_json = json.dumps({
"valid": True,
"result": {"query": "海边日落", "modality": ["image"]}
})
result = build_query(scalar_json, semantic_json)
results.record(
"多模态标量与语义收敛到单模态",
result["has_query"] == True and 'category = "image"' in result["query"] and "video" not in result["query"],
f"得到 query: {result.get('query')}"
)
# 仅语义有 modality
semantic_json = json.dumps({
"valid": True,
"result": {"query": "会议", "modality": ["document"]}
})
result = build_query(None, semantic_json)
results.record(
"仅语义有modality",
result["has_query"] == True and "category" in result["query"] and "doc" in result["query"],
f"得到 query: {result.get('query')}"
)
# 仅标量有 category
scalar_json = json.dumps({
"valid": True,
"result": {
"Query": {
"Operation": "and",
"SubQueries": [
{"Operation": "eq", "Field": "category", "Value": "image"},
{"Operation": "gt", "Field": "size", "Value": "1000"}
]
}
}
})
result = build_query(scalar_json, None)
results.record(
"仅标量有category",
result["has_query"] == True and "category" in result["query"] and "image" in result["query"],
f"得到 query: {result.get('query')}"
)
def test_combined_scalar_semantic():
"""测试标量+语义组合查询"""
print("\n[测试标量+语义组合]")
scalar_json = json.dumps({
"valid": True,
"result": {"Query": {"Operation": "gt", "Field": "size", "Value": "1000"}}
})
semantic_json = json.dumps({
"valid": True,
"result": {"query": "风景照片", "modality": ["image"]}
})
result = build_query(scalar_json, semantic_json)
results.record(
"标量+语义组合",
result["has_query"] == True and "size > 1000" in result["query"] and "semantic_text" in result["query"],
f"得到 query: {result.get('query')}"
)
def test_edge_cases():
"""测试边界情况"""
print("\n[测试边界情况]")
# 两个查询都 valid=false
scalar_json = json.dumps({"valid": False})
semantic_json = json.dumps({"valid": False})
result = build_query(scalar_json, semantic_json)
results.record(
"两个查询都invalid",
result["has_query"] == False and result["message"] is not None,
f"得到 has_query: {result.get('has_query')}, message: {result.get('message')[:30]}..."
)
# 只有标量 valid
scalar_json = json.dumps({
"valid": True,
"result": {"Query": {"Operation": "eq", "Field": "name", "Value": "test.pdf"}}
})
semantic_json = json.dumps({"valid": False})
result = build_query(scalar_json, semantic_json)
results.record(
"只有标量valid",
result["has_query"] == True and "name" in result["query"],
f"得到 query: {result.get('query')}"
)
# 只有语义 valid
scalar_json = json.dumps({"valid": False})
semantic_json = json.dumps({
"valid": True,
"result": {"query": "日落", "modality": ["image"]}
})
result = build_query(scalar_json, semantic_json)
results.record(
"只有语义valid",
result["has_query"] == True and "semantic_text" in result["query"],
f"得到 query: {result.get('query')}"
)
# 语义检索不允许多模态
semantic_json = json.dumps({
"valid": True,
"result": {"query": "日落", "modality": ["image", "video"]}
})
result = build_query(None, semantic_json)
results.record(
"语义检索禁止多模态",
result["has_query"] == False and "only a single modality" in result["message"],
f"得到结果: has_query={result.get('has_query')}, message={result.get('message')}"
)
# 未知字段默认加引号
query = {"Operation": "eq", "Field": "unknown_custom_field", "Value": "test"}
result_str, cats = _parse_query_recursive(query)
results.record(
"未知字段默认加引号",
result_str == '(unknown_custom_field = "test")',
f"期望 '(unknown_custom_field = \"test\")', 得到 '{result_str}'"
)
# JSON 解析失败
result = build_query("invalid json", None)
results.record(
"JSON解析失败处理",
result["has_query"] == False,
f"得到 has_query: {result.get('has_query')}"
)
# None 输入
result = build_query(None, None)
results.record(
"None输入处理",
result["has_query"] == False,
f"得到 has_query: {result.get('has_query')}"
)
def test_sort_order():
"""测试 Sort 和 Order 处理"""
print("\n[测试 Sort/Order]")
# 单字段排序
scalar_json = json.dumps({
"valid": True,
"result": {
"Query": {"Operation": "eq", "Field": "type", "Value": "file"},
"Sort": "size",
"Order": "desc"
}
})
result = build_query(scalar_json, None)
results.record(
"单字段排序",
result["order_by"] == "size DESC",
f"期望 'size DESC', 得到 '{result.get('order_by')}'"
)
# 多字段排序
scalar_json = json.dumps({
"valid": True,
"result": {
"Query": {"Operation": "eq", "Field": "type", "Value": "file"},
"Sort": "size,name",
"Order": "desc,asc"
}
})
result = build_query(scalar_json, None)
results.record(
"多字段排序",
result["order_by"] == "size DESC,name ASC",
f"期望 'size DESC,name ASC', 得到 '{result.get('order_by')}'"
)
# Order 数量少于 Sort(默认 ASC)
scalar_json = json.dumps({
"valid": True,
"result": {
"Query": {"Operation": "eq", "Field": "type", "Value": "file"},
"Sort": "size,name",
"Order": "desc"
}
})
result = build_query(scalar_json, None)
results.record(
"Order数量少于Sort默认ASC",
result["order_by"] == "size DESC,name ASC",
f"期望 'size DESC,name ASC', 得到 '{result.get('order_by')}'"
)
# 无效 Order 值默认 ASC
scalar_json = json.dumps({
"valid": True,
"result": {
"Query": {"Operation": "eq", "Field": "type", "Value": "file"},
"Sort": "size",
"Order": "invalid"
}
})
result = build_query(scalar_json, None)
results.record(
"无效Order默认ASC",
result["order_by"] == "size ASC",
f"期望 'size ASC', 得到 '{result.get('order_by')}'"
)
# 只有 Sort 没有 Order
scalar_json = json.dumps({
"valid": True,
"result": {
"Sort": "name"
}
})
result = build_query(scalar_json, None)
results.record(
"只有Sort无Order",
result["order_by"] == "name ASC",
f"期望 'name ASC', 得到 '{result.get('order_by')}'"
)
def test_modality_to_category():
"""测试 modality 到 category 的映射"""
print("\n[测试 modality 映射]")
results.record(
"document -> doc",
_modality_to_category("document") == "doc",
f"得到 '{_modality_to_category('document')}'"
)
results.record(
"image -> image",
_modality_to_category("image") == "image",
f"得到 '{_modality_to_category('image')}'"
)
results.record(
"video -> video",
_modality_to_category("video") == "video",
f"得到 '{_modality_to_category('video')}'"
)
results.record(
"audio -> audio",
_modality_to_category("audio") == "audio",
f"得到 '{_modality_to_category('audio')}'"
)
results.record(
"未知模态 -> None",
_modality_to_category("all") is None,
f"得到 '{_modality_to_category('all')}'"
)
results.record(
"大小写不敏感 IMAGE -> image",
_modality_to_category("IMAGE") == "image",
f"得到 '{_modality_to_category('IMAGE')}'"
)
def test_unknown_operation():
"""测试未知操作符"""
print("\n[测试未知操作符]")
query = {"Operation": "unknown_op", "Field": "name", "Value": "test"}
result, cats = _parse_query_recursive(query)
results.record(
"未知操作符返回空",
result == "",
f"期望 '', 得到 '{result}'"
)
def test_single_part_query():
"""测试单一部分查询(去掉外层括号)"""
print("\n[测试单一部分查询]")
# 只有标量查询
scalar_json = json.dumps({
"valid": True,
"result": {"Query": {"Operation": "eq", "Field": "name", "Value": "test.pdf"}}
})
result = build_query(scalar_json, None)
# 只有一个部分时,应该去掉外层括号
results.record(
"单一标量查询去掉外层括号",
result["query"] == 'name = "test.pdf"',
f"期望 'name = \"test.pdf\"', 得到 '{result.get('query')}'"
)
def main():
"""运行所有测试"""
print("=" * 60)
print("build_query.py 全分支覆盖测试")
print("=" * 60)
# 运行所有测试
test_escape_value()
test_format_value()
test_basic_operations()
test_comparison_operators()
test_logical_operators()
test_category_extraction()
test_nested_queries()
test_semantic_queries()
test_category_modality_merge()
test_combined_scalar_semantic()
test_edge_cases()
test_sort_order()
test_modality_to_category()
test_unknown_operation()
test_single_part_query()
# 输出结果
success = results.summary()
sys.exit(0 if success else 1)
if __name__ == "__main__":
main()
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Build a PDS SearchFile API query string.
Combines scalar-query JSON and semantic-query JSON into the final SearchFile
API `query` string. Supports recursively parsing nested query conditions and
merging `modality` with `category` constraints.
Usage:
python build_query.py --scalar-json '<json>' --semantic-json '<json>'
"""
import argparse
import json
import sys
from typing import Dict, Any, Optional, List, Set, Tuple
from get_scalar_query_prompt import field_schema
def _escape_value(value: str) -> str:
"""Escape backslashes and double quotes in query values."""
return value.replace("\\", "\\\\").replace('"', '\\"')
def _format_value(field: str, value: str) -> str:
"""
Format a value according to the field type.
Args:
field: Field name.
value: Field value.
Returns:
A formatted value string. String/date fields are quoted, while
long/boolean fields are not.
"""
# Unknown fields default to string handling.
field_info = field_schema.get(field.lower(), {})
field_type = field_info.get("type", "string")
# long and boolean values are emitted without quotes.
if field_type in ("long", "boolean"):
return str(value)
# string and date values are quoted and escaped.
escaped = _escape_value(str(value))
return f'"{escaped}"'
def _parse_query_recursive(query: Dict[str, Any]) -> Tuple[str, Set[str]]:
"""
Recursively parse a Query object into a query string.
Args:
query: A Query JSON object.
Returns:
A tuple of (query_string, extracted_category_values).
"""
operation = query.get("Operation", "").lower()
categories_found: Set[str] = set()
# Operator mapping from JSON schema to SearchFile query syntax.
op_map = {
"lt": "<",
"lte": "<=",
"eq": "=",
"gt": ">",
"gte": ">=",
"match": "match",
"prefix": "prefix",
}
# Logical operators work on SubQueries.
if operation in ("and", "or", "not"):
sub_queries = query.get("SubQueries", [])
if not sub_queries:
return "", categories_found
sub_parts = []
for sub in sub_queries:
sub_str, sub_cats = _parse_query_recursive(sub)
categories_found.update(sub_cats)
# Filter out empty parts. This happens when category clauses are
# extracted and removed from the recursive query string.
if sub_str:
sub_parts.append(sub_str)
# Re-check the remaining subqueries after filtering.
if not sub_parts:
return "", categories_found
if operation == "not":
return f"not ({sub_parts[0]})", categories_found
if len(sub_parts) == 1:
return sub_parts[0], categories_found
joined = f" {operation} ".join(sub_parts)
return f"({joined})", categories_found
# Comparison / match operators.
if operation in op_map:
field = query.get("Field", "")
value = query.get("Value", "")
api_op = op_map[operation]
# Extract category constraints and rebuild them later in a dedicated
# merge step instead of leaving them inline.
if field.lower() == "category":
categories_found.add(value)
return "", categories_found
formatted_value = _format_value(field, value)
return f"({field} {api_op} {formatted_value})", categories_found
return "", categories_found
def _modality_to_category(modality: str) -> Optional[str]:
"""
Map a semantic modality to a scalar category.
Args:
modality: A modality value.
Returns:
The corresponding category value, or None if the modality is unsupported.
"""
mapping = {
"document": "doc",
"doc": "doc",
"image": "image",
"video": "video",
"audio": "audio",
}
return mapping.get(modality.lower())
def _build_category_query(categories: Set[str]) -> str:
"""Build a category query fragment from a set of category values."""
if not categories:
return ""
if len(categories) == 1:
cat = next(iter(categories))
return f'category = "{_escape_value(cat)}"'
escaped_cats = [f'"{_escape_value(cat)}"' for cat in sorted(categories)]
return f'category in [{", ".join(escaped_cats)}]'
def build_query(
scalar_json: Optional[str],
semantic_json: Optional[str]
) -> Dict[str, Any]:
"""
Build the final query payload.
Args:
scalar_json: Scalar-query JSON string.
semantic_json: Semantic-query JSON string.
Returns:
A dictionary containing has_query, query, order_by, and message.
"""
scalar_data = None
semantic_data = None
# Parse scalar query JSON.
if scalar_json:
try:
scalar_data = json.loads(scalar_json)
except json.JSONDecodeError as e:
print(f"[WARN] Failed to parse scalar query JSON: {e}", file=sys.stderr)
# Parse semantic query JSON.
if semantic_json:
try:
semantic_data = json.loads(semantic_json)
except json.JSONDecodeError as e:
print(f"[WARN] Failed to parse semantic query JSON: {e}", file=sys.stderr)
# Check whether at least one side is valid.
scalar_valid = scalar_data and scalar_data.get("valid", False)
semantic_valid = semantic_data and semantic_data.get("valid", False)
if not scalar_valid and not semantic_valid:
return {
"has_query": False,
"query": None,
"order_by": None,
"message": (
"Sorry, I can't understand your search intent yet. "
"Supported search types currently include:\n"
"1. File-attribute search, such as filename, type, size, and creation time\n"
"2. Content-based semantic search, such as file topics or scenes\n\n"
"Try describing the file more specifically, for example:\n"
'- "Find last year\'s PDF documents"\n'
'- "Photos of beach sunsets"\n'
'- "Video files larger than 10 MB"'
),
}
query_parts: List[str] = []
scalar_categories: Set[str] = set()
# Process scalar query.
scalar_query_str = ""
if scalar_valid:
result = scalar_data.get("result", {})
query_obj = result.get("Query")
if query_obj:
# Parse recursively while extracting category clauses out of the
# main scalar query string.
scalar_query_str, cats_from_scalar = _parse_query_recursive(query_obj)
scalar_categories.update(cats_from_scalar)
# Process semantic query.
semantic_query_str = ""
semantic_category: Optional[str] = None
if semantic_valid:
result = semantic_data.get("result", {})
query_text = result.get("query", "")
modalities = result.get("modality", [])
if query_text:
escaped_text = _escape_value(query_text)
semantic_query_str = f'semantic_text = "{escaped_text}"'
# Semantic search only supports a single modality.
if not isinstance(modalities, list) or len(modalities) != 1:
return {
"has_query": False,
"query": None,
"order_by": None,
"message": (
"Semantic search currently supports only a single modality. "
"Please specify exactly one of document, image, video, or audio."
),
}
semantic_category = _modality_to_category(str(modalities[0]))
if not semantic_category:
return {
"has_query": False,
"query": None,
"order_by": None,
"message": (
"Semantic search currently supports only the four single "
"modalities: document, image, video, and audio."
),
}
# Build the final category condition:
# 1. Pure scalar retrieval allows multiple categories.
# 2. Pure semantic retrieval must be single-modality.
# 3. Mixed retrieval must converge to the semantic modality.
final_categories: Set[str] = set()
if semantic_category:
if scalar_categories and semantic_category not in scalar_categories:
supported_modalities = ", ".join(sorted(scalar_categories))
return {
"has_query": False,
"query": None,
"order_by": None,
"message": (
"The semantic modality conflicts with the scalar filters. "
f"The semantic modality is {semantic_category}, but the "
f"scalar filter only allows {supported_modalities}. "
"Please adjust the conditions and try again."
),
}
final_categories = {semantic_category}
else:
final_categories = scalar_categories
category_str = _build_category_query(final_categories)
# Assemble the final query string.
if scalar_query_str:
query_parts.append(scalar_query_str)
if semantic_query_str:
query_parts.append(f"({semantic_query_str})")
if category_str:
query_parts.append(f"({category_str})")
# Remove a redundant outer pair of parentheses for single-part queries.
if len(query_parts) == 1:
part = query_parts[0]
if part.startswith("(") and part.endswith(")"):
final_query = part[1:-1]
else:
final_query = part
else:
final_query = " and ".join(query_parts)
# Build order_by from Sort and Order.
order_by = None
if scalar_valid:
result = scalar_data.get("result", {})
sort_field = result.get("Sort")
order_direction = result.get("Order", "")
if sort_field:
sort_fields = [f.strip() for f in sort_field.split(",")]
order_directions = [d.strip().upper() for d in order_direction.split(",")] if order_direction else []
order_parts = []
for i, field in enumerate(sort_fields):
direction = order_directions[i] if i < len(order_directions) else "ASC"
if direction not in ("ASC", "DESC"):
direction = "ASC"
order_parts.append(f"{field} {direction}")
order_by = ",".join(order_parts)
return {
"has_query": True,
"query": final_query if final_query else None,
"order_by": order_by,
"message": None,
}
def main():
parser = argparse.ArgumentParser(
description="Combine scalar-query and semantic-query JSON into a SearchFile API query string.",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
# Scalar query only
python build_query.py --scalar-json '{"valid": true, "result": {"Query": {"Operation": "gte", "Field": "size", "Value": "1000"}}}'
# Semantic query only
python build_query.py --semantic-json '{"valid": true, "result": {"query": "beach sunset", "modality": ["image"]}}'
# Mixed query
python build_query.py \\
--scalar-json '{"valid": true, "result": {"Query": {"Operation": "gt", "Field": "size", "Value": "1000"}, "Sort": "size", "Order": "desc"}}' \\
--semantic-json '{"valid": true, "result": {"query": "landscape photo", "modality": ["image"]}}'
Output:
{
"has_query": true,
"query": "combined query string",
"order_by": "size DESC",
"message": null
}
""",
)
parser.add_argument(
"--scalar-json",
default=None,
help="Scalar-query JSON string containing valid and result fields.",
)
parser.add_argument(
"--semantic-json",
default=None,
help="Semantic-query JSON string containing valid and result fields.",
)
args = parser.parse_args()
# Input validation.
if not args.scalar_json and not args.semantic_json:
print("[INFO] No query parameters were provided.", file=sys.stderr)
# Build and print the final query result.
result = build_query(args.scalar_json, args.semantic_json)
print(json.dumps(result, ensure_ascii=False, indent=2))
if __name__ == "__main__":
main()
#!/usr/bin/env python3
"""
PDS 文档精读结果格式化脚本
功能:
- 下载并解析签名文件
- 格式化输出文档分析结果
- 支持全文总结、章节总结、关键词、问题导读等
注意:如需提交精读任务并轮询,请使用 pds_poll_processor.py
"""
import requests
import json
import argparse
from pathlib import Path
def download_and_parse(signed_url):
"""下载并解析签名文件"""
response = requests.get(signed_url, timeout=30)
response.raise_for_status()
return response.json()
def format_document_analysis(result, output_file=None):
"""
格式化文档分析结果
参数:
result: 精读 API 返回的完整结果 (dict 或 JSON 文件路径)
output_file: 输出文件路径,如果为 None 则打印到控制台
"""
# 1. 加载结果数据
if isinstance(result, str):
with open(result, 'r', encoding='utf-8') as f:
result_data = json.load(f)
else:
result_data = result
output = []
# 1. 全文总结
if "summary" in result_data and result_data["summary"]:
try:
summary_data = download_and_parse(result_data["summary"][0])
output.append("=" * 50)
output.append("📄 【全文总结】")
output.append("=" * 50)
output.append("")
for item in summary_data:
if "Text" in item:
output.append(item["Text"])
output.append("")
if "Image" in item:
img = item["Image"]
page_num = img.get('PageNumber', 0) + 1
output.append(f"🖼️ 图片:{img['ImagePath']} (第{page_num}页)")
output.append("")
except Exception as e:
output.append(f"⚠️ 获取全文总结失败:{e}")
output.append("")
# 2. 关键词
if "keywords" in result_data and result_data["keywords"]:
try:
keywords_data = download_and_parse(result_data["keywords"][0])
output.append("=" * 50)
output.append("🏷️ 【关键词】")
output.append("=" * 50)
keywords_str = " | ".join([f"#{kw}" for kw in keywords_data])
output.append(keywords_str)
output.append("")
except Exception as e:
output.append(f"⚠️ 获取关键词失败:{e}")
output.append("")
# 3. 章节总结
if "chapter_summaries" in result_data and result_data["chapter_summaries"]:
try:
chapters_data = download_and_parse(result_data["chapter_summaries"][0])
output.append("=" * 50)
output.append("📚 【章节总结】")
output.append("=" * 50)
output.append("")
for chapter in chapters_data:
title = chapter.get('Title', '无标题')
output.append(f"▶️ {title}")
output.append("-" * 40)
for item in chapter.get("Summary", []):
# 兼容不同大小写的字段
text = item.get("Text") or item.get("text")
if text:
output.append(f" {text}")
output.append("")
img = item.get("Image") or item.get("image")
if img:
output.append(f" 🖼️ 图片:{img.get('ImagePath', '未知路径')}")
output.append("")
output.append("")
except Exception as e:
output.append(f"⚠️ 获取章节总结失败:{e}")
output.append("")
# 4. 问题导读
if "guiding_questions" in result_data and result_data["guiding_questions"]:
try:
qa_data = download_and_parse(result_data["guiding_questions"][0])
output.append("=" * 50)
output.append("❓ 【问题导读】")
output.append("=" * 50)
output.append("")
for i, qa in enumerate(qa_data, 1):
output.append(f"Q{i}: {qa.get('Question', '无问题')}")
output.append(f"A{i}: {qa.get('Answer', '无答案')}")
output.append("")
except Exception as e:
output.append(f"⚠️ 获取问题导读失败:{e}")
output.append("")
# 5. 论文专有字段 (可选)
for field_name, field_label in [
("method_description", "方法介绍"),
("experiment_description", "实验介绍"),
("conclusion_description", "结论介绍")
]:
if field_name in result_data and result_data[field_name]:
try:
desc_data = download_and_parse(result_data[field_name][0])
output.append("=" * 50)
output.append(f"📝 【{field_label}】")
output.append("=" * 50)
output.append("")
description = desc_data.get("Description", [])
for item in description:
text = item.get("text")
if text:
output.append(text)
output.append("")
img = item.get("image")
if img:
output.append(f"🖼️ 图片:{img.get('ImagePath', '未知路径')}")
output.append("")
except Exception as e:
output.append(f"⚠️ 获取{field_label}失败:{e}")
output.append("")
# 6. 图片列表 (如果有额外图片)
if "images" in result_data and result_data["images"]:
output.append("=" * 50)
output.append("🖼️ 【图片列表】")
output.append("=" * 50)
output.append("")
for img_path, img_info in result_data["images"].items():
output.append(f"📎 {img_path}")
if "url" in img_info:
output.append(f" URL: {img_info['url']}")
if "thumbnail" in img_info:
output.append(f" 缩略图:{img_info['thumbnail']}")
output.append("")
# 输出结果
formatted_output = "\n".join(output)
if output_file:
with open(output_file, 'w', encoding='utf-8') as f:
f.write(formatted_output)
print(f"✅ 格式化结果已保存到:{output_file}")
else:
print(formatted_output)
return formatted_output
def main():
"""主函数"""
parser = argparse.ArgumentParser(
description='PDS 文档精读结果格式化工具',
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
示例用法:
# 格式化已有的 JSON 结果文件
python doc_analysis_formatter.py result.json
python doc_analysis_formatter.py result.json -o formatted_output.txt
"""
)
parser.add_argument(
'input_file',
help='精读 API 返回的 JSON 结果文件路径'
)
parser.add_argument(
'-o', '--output',
help='格式化输出文件路径 (默认输出到控制台)'
)
args = parser.parse_args()
# 检查输入文件是否存在
input_path = Path(args.input_file)
if not input_path.exists():
print(f"❌ 文件不存在:{args.input_file}")
return 1
try:
format_document_analysis(args.input_file, args.output)
return 0
except Exception as e:
print(f"❌ 处理失败:{e}")
import traceback
traceback.print_exc()
return 1
if __name__ == "__main__":
exit(main())
import json
# Defines the JSON schema for metadata-based file search queries.
param_schema = {
"type": "object",
"properties": {
"Query": {
"type": "object",
"$id": "#query",
"description": "Defines metadata search conditions for files. Nested query structures are supported.",
"properties": {
"Operation": {
"type": "string",
"enum": ["not", "or", "prefix", "and", "lt", "match", "gte", "eq", "lte", "gt"],
"description": "Required. Specifies the operation type. Supported operations include:\\n- Logical operators: and, or, not (require SubQueries)\\n- Comparison operators: lt, lte, gt, gte, eq\\n- String operators: prefix, match",
"examples": ["and"]
},
"Field": {
"type": "string",
"description": "The metadata field to query. Required for all operations except logical operators (and, or, not).",
"examples": ["size"]
},
"Value": {
"type": "string",
"description": "The target value to query. All values, including numbers and timestamps, must be provided as strings. Not applicable to logical operators (and, or, not).",
"examples": ["10"]
},
"SubQueries": {
"type": "array",
"description": "Required when the operation is a logical operator (and, or, not). Contains nested query conditions that follow the logic of the parent operator. For example, when Operation is 'and', all sub-queries must be true.",
"items": {
"$ref": "#query"
}
}
}
},
"Sort": {
"type": "string",
"description": "The fields used for sorting, separated by commas. Up to 5 fields are allowed. Field order determines sort priority. For example: 'size,name'.",
"examples": ["size,name"]
},
"Order": {
"type": "string",
"enum": ["asc", "desc"],
"description": "The sort direction for each field in Sort. Supported values:\\n- asc: ascending (default)\\n- desc: descending\\nYou may provide multiple directions separated by commas, for example 'asc,desc'. The number of directions cannot exceed the number of Sort fields. If a field has no explicit direction, it defaults to 'asc'.",
"examples": ["asc,desc"]
}
},
"required": []
}
# Defines supported file metadata fields and their value types.
field_schema = {
"parent_file_id": {
"type": "string",
"description": "The parent folder ID.",
"examples": ["root"]
},
"name": {
"type": "string",
"description": "The file name. Supports fuzzy matching with `match`.",
"examples": ["sampleobject.jpg"]
},
"type": {
"type": "string",
"enum": ["file", "folder"],
"description": "The file type: `file` or `folder`.",
"examples": ["file"]
},
"file_extension": {
"type": "string",
"description": "The file extension without the dot, such as `pdf` or `jpg`.",
"examples": ["pdf"]
},
"mime_type": {
"type": "string",
"description": "The MIME type representing the file format.",
"examples": ["image/jpeg"]
},
"starred": {
"type": "boolean",
"description": "Whether the file is starred.",
"examples": ["true"]
},
"created_at": {
"type": "date",
"description": "The creation time in UTC, formatted as 2006-01-02T00:00:00.",
"examples": ["2025-01-01T00:00:00"]
},
"updated_at": {
"type": "date",
"description": "The last modification time in UTC, formatted as 2006-01-02T00:00:00.",
"examples": ["2025-01-01T00:00:00"]
},
"status": {
"type": "string",
"description": "The file status. Currently `available`.",
"examples": ["available"]
},
"hidden": {
"type": "boolean",
"description": "Whether the file is hidden.",
"examples": ["false"]
},
"size": {
"type": "long",
"description": "The file size in bytes.",
"examples": ["1000"]
},
"image_time": {
"type": "date",
"description": "The capture time of an image or video from EXIF metadata, formatted as 2006-01-02T00:00:00.",
"examples": ["2025-01-01T00:00:00"]
},
"last_access_at": {
"type": "date",
"description": "The most recent access time, formatted as 2006-01-02T00:00:00.",
"examples": ["2025-01-01T00:00:00"]
},
"category": {
"type": "string",
"enum": ["image", "video", "audio", "doc", "app", "others"],
"description": "The file category: image, video, audio, doc, app, or others.",
"examples": ["image"]
},
"label": {
"type": "string",
"description": "The system label name.",
"examples": ["landscape"]
},
"face_group_id": {
"type": "string",
"description": "The face-group ID. Use the face-group listing API to get this ID and query photos in that group.",
"examples": ["group-id-xxx"]
},
"address": {
"type": "string",
"description": "The address. Query only one administrative level at a time, such as country ('China'), province ('Zhejiang Province'), city ('Hangzhou'), district/county ('Xihu District' or 'Tonglu County'), or street/town ('Xixi Street' or 'Sandun Town').",
"examples": ["Hangzhou"]
}
}
# Standard scalar-query JSON schema.
def get_json_schema(param_schema: dict) -> dict:
json_schema = {
"type": "object",
"properties": {
"valid": {
"type": "boolean",
"description": "A boolean flag indicating whether the user's input can be mapped to the defined query schema. It must be false in either of these cases: 1) the input does not contain any recognizable reference to a supported field; 2) the input only contains terms for fields that are not defined in the schema, such as 'color' or 'importance'."
},
"result": param_schema
},
"required": ["valid"]
}
return json_schema
# Describes how to decide whether scalar search should be used and how to extract its parameters.
def schalar_search_prompt() -> str:
output = f"""
# Task
Convert natural-language input into database query parameters:
{json.dumps(param_schema, ensure_ascii=False, indent=None)}
## Supported Query Fields
{json.dumps(field_schema, ensure_ascii=False, indent=None)}
## Field Validation Rules
Before processing any query, you must:
1. Only process queries that clearly refer to fields defined above.
2. If the input violates this rule, for example because it does not refer to any supported field or only refers to unsupported concepts, you must return an output with `"valid": false`.
Examples that must return {json.dumps({"valid": False}, ensure_ascii=False)}:
- "Leshan Giant Buddha" (no field is specified; do not assume this means filename)
- "red ones" (`color` is not a supported field)
- "important files" (`importance` is not a supported field)
Example that should not be marked invalid:
- "image files" (`category eq image`)
## Query Operation Guide
Each operation has a specific meaning and usage.
### Numeric comparison operations
- `eq`: exact equality, such as "equals", "is", "is set to"
- `gt`: greater than, such as "greater than", "over"
- `gte`: greater than or equal to, such as "at least", "no less than"
- `lt`: less than, such as "less than", "below"
- `lte`: less than or equal to, such as "at most", "no more than"
### Text operations
- `match`: search for specific text within a field
- `prefix`: prefix matching for path-like values or string prefixes
### Logical operations
- `and`: all conditions must be true
- `or`: any condition may be true
- `not`: negate a condition
## Sort and Order Guide
### Sort
Supports up to 5 comma-separated sort fields. Field order determines priority. Common usage:
- Single field: `"size"`
- Multiple fields: `"size,name"` meaning sort by size first, then by name when sizes are equal
Common mappings:
- "sort by size" -> `"size"`
- "sort by capture time" -> `"image_time"`
- "sort by name" -> `"name"`
- "sort by size and then by time" -> `"size,image_time"`
Recommended combinations:
- `"size,name"`: useful when sorting by size while keeping a deterministic tie-breaker
- `"image_time,name"`: useful when listing results in time order while keeping a deterministic tie-breaker
- `"name"`: alphabetical order
### Order
Use comma-separated directions:
- `asc`: ascending
- `desc`: descending
If fewer Order values are provided than Sort fields, the remaining fields default to `asc`. For example:
- Sort: `"size,image_time,name"`, Order: `"desc"` -> equivalent to `"desc,asc,asc"`
## Important Rules
1. `match` can only be used for filename search on `name`.
2. `prefix` must not be used for `name`.
3. Follow the principle of minimal inference: only add filters that are explicitly requested by the user. For example, if the user does not mention filename conditions, the query must not contain `name` filters. This rule applies to all fields.
4. The `category` field can be used for multi-modal filtering in pure scalar search, such as images or videos.
5. If this scalar query will later be combined with semantic search, the final modality will be narrowed to the single modality chosen by semantic search. Therefore, when the user already implies a semantic target, keep `category` compatible with that semantic modality whenever possible.
""".strip()
output += "\n\n"
output += """
## Examples
### Example 1
Some natural-language inputs are colloquial and use abbreviations.
`"The file's mime type is docx"`
You should normalize the abbreviation to its canonical form whenever possible:
`{"Query": {"Operation": "eq", "Field": "mime_type", "Value": "application/vnd.openxmlformats-officedocument.wordprocessingml.document"}}`
`"Search for pdf files"`
You should normalize the abbreviation to its canonical form whenever possible:
`{"Query": {"Operation": "eq", "Field": "file_extension", "Value": "pdf"}}`
### Example 2: Time-range queries
Time expressions usually imply a range rather than a single point in time.
UserQueryDatetime: 2025-05-26T11:33:52+08:00
Input: `"Files created on June 6"`
This should be converted into a full-day range in UTC. For the entire day of June 6 in Beijing time, the UTC start should be `2025-06-05T16:00:00` and the end should be `2025-06-06T16:00:00`. The start of day and the end boundary must always use exactly `16:00:00`. Do not use non-zero minutes or seconds at day boundaries:
`{"Query": {"Operation":"and","SubQueries":[{"Operation":"gte","Field":"created_at","Value":"2025-06-05T16:00:00"},{"Operation":"lt","Field":"created_at","Value":"2025-06-06T16:00:00"}]}}`
UserQueryDatetime: 2025-05-26T11:33:52+08:00
Input: `"Accessed in the last three and a half hours"`
This should be converted into a UTC time span:
`{"Query": {"Operation":"and","SubQueries":[{"Operation":"gte","Field":"last_access_at","Value":"2025-05-26T00:03:52"},{"Operation":"lte","Field":"last_access_at","Value":"2025-05-26T03:33:52"}]}}`
Key patterns:
- A calendar date = a full-day range
- "recent" = a range ending at the current time
- "before/after" = a bounded interval
- Finally, always convert from Beijing time to UTC. The timestamp must end at whole seconds, with no milliseconds or timezone suffix.
### Example 3: Language consistency
Always keep the output in the same language as the input query.
Input: `"Find files whose name is 蛋糕"`
Correct: `{"Query": {"Operation": "eq", "Field": "name", "Value": "蛋糕"}}`
Incorrect: `{"Query": {"Operation": "eq", "Field": "name", "Value": "cake"}}`
### Example 4: Choosing the right time field
There are four different time fields:
- `image_time`: when an image or video was captured
- Example: `"Photos taken in the summer of 2023"` -> use `image_time`
- `last_access_at`: when the file was last accessed from the drive
- Example: `"Files accessed yesterday"` -> use `last_access_at`
- `created_at`: when the file was created or uploaded into the drive
- Example: `"Files created yesterday"` -> use `created_at`
- Example: `"Files uploaded yesterday"` -> use `created_at`
- `updated_at`: when the file was last updated
- Example: `"Files updated yesterday"` -> use `updated_at`
For photo or video capture-time queries, always prefer `image_time`.
### Example 5: Basic sorting
Input: `"Find files larger than 1 GB and sort by size in descending order"`
`{"Query": {"Operation": "gt", "Field": "size", "Value": "1073741824"}, "Sort": "size", "Order": "desc"}`
### Example 6: Multi-field sorting
Input: `"Find docx files, sort by modification time descending, then by name ascending"`
`{"Query": {"Operation": "eq", "Field": "mime_type", "Value": "application/vnd.openxmlformats-officedocument.wordprocessingml.document"}, "Sort": "updated_at,name", "Order": "desc,asc"}`
### Example 7: Pure sorting with no query condition
Input: `"Sort all files by name"`
`{"Sort": "name"}`
### Example 8: Multi-field sorting with default order
Input: `"Sort all files by size and creation time"`
`{"Sort": "size,created_at"}`
### Example 9: Query simplification
Input: `"Document files or text files"`
`{"Query": {"Operation": "eq", "Field": "category", "Value": "doc"}}`
### Example 10: Pure scalar multi-modal filtering
Input: `"Images or videos larger than 10 MB"`
`{"Query": {"Operation": "and", "SubQueries": [{"Operation": "gt", "Field": "size", "Value": "10485760"}, {"Operation": "or", "SubQueries": [{"Operation": "eq", "Field": "category", "Value": "image"}, {"Operation": "eq", "Field": "category", "Value": "video"}]}]}}`
"""
output += "\n\n"
json_schema = get_json_schema(param_schema=param_schema)
output += f"""
## JSON Output Format
Your output must strictly follow this JSON schema:
```json
{json.dumps(json_schema, indent=None, ensure_ascii=False)}
```
### Output Requirements
- Your response must be a single valid JSON object.
- Critically important: do not add any explanatory text, comments, or extra content outside the JSON object. Your entire response must contain only that JSON.
""".strip()
return output
if __name__ == "__main__":
print(schalar_search_prompt())
import argparse
import base64
from typing import Optional
def url_safe_base64_encode(text):
"""将文本编码为 URL 安全的 base64 格式"""
if not text:
raise ValueError("输入文本不能为空")
encoded = base64.b64encode(text.encode('utf-8')).decode('utf-8')
url_safe = encoded.replace('+', '-').replace('/', '_').rstrip('=')
return url_safe
def generate_x_pds_process_for_vss(source_domain_id: str, source_drive_id: str, source_file_id: str,
source_revision_id: str, query: Optional[str] = None,
limit: Optional[int] = None) -> str:
"""生成以图搜图的请求参数 x-pds-process"""
if not source_drive_id or not source_file_id or not source_revision_id:
raise ValueError("输入参数不能为空")
pds_uri = f"pds://domains/{source_domain_id}/drives/{source_drive_id}/files/{source_file_id}/revisions/{source_revision_id}"
x_pds_process = f"vision/similar-search,s_{url_safe_base64_encode(pds_uri)}"
if query:
real_query = "semantic_text = \"{query}\""
x_pds_process += f",q_{url_safe_base64_encode(real_query)}"
if limit:
x_pds_process += f",l_{limit}"
x_pds_process += ",/c,v_aW1hZ2U"
return x_pds_process
def main():
parser = argparse.ArgumentParser(description='生成以图搜图的请求参数 x-pds-process')
parser.add_argument('--source_domain_id', required=True, help='要搜索的图片的 domain_id')
parser.add_argument('--source_file_id', required=True, help='要搜索的图片的 file_id')
parser.add_argument('--source_drive_id', required=True, help='要搜索的图片的 drive_id')
parser.add_argument('--source_revision_id', required=True, help='要搜索的图片的 revision_id')
parser.add_argument('--query', required=False, help='要搜索的语义文本')
parser.add_argument('--limit', required=False, default=100, help='返回相似图片的最大数量')
args = parser.parse_args()
x_pds_process = generate_x_pds_process_for_vss(args.source_domain_id, args.source_drive_id, args.source_file_id, args.source_revision_id, args.query, args.limit)
print(x_pds_process)
if __name__ == '__main__':
main()requests==2.32.2
python-pptx==0.6.23
Pillow==12.1.1