Python] Build a Smart Document Assistant LINE Bot with Python + Gemini File Search: Let AI Help You Read Documents

Background

In work and life, we often need to deal with a large number of documents: meeting minutes, technical documents, contracts, research reports, and so on. Every time we want to find specific information, we have to flip through the documents page by page, which is time-consuming and prone to missing key points.

Recently, Google launched the Gemini File Search API, allowing AI to directly analyze uploaded documents and answer questions. I thought, if we could combine it with a LINE Bot, so that everyone could "ask" document questions through the most commonly used communication software, wouldn't that be convenient?

Imagine these scenarios:

• 📄 Meeting Minutes: "What are the main resolutions of this meeting?"
• 📊 Technical Documents: "What are the parameters of this API?"
• 🖼️ Image Content: "What's in this picture?"
• 📑 Research Report: "What is the conclusion of this report?"

So I decided to build this "Smart Document Assistant LINE Bot" to make AI your personal document analyst!

Project Code

https://github.com/kkdai/linebot-file-search-adk

(Through this code, you can quickly deploy to GCP Cloud Run and enjoy the convenience of serverless)

📚 Project Feature Introduction

Core Features

1. 📤 Multi-format File Upload
   • Supports document files: PDF, Word (DOCX), plain text (TXT), etc.
   • Supports image files: JPG, PNG, etc. (using Gemini Image Understanding for image content)
   • Automatically handles Chinese filenames to avoid encoding problems
   • Real-time feedback on upload status
2. 🤖 AI Smart Q&A
   • Based on the Google Gemini 2.5 Flash model
   • Searches for relevant content from uploaded documents and answers questions
   • Supports multiple languages such as Traditional Chinese, English, etc.
   • Understands context and provides accurate answers
3. 👥 Multi-conversation Isolation
   • 1-on-1 Chat: Each person has an independent document library (completely isolated)
   • Group Chat: Group members share a document library (collaborative query)
   • Automatically identifies the conversation type, no manual setup required
   • File Search Store automatically created and managed
4. 🔄 Intelligent Error Handling
   • Automatic retry if file upload fails
   • Guides users to upload if there are no documents
   • Detailed error log recording

💻 Core Feature Implementation

1. Automatic Management of File Search Store

This is the core of the entire system, responsible for managing the document library of each user or group.

Store Naming Strategy

Automatically generate a unique store name based on the conversation type:

def get_store_name(event: MessageEvent) -> str:
    """
    Get the file search store name based on the message source.
    Returns user_id for 1-on-1 chat, group_id for group chat.
    """
    if event.source.type == "user":
        return f"user_{event.source.user_id}"
    elif event.source.type == "group":
        return f"group_{event.source.group_id}"
    elif event.source.type == "room":
        return f"room_{event.source.room_id}"
    else:
        return f"unknown_{event.source.user_id}"

Store Existence Check and Creation

The key design is: The name of the File Search Store is automatically generated by the API (e.g., fileSearchStores/abc123), and we can only set the display_name. Therefore, we need to find it through list() and display_name:

async def ensure_file_search_store_exists(store_name: str) -> tuple[bool, str]:
    """
    Ensure file search store exists, create if not.
    Returns (success, actual_store_name).
    """
    try:
        # List all stores and check if one with our display_name exists
        stores = client.file_search_stores.list()
        for store in stores:
            if hasattr(store, 'display_name') and store.display_name == store_name:
                print(f"File search store '{store_name}' already exists: {store.name}")
                return True, store.name

        # Store doesn't exist, create it
        print(f"Creating file search store with display_name '{store_name}'...")
        store = client.file_search_stores.create(
            config={'display_name': store_name}
        )
        print(f"File search store created: {store.name} (display_name: {store_name})")
        return True, store.name

    except Exception as e:
        print(f"Error ensuring file search store exists: {e}")
        return False, ""

Cache Mechanism Optimization

To avoid listing all stores every time, we added a cache mechanism:

# Cache to store display_name -> actual_name mapping
store_name_cache = {}

# 在上傳時使用 cache
if store_name in store_name_cache:
    actual_store_name = store_name_cache[store_name]
else:
    success, actual_store_name = await ensure_file_search_store_exists(store_name)
    store_name_cache[store_name] = actual_store_name

2. Handling Encoding Issues with Chinese Filenames

Problem Analysis

When the file name contains Chinese characters, directly passing it to the API will encounter ASCII encoding errors:

Error: 'ascii' codec can't encode characters in position 19-21: ordinal not in range(128)

Solution: Filename Sanitization

We adopt the strategy of "using ASCII filenames locally and displaying the original filenames":

async def download_line_content(message_id: str, file_name: str) -> Optional[Path]:
    """
    Download file content from LINE and save to local uploads directory.
    """
    try:
        message_content = await line_bot_api.get_message_content(message_id)

        # Extract file extension from original file name
        _, ext = os.path.splitext(file_name)
        # Use safe file name (ASCII only) to avoid encoding issues
        safe_file_name = f"{message_id}{ext}"
        file_path = UPLOAD_DIR / safe_file_name

        async with aiofiles.open(file_path, 'wb') as f:
            async for chunk in message_content.iter_content():
                await f.write(chunk)

        print(f"Downloaded file: {file_path} (original: {file_name})")
        return file_path
    except Exception as e:
        print(f"Error downloading file: {e}")
        return None

The benefits of doing this:

• ✅ Local file paths are all ASCII (avoiding encoding problems)
• ✅ Users still see the original Chinese filenames
• ✅ Supports filenames in any language

3. File Upload and Status Management

The complete file upload process, including waiting for the API to finish processing:

async def upload_to_file_search_store(file_path: Path, store_name: str, display_name: Optional[str] = None) -> bool:
    """
    Upload a file to Gemini file search store.
    Returns True if successful, False otherwise.
    """
    try:
        # Check cache first
        if store_name in store_name_cache:
            actual_store_name = store_name_cache[store_name]
        else:
            success, actual_store_name = await ensure_file_search_store_exists(store_name)
            if not success:
                return False
            store_name_cache[store_name] = actual_store_name

        # Upload to file search store
        config_dict = {}
        if display_name:
            config_dict['display_name'] = display_name

        operation = client.file_search_stores.upload_to_file_search_store(
            file_search_store_name=actual_store_name,
            file=str(file_path),
            config=config_dict if config_dict else None
        )

        # Wait for operation to complete (with timeout)
        max_wait = 60 # seconds
        elapsed = 0
        while not operation.done and elapsed < max_wait:
            await asyncio.sleep(2)
            operation = client.operations.get(operation)
            elapsed += 2

        if operation.done:
            print(f"File uploaded successfully")
            return True
        else:
            print(f"Upload operation timeout")
            return False

    except Exception as e:
        print(f"Error uploading to file search store: {e}")
        return False

4. Smart Query and File Search Integration

When a user asks a question, the system first checks if there are uploaded documents, and then uses File Search to query:

async def query_file_search(query: str, store_name: str) -> str:
    """
    Query the file search store using generate_content.
    Returns the AI response text.
    """
    try:
        # Get actual store name from cache or by searching
        actual_store_name = None

        if store_name in store_name_cache:
            actual_store_name = store_name_cache[store_name]
        else:
            # Try to find the store by display_name
            stores = client.file_search_stores.list()
            for store in stores:
                if hasattr(store, 'display_name') and store.display_name == store_name:
                    actual_store_name = store.name
                    store_name_cache[store_name] = actual_store_name
                    break

        if not actual_store_name:
            # Store doesn't exist - guide user to upload files
            return "📁 您還沒有上傳任何檔案。\n\n請先傳送文件檔案（PDF、DOCX、TXT 等）或圖片給我，上傳完成後就可以開始提問了！"

        # Create FileSearch tool with actual store name
        tool = types.Tool(
            file_search=types.FileSearch(
                file_search_store_names=[actual_store_name]
            )
        )

        # Generate content with file search
        response = client.models.generate_content(
            model=MODEL_NAME,
            contents=query,
            config=types.GenerateContentConfig(
                tools=[tool],
                temperature=0.7,
            )
        )

        if response.text:
            return response.text
        else:
            return "抱歉，我無法從文件中找到相關資訊。"

    except Exception as e:
        print(f"Error querying file search: {e}")
        return f"查詢時發生錯誤：{str(e)}"

5. LINE Bot Webhook Handling

FastAPI's webhook handling, supporting three message types: text, files, and images:

@app.post("/")
async def handle_callback(request: Request):
    signature = request.headers["X-Line-Signature"]
    body = await request.body()
    body = body.decode()

    try:
        events = parser.parse(body, signature)
    except InvalidSignatureError:
        raise HTTPException(status_code=400, detail="Invalid signature")

    for event in events:
        if not isinstance(event, MessageEvent):
            continue

        if event.message.type == "text":
            # Process text message
            await handle_text_message(event, event.message)
        elif event.message.type == "file":
            # Process file message
            await handle_file_message(event, event.message)
        elif event.message.type == "image":
            # Process image message
            await handle_file_message(event, event.message)

    return "OK"

🔧 Challenges and Solutions Encountered

1. Name Design of the File Search Store API

Problem: Initially, I thought I could directly specify the name of the store, but in reality, create() does not accept the name parameter.

Error Message:

FileSearchStores.create() got an unexpected keyword argument 'name'

Cause Analysis:

• The name of the File Search Store is automatically generated by the API (format: fileSearchStores/xxxxx)
• We can only set display_name as an identifier
• Need to find the corresponding store through list() traversal

Solution:

1. Use display_name to store the name we defined (e.g., user_U123456)
2. Find the corresponding store through list() and get the actual name
3. Create a cache to avoid repeated queries

2. Encoding Issues with Chinese Filenames

Problem: When the file name contains Chinese characters, the API call will fail.

Error Message:

'ascii' codec can't encode characters in position 19-21: ordinal not in range(128)

Problem Analysis:

# 問題代碼：檔案路徑包含中文
file_path = "uploads/123456_會議記錄.pdf" # ❌ 編碼錯誤

Solution:

# 解決方案：使用 ASCII 檔名，保留原始名稱供顯示
_, ext = os.path.splitext("會議記錄.pdf")
safe_file_name = f"{message_id}{ext}" # "123456.pdf" ✅
file_path = UPLOAD_DIR / safe_file_name

# 在 config 中保留原始檔名
config = {'display_name': '會議記錄.pdf'} # 用於 AI 回答時的引用

Benefits:

• File system operations use ASCII paths (no errors)
• AI answers still display the original Chinese filenames (user-friendly)

3. 404 Error When the Store Doesn't Exist

Problem: When uploading a file for the first time, the store does not yet exist, and the upload is attempted.

Error Message:

404 Not Found. {'message': '', 'status': 'Not Found'}

Solution: Check and create the store before uploading:

# 1. 檢查 cache
if store_name in store_name_cache:
    actual_store_name = store_name_cache[store_name]
else:
    # 2. 檢查是否存在，不存在則建立
    success, actual_store_name = await ensure_file_search_store_exists(store_name)
    # 3. 加入 cache
    store_name_cache[store_name] = actual_store_name

# 4. 使用實際的 store name 上傳
operation = client.file_search_stores.upload_to_file_search_store(
    file_search_store_name=actual_store_name,
    file=str(file_path),
    config=config_dict
)

4. Asynchronous File Processing

Problem: Uploading files is a time-consuming operation and needs to wait for processing to complete.

Solution:

1. Use aiofiles for asynchronous file reading and writing
2. Use asyncio.sleep() instead of time.sleep()
3. Implement a polling mechanism to wait for the operation to complete

# 等待上傳完成
max_wait = 60 # seconds
elapsed = 0
while not operation.done and elapsed < max_wait:
    await asyncio.sleep(2) # 異步等待
    operation = client.operations.get(operation)
    elapsed += 2

5. VertexAI Does Not Support the File Search API

Problem: Originally wanted to support VertexAI, but found that the File Search API only supports the Gemini API.

Official Explanation: According to Google AI documentation, the File Search function currently only supports use through the Gemini API.

Solution:

• Remove all VertexAI related code and settings
• Simplify environment variable configuration
• Only GOOGLE_API_KEY is needed

📊 Deployment and Maintenance

Local Development Setup

# 1. 安裝依賴
pip install -r requirements.txt

# 2. 設定環境變數
export ChannelSecret="你的 LINE Channel Secret"
export ChannelAccessToken="你的 LINE Channel Access Token"
export GOOGLE_API_KEY="你的 Google Gemini API Key"

# 3. 啟動服務
uvicorn main:app --reload

Docker Deployment

# 建立映像
docker build -t linebot-file-search .

# 啟動容器
docker run -p 8000:8000 \
  -e ChannelSecret=你的SECRET \
  -e ChannelAccessToken=你的TOKEN \
  -e GOOGLE_API_KEY=你的API_KEY \
  linebot-file-search

Google Cloud Run Deployment

# 1. 建立並推送映像
gcloud builds submit --tag gcr.io/你的專案ID/linebot-file-search

# 2. 部署到 Cloud Run
gcloud run deploy linebot-file-search \
  --image gcr.io/你的專案ID/linebot-file-search \
  --platform managed \
  --region asia-east1 \
  --allow-unauthenticated \
  --set-env-vars ChannelSecret=你的SECRET,ChannelAccessToken=你的TOKEN,GOOGLE_API_KEY=你的API_KEY

# 3. 取得服務網址
gcloud run services describe linebot-file-search \
  --platform managed \
  --region asia-east1 \
  --format 'value(status.url)'

🎯 Summary and Future Improvements

Project Highlights

1. Out-of-the-box Document Assistant: No need to install an APP, use it through LINE
2. Smart Document Analysis: Combines the powerful capabilities of Gemini 2.5 Flash
3. Chinese Friendly: Fully supports Chinese filenames and queries
4. Isolation Mechanism: Each conversation has an independent document library, safe and reliable
5. Automated Management: File Search Store automatically created, users are unaware

Practical Experience Sharing

During the development process, I deeply realized:

1. Differences in API Design

The API design concepts of different cloud services are very different:

• Google Gemini: name is generated by the system, developers set display_name
• Need to adapt: Find resources through list + traversal

This reminds us: Reading the official documentation is more important than guessing the API behavior.

2. Encoding Problems Are Everywhere

Even in 2024, encoding problems still exist:

• The file system may not support Unicode
• The API may have limitations on special characters
• Solution: Separate "storage filenames" and "display filenames"

3. The Importance of Asynchronous Programming

When dealing with external APIs:

• Use async/await to avoid blocking
• Use asyncio.sleep() instead of time.sleep()
• Appropriate timeout settings to avoid infinite waiting

Future Improvement Directions

1. Performance Optimization
   • Implement a more complete cache mechanism
   • Batch processing of multiple file uploads
   • Compress large files
2. Feature Expansion
   • Support file deletion function
   • Support listing uploaded files
   • Support file summary generation
   • Integrate more Gemini features (such as image understanding)
3. User Experience Optimization
   • Rich Menu design
   • More friendly error prompts
   • Upload progress display
   • Query history
4. Security Enhancement
   • File size limits
   • File type validation
   • User quota management
   • Sensitive data filtering

Key Learning

Through this project, I learned:

1. The correct way to use Google Gemini File Search
2. The efficiency of FastAPI in handling LINE Bot webhooks
3. The importance of Python async/await in I/O-intensive applications
4. The handling strategies for encoding problems
5. The design patterns of cloud-native applications

Most importantly: AI is not just a chatbot, but a powerful content analysis tool. The File Search API allows us to easily build a professional-grade document Q&A system.

I hope this experience sharing can help friends who are exploring AI application development!

Related Resources

• Project GitHub Repository
• LINE Bot SDK for Python
• Google Gemini File Search API
• FastAPI Documentation
• Google Cloud Run Documentation

---

If you find this project helpful, please give it a Star ⭐, or share it with friends who need it!