多模態 AI 強化無人機視覺應用：從向量檢索到智慧物件辨識

# 匯入 LlamaIndex 核心元件
# GPTVectorStoreIndex 用於建構向量儲存索引
# Document 類別用於封裝文件資料
from llama_index.core import GPTVectorStoreIndex, Document

# 假設 documents_llm 是預先準備好的文件集合
# 每個文件包含無人機相關的技術說明或操作指南
# 文件內容會被嵌入模型轉換為向量表示
documents_llm = [
    Document(text="無人機透過電腦視覺技術識別地面上的車輛，包括卡車、轎車和機車等。"),
    Document(text="物件偵測模型會為每個識別到的物體繪製邊界框，並標示類別名稱。"),
    Document(text="無人機搭載的攝影機具備高解析度影像擷取能力，支援即時串流傳輸。"),
]

# 從文件集合建立向量儲存索引
# from_documents 方法會自動處理文件嵌入與索引建構
# 預設使用 OpenAI 的嵌入模型進行向量轉換
vector_store_index_llm = GPTVectorStoreIndex.from_documents(documents_llm)

# 將索引轉換為查詢引擎
# 查詢引擎提供了便捷的介面進行語義檢索
# similarity_top_k 參數控制回傳最相似的前 k 個結果
vector_query_engine_llm = vector_store_index_llm.as_query_engine(
    similarity_top_k=3  # 回傳最相似的前 3 個文件
)

print("向量儲存索引建構完成，查詢引擎已就緒")

# 匯入時間模組用於測量查詢效能
import time

# 定義使用者查詢
# 這個查詢詢問無人機如何識別特定類型的車輛
user_input = "無人機如何識別卡車？"

# 記錄查詢開始時間
start_time = time.time()

# 執行向量查詢
# query 方法會進行語義檢索並生成回應
# 回應包含根據檢索結果生成的答案
response = vector_query_engine_llm.query(user_input)

# 記錄查詢結束時間並計算執行時間
end_time = time.time()
elapsed_time = end_time - start_time

# 輸出查詢結果與效能指標
print(f"查詢: {user_input}")
print(f"執行時間: {elapsed_time:.4f} 秒")
print(f"回應: {response}")

# 匯入 Deep Lake 函式庫
# deeplake 提供了便捷的資料集管理介面
import deeplake

# 定義資料集路徑
# hub://activeloop/visdrone-det-train 是 VisDrone 訓練集的 Deep Lake 路徑
# 這個資料集包含從無人機視角拍攝的影像與物件標註
dataset_path = 'hub://activeloop/visdrone-det-train'

# 載入資料集
# load 方法會從 Deep Lake Hub 下載並載入資料集
# 載入後的資料集物件提供了存取各種資料的介面
ds = deeplake.load(dataset_path)

# 顯示資料集摘要資訊
# summary 方法會輸出資料集的結構資訊
# 包括各個 tensor 的型別、形狀和壓縮方式
print("資料集摘要：")
ds.summary()

# 取得資料集大小
print(f"\n資料集包含 {len(ds)} 個樣本")

# 視覺化資料集
# visualize 方法會在 Jupyter Notebook 中顯示互動式視覺化介面
# 可以瀏覽影像、檢視邊界框和標籤
ds.visualize()

# 取得標籤類別列表
# labels tensor 的 info 屬性包含類別名稱等元資料
labels_list = ds.labels.info['class_names']
print("資料集類別：")
print(labels_list)

# 預期輸出：
# ['ignored regions', 'pedestrian', 'people', 'bicycle', 'car',
#  'van', 'truck', 'tricycle', 'awning-tricycle', 'bus', 'motor', 'others']

# 匯入 pandas 用於資料分析
import pandas as pd

# 建立空的 DataFrame 用於儲存資料集資訊
# 包含影像資料、邊界框坐標和標籤文字
df = pd.DataFrame(columns=['image', 'boxes', 'labels', 'doc_id'])

# 限制處理的樣本數量以加速示範
# 實際應用中可以處理完整資料集
num_samples = min(100, len(ds))

# 迭代資料集中的每個樣本
# 提取影像、邊界框和標籤資訊
for i in range(num_samples):
    sample = ds[i]

    # 將影像資料轉換為位元組格式儲存
    df.loc[i, 'image'] = sample.images.tobytes()

    # 提取邊界框坐標
    # boxes tensor 包含 [x, y, width, height] 格式的坐標
    boxes_list = sample.boxes.numpy(aslist=True)
    df.loc[i, 'boxes'] = [box.tolist() for box in boxes_list]

    # 提取標籤資訊
    # labels tensor 包含每個邊界框對應的類別索引
    label_data = sample.labels.data()
    df.loc[i, 'labels'] = label_data.get('text', [])

    # 為每個樣本分配唯一識別碼
    df.loc[i, 'doc_id'] = str(i)

print(f"已處理 {len(df)} 個樣本")
print(df.head())

# 匯入影像處理相關函式庫
import io
from PIL import Image, ImageDraw, ImageFont

def display_image_with_bboxes(image_data, bboxes, labels, target_label):
    """
    在影像上繪製特定類別的邊界框並顯示

    這個函式接受影像位元組資料、邊界框坐標列表和標籤列表，
    然後篩選出符合目標標籤的邊界框並繪製在影像上。
    邊界框使用紅色矩形表示，並在左上角標示類別名稱。

    參數:
        image_data: 影像的位元組資料
        bboxes: 邊界框坐標列表，每個邊界框為 [x, y, width, height] 格式
        labels: 標籤列表，與邊界框一一對應
        target_label: 要顯示的目標標籤名稱

    回傳:
        PIL Image 物件，包含繪製邊界框後的影像
    """
    # 將位元組資料轉換為 PIL Image 物件
    # io.BytesIO 將位元組資料包裝為檔案物件
    image_bytes = io.BytesIO(image_data)
    img = Image.open(image_bytes)

    # 建立繪圖物件
    # ImageDraw 提供了在影像上繪製圖形和文字的功能
    draw = ImageDraw.Draw(img)

    # 計數符合目標標籤的邊界框數量
    match_count = 0

    # 迭代每個邊界框與對應標籤
    for idx, box in enumerate(bboxes):
        # 檢查當前標籤是否包含目標標籤
        # labels[idx] 可能是字串或列表格式
        current_label = labels[idx] if idx < len(labels) else ""

        # 判斷是否符合目標標籤
        if target_label.lower() in str(current_label).lower():
            # 解析邊界框坐標
            # 格式為 [x, y, width, height]
            x, y, w, h = box
            x1, y1 = x, y
            x2, y2 = x + w, y + h

            # 繪製紅色矩形邊界框
            # outline 指定框線顏色，width 指定框線寬度
            draw.rectangle([x1, y1, x2, y2], outline="red", width=2)

            # 在邊界框左上角標示類別名稱
            draw.text((x1, y1 - 15), target_label, fill="red")

            match_count += 1

    print(f"找到 {match_count} 個 '{target_label}' 物件")

    return img

# 使用範例
# 從資料集中取得第一個樣本
sample_index = 0
image_data = df.loc[sample_index, 'image']
bboxes = df.loc[sample_index, 'boxes']
labels = df.loc[sample_index, 'labels']

# 指定要顯示的目標標籤
target_label = "truck"

# 繪製邊界框並顯示影像
result_image = display_image_with_bboxes(image_data, bboxes, labels, target_label)

# 儲存結果影像
save_path = "boxed_image.jpg"
result_image.save(save_path)
print(f"結果影像已儲存至 {save_path}")

# 在 Jupyter Notebook 中顯示影像
# display(result_image)

# 匯入必要的函式庫
from llama_index.core import GPTVectorStoreIndex, Document

# 建立文件列表用於向量索引
# 每個文件包含樣本的標籤資訊
documents = []

# 迭代 DataFrame 中的每一行
# 將標籤文字轉換為 Document 物件
for idx, row in df.iterrows():
    # 取得標籤列表並轉換為文字
    text_labels = row['labels']

    # 處理標籤格式
    # 標籤可能是列表或其他格式
    if isinstance(text_labels, list):
        text = " ".join(str(label) for label in text_labels)
    else:
        text = str(text_labels)

    # 建立 Document 物件
    # doc_id 用於後續追蹤文件來源
    document = Document(
        text=text,
        doc_id=row['doc_id'],
        metadata={'sample_index': idx}
    )
    documents.append(document)

print(f"已建立 {len(documents)} 個文件")

# 建立向量儲存索引
# 索引會根據標籤文字的語義建立
vector_store_index = GPTVectorStoreIndex.from_documents(documents)

# 設定查詢引擎
# similarity_top_k=1 表示只回傳最相關的一個結果
vector_query_engine = vector_store_index.as_query_engine(similarity_top_k=1)

print("多模態查詢引擎建構完成")

import time

def multimodal_query(query_text, target_label=None):
    """
    執行多模態查詢並回傳結果

    這個函式整合了文字檢索與影像處理，
    根據使用者查詢檢索相關影像並繪製邊界框。

    參數:
        query_text: 使用者的自然語言查詢
        target_label: 要標示的目標標籤，若為 None 則自動從查詢推斷

    回傳:
        dict: 包含查詢結果的字典
    """
    # 記錄執行時間
    start_time = time.time()

    # 執行向量查詢
    response = vector_query_engine.query(query_text)

    # 解析回應取得來源文件資訊
    # source_nodes 包含檢索到的文件節點
    if hasattr(response, 'source_nodes') and response.source_nodes:
        source_node = response.source_nodes[0]
        doc_id = source_node.node.doc_id

        # 根據 doc_id 找到對應的 DataFrame 行
        sample_index = int(doc_id)

        # 取得影像與標註資料
        image_data = df.loc[sample_index, 'image']
        bboxes = df.loc[sample_index, 'boxes']
        labels = df.loc[sample_index, 'labels']

        # 如果未指定目標標籤，嘗試從查詢推斷
        if target_label is None:
            # 簡單的關鍵字比對
            for label in labels_list:
                if label.lower() in query_text.lower():
                    target_label = label
                    break
            if target_label is None:
                target_label = "car"  # 預設標籤

        # 繪製邊界框
        result_image = display_image_with_bboxes(
            image_data, bboxes, labels, target_label
        )

        # 計算執行時間
        elapsed_time = time.time() - start_time

        return {
            'query': query_text,
            'response': str(response),
            'target_label': target_label,
            'image': result_image,
            'elapsed_time': elapsed_time,
            'sample_index': sample_index
        }
    else:
        return {
            'query': query_text,
            'response': "未找到相關影像",
            'elapsed_time': time.time() - start_time
        }

# 測試多模態查詢
result = multimodal_query("請找出包含卡車的影像", target_label="truck")

print(f"查詢: {result['query']}")
print(f"目標標籤: {result['target_label']}")
print(f"執行時間: {result['elapsed_time']:.4f} 秒")
print(f"文字回應: {result['response']}")

# 儲存結果影像
if 'image' in result:
    result['image'].save("multimodal_result.jpg")
    print("結果影像已儲存至 multimodal_result.jpg")

# 匯入相似度計算相關函式庫
from sklearn.metrics.pairwise import cosine_similarity
from sentence_transformers import SentenceTransformer

# 初始化 SentenceTransformer 模型
# all-MiniLM-L6-v2 是輕量級但效能良好的嵌入模型
# 適合用於計算文字語義相似度
model = SentenceTransformer('all-MiniLM-L6-v2')

def calculate_cosine_similarity(text1, text2):
    """
    計算兩個文字之間的餘弦相似度

    使用 SentenceTransformer 將文字轉換為向量表示，
    然後計算兩個向量之間的餘弦相似度。

    參數:
        text1: 第一個文字字串
        text2: 第二個文字字串

    回傳:
        float: 餘弦相似度分數，範圍為 -1 到 1
    """
    # 將兩個文字轉換為向量表示
    # encode 方法使用預訓練模型生成嵌入向量
    embeddings1 = model.encode(text1)
    embeddings2 = model.encode(text2)

    # 計算餘弦相似度
    # cosine_similarity 期望輸入為二維陣列
    similarity = cosine_similarity([embeddings1], [embeddings2])

    return similarity[0][0]

# 測試相似度計算
user_query = "無人機如何識別卡車？"
system_response = "無人機透過電腦視覺技術和物件偵測模型識別地面上的卡車。"

similarity_score = calculate_cosine_similarity(user_query, system_response)
print(f"使用者查詢: {user_query}")
print(f"系統回應: {system_response}")
print(f"餘弦相似度: {similarity_score:.4f}")

# 匯入 OpenAI 函式庫
import openai
import base64

def analyze_image_with_gpt4v(image_path, prompt):
    """
    使用 GPT-4 Vision 分析影像

    將影像傳送至 GPT-4 Vision API 進行分析，
    並根據提示詞生成影像描述。

    參數:
        image_path: 影像檔案路徑
        prompt: 分析提示詞

    回傳:
        str: GPT-4 Vision 生成的影像描述
    """
    # 讀取影像並轉換為 base64 編碼
    with open(image_path, "rb") as image_file:
        image_base64 = base64.b64encode(image_file.read()).decode('utf-8')

    # 建構 API 請求
    # GPT-4 Vision 支援影像輸入
    response = openai.ChatCompletion.create(
        model="gpt-4-vision-preview",
        messages=[
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": prompt},
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:image/jpeg;base64,{image_base64}"
                        }
                    }
                ]
            }
        ],
        max_tokens=500
    )

    return response.choices[0].message.content

def evaluate_image_retrieval(image_path, expected_label):
    """
    評估影像檢索結果的準確性

    使用 GPT-4 Vision 分析影像內容，
    並計算分析結果與預期標籤的相似度。

    參數:
        image_path: 影像檔案路徑
        expected_label: 預期應該在影像中出現的標籤

    回傳:
        dict: 包含分析結果和相似度分數的字典
    """
    # 定義分析提示詞
    prompt = f"請描述這張影像中的主要物體，特別關注是否有 {expected_label}。"

    # 使用 GPT-4 Vision 分析影像
    image_description = analyze_image_with_gpt4v(image_path, prompt)

    # 計算描述與預期標籤的相似度
    similarity = calculate_cosine_similarity(image_description, expected_label)

    return {
        'description': image_description,
        'expected_label': expected_label,
        'similarity': similarity
    }

# 測試影像評估（需要有效的 OpenAI API 金鑰）
# result = evaluate_image_retrieval("multimodal_result.jpg", "truck")
# print(f"影像描述: {result['description']}")
# print(f"預期標籤: {result['expected_label']}")
# print(f"相似度: {result['similarity']:.4f}")

def adaptive_response(user_query, ranking):
    """
    根據排名自適應選擇回應策略

    這個函式實現了混合自適應 RAG 的核心邏輯，
    根據使用者提問的排名選擇不同的回應策略。

    參數:
        user_query: 使用者的提問
        ranking: 提問的複雜度排名（1-10）

    回傳:
        dict: 包含回應策略和結果的字典
    """
    # 判斷回應策略
    if ranking < 3:
        # 低排名：使用標準回應
        strategy = "standard"
        response = generate_standard_response(user_query)
        source = "預設知識庫"

    elif 3 <= ranking < 5:
        # 中等排名：引入專家回饋
        strategy = "human_feedback"
        expert_content = load_expert_feedback()
        response = generate_response_with_context(user_query, expert_content)
        source = "專家回饋"

    else:
        # 高排名：完整 RAG 流程
        strategy = "rag"
        retrieved_docs = retrieve_documents(user_query)
        response = generate_response_with_context(user_query, retrieved_docs)
        source = "RAG 檢索"

    return {
        'query': user_query,
        'ranking': ranking,
        'strategy': strategy,
        'response': response,
        'source': source
    }

def generate_standard_response(query):
    """
    生成標準回應

    對於簡單問題，直接使用預設知識生成回應。

    參數:
        query: 使用者查詢

    回傳:
        str: 生成的回應
    """
    # 簡化實作，實際應用中會使用 LLM 生成
    return f"關於「{query}」的標準回應：請參考使用手冊或聯繫客服。"

def load_expert_feedback():
    """
    載入專家回饋內容

    從專家知識庫中載入相關的回饋內容。

    回傳:
        str: 專家回饋內容
    """
    # 簡化實作，實際應用中會從資料庫載入
    return """
    無人機視覺系統採用深度學習物件偵測模型，
    能夠識別多種地面目標包括車輛、行人和建築物。
    系統會為每個識別到的物體標示邊界框和類別名稱。
    """

def retrieve_documents(query):
    """
    檢索相關文件

    使用向量檢索取得與查詢相關的文件。

    參數:
        query: 使用者查詢

    回傳:
        str: 檢索到的文件內容
    """
    # 使用先前建構的查詢引擎
    response = vector_query_engine_llm.query(query)
    return str(response)

def generate_response_with_context(query, context):
    """
    根據上下文生成回應

    將檢索到的上下文與查詢結合，生成完整的回應。

    參數:
        query: 使用者查詢
        context: 上下文內容

    回傳:
        str: 生成的回應
    """
    # 簡化實作，實際應用中會使用 LLM 生成
    return f"根據相關資料回答「{query}」：\n{context[:500]}..."

# 測試自適應回應
test_cases = [
    ("無人機的電池續航時間是多少？", 2),
    ("如何設定無人機的物件偵測參數？", 4),
    ("無人機如何在複雜環境中進行自主避障？", 7)
]

for query, ranking in test_cases:
    result = adaptive_response(query, ranking)
    print(f"\n查詢: {result['query']}")
    print(f"排名: {result['ranking']}")
    print(f"策略: {result['strategy']}")
    print(f"來源: {result['source']}")
    print(f"回應: {result['response'][:100]}...")