Google 主催『Gen AI Intensive Course Capstone 2025Q1』に参加しました！

2025年3月31日（月）から4月4日（金）まで開催されたGoogle主催の5日間集中型コース「Gen AI Intensive Course with Google」に参加し、その集大成となる『Capstone Project』を制作しました。

このCapstoneプロジェクトでは、実際に学んだ生成AI技術を応用し、自分自身で選んだ課題やアイデアを実現します。
特に重視されているのは、以下のような生成AIの能力を最低でも3つ組み合わせて実用的な問題を解決することです：

• 構造化出力（JSONモード）
• Few-shot プロンプト
• 画像認識やドキュメント理解
• エージェントや関数呼び出し(Function Calling) など

優秀な作品として評価されると、KaggleやGoogle公式SNSで紹介される特典があります。

私のCapstoneプロジェクト：「生成AIによるカップルの関係性分析プログラム」

私が取り組んだのは、Google Gemini AI を活用して画像認識と会話履歴からカップル・夫婦間の関係性や感情の状態を評価するプログラムです。

具体的には、

• カップルや夫婦の写真を分析し、「身体の距離感」「視線の接触」「表情の感情表現」などを数値化して評価。
• 会話履歴を分析して「会話の主導性」「共感や価値観の共有度」「関係の温かさ」「ポジティブまたはネガティブな感情比率」などの指標を抽出。
• これらを総合的にAIが判断し、関係性をスコア化して、より詳細な分析や改善提案をレポート形式で出力します。

本プロジェクトは以下のような用途を想定しています：

• パートナー間のコミュニケーション改善ツールとしての活用
• プロのカウンセラーやセラピストの支援ツールとしての提供
• 自己認識や自己改善を促す個人向けのアプリケーション化

将来的には、このプロトタイプをベースにして、実際に誰でも使えるアプリとして開発していく予定です。このブログでは、開発の進捗や工夫した点、AI活用のコツなどを継続的に記録していきますので、ぜひお楽しみに！

Gen AIを活用した関係性分析パイプラインの構築

このCapstoneプロジェクトでは、最新のGenerative AI技術を用いて、画像と会話テキストから人間関係の感情的な状態を分析するインタラクティブなパイプラインを構築しました。以下、各セクションごとの詳細な説明とコードを紹介します。

1. 環境構築とパッケージのインストール

本プロジェクトでは、Kaggleのベース環境との競合を防ぐため不要なパッケージを削除し、必要な最新パッケージをインストールしました。主にLangGraphとGoogle Generative AI関連のパッケージを使用しています。

# 不要なパッケージのアンインストール
!pip uninstall -qqy kfp jupyterlab libpysal thinc spacy fastai ydata-profiling google-cloud-bigquery google-generativeai

# 必要なパッケージのインストール
!pip install -qU 'langgraph==0.3.21' 'langchain-google-genai==2.1.2' 'langgraph-prebuilt==0.1.7'

2. APIキー設定とライブラリのインポート

Google APIキー（Geminiモデル用）は安全にKaggleのシークレット機能を用いて設定します。また、分析に必要な各種ライブラリをインポートしています。

import os
from kaggle_secrets import UserSecretsClient

GOOGLE_API_KEY = UserSecretsClient().get_secret("Gemini API")
os.environ["GOOGLE_API_KEY"] = GOOGLE_API_KEY

%matplotlib inline
import json
import matplotlib.pyplot as plt
import numpy as np
from google import genai
from pydantic import BaseModel, Field
from langgraph.graph import StateGraph

3. データスキーマの定義

Pydanticを用いてモデルからの出力を構造化し、データの品質と整合性を担保します。

• 総合感情評価
• 画像分析結果
• 会話分析結果
• 複合分析レポート

class SummaryRating(BaseModel):
    comprehensive_emotional_index: int = Field(..., ge=1, le=100, description="Overall emotional index, between 1-100.")
    confidence_score: int = Field(..., ge=1, le=100, description="Confidence score, between 1-100.")
    rating_reason: str = Field(..., description="A one-sentence summary of the rating rationale.")
    supplement_suggestion: str = Field(..., description="Suggestions for supplementary user input.")

class ImageAnalysisResponse(BaseModel):
    description: str = Field(..., description="A textual description of the image's content.")
    proximity_score: float = Field(..., ge=0, le=1, description="Score indicating closeness between subjects (0 to 1).")
    eye_contact_score: float = Field(..., ge=0, le=1, description="Score reflecting eye contact (0 to 1).")
    facial_expression_score: float = Field(..., ge=0, le=1, description="Score of facial expressions (0 to 1).")
    body_touch_score: float = Field(..., ge=0, le=1, description="Score representing physical contact (0 to 1).")
    reason: str = Field(..., description="Analysis reasoning based on image observations.")

class ConversationAnalysisResponse(BaseModel):
    positive_ratio: float = Field(..., ge=0, le=1, description="Proportion of positive sentiment.")
    negative_ratio: float = Field(..., ge=0, le=1, description="Proportion of negative sentiment.")
    initiative_score: float = Field(..., ge=0, le=1, description="Score for conversational initiative (0 to 1).")
    value_alignment_score: float = Field(..., ge=0, le=1, description="Score for shared values (0 to 1).")
    relationship_warmth_score: float = Field(..., ge=0, le=1, description="Score for the warmth of the relationship (0 to 1).")
    toxicity_probability: float = Field(..., ge=0, le=1, description="Probability score for potential toxicity.")
    reason: str = Field(..., description="Rationale derived from the conversation and image analysis.")

class CompositeReport(BaseModel):
    composite_reason: str = Field(..., description="Overall composite rating rationale provided by a mental health and relationship expert perspective.")
    detailed_report: str = Field(..., description="Comprehensive analysis report with insights and recommendations.")
    model_config = ConfigDict(ref_template=None)

4. Gemini APIクライアントの初期化

Geminiの「gemini-2.0-flash」モデルを利用してコンテンツを生成します。

client = genai.Client(api_key=GOOGLE_API_KEY)
llm = client.chats.create(model='gemini-2.0-flash')

5. JSONレスポンス抽出用ユーティリティ関数

モデルからのレスポンス内に含まれるJSONデータを安全に取り出し、エラー時の対応も実装しています。

def extract_json_from_response(response_text) -> dict:
    #print("response_text:", response_text)
    
    # Convert non-string responses to string
    if not isinstance(response_text, str):
        response_text = str(response_text)
    
    # Extract JSON content from markdown code fences if present
    m = re.search(r"```json\s*(\{.*\})\s*```", response_text, re.DOTALL)
    if m:
        json_str = m.group(1)
    else:
        # Fallback: extract from first '{' to last '}'
        start = response_text.find('{')
        end = response_text.rfind('}')
        if start != -1 and end != -1 and end > start:
            json_str = response_text[start:end+1]
        else:
            print("Unable to locate valid JSON content in the response.")
            return {
                "comprehensive_emotional_index": 50,
                "confidence_score": 50,
                "rating_reason": "Unable to retrieve a valid rating.",
                "supplement_suggestion": "Please provide more conversational data."
            }
    try:
        data = json.loads(json_str)
        return data
    except Exception as e:
        print("JSON parsing error:", e)
        return {
            "comprehensive_emotional_index": 50,
            "confidence_score": 50,
            "rating_reason": "Unable to retrieve a valid rating.",
            "supplement_suggestion": "Please provide more conversational data."
        }

6. ワークフローノード関数の定義

各ノードは画像と会話データのアップロード、要件チェック、画像分析、会話分析、レポート生成、分析の再調整、最終レポートの保存を行います。画像や会話の感情スコアをグラフ化して視覚的にもわかりやすく提示します。

プロジェクトでは、LangGraphフレームワークを利用してステートグラフを可視化しています。各ノードの繋がりが明確に視覚化されることで、処理の流れが直感的に理解できます。また、実際の分析もこのグラフの流れに沿って自動的に実行され、最終的なレポートを出力します。

「ステートグラフ（State Graph）」について詳しく解説

このプロジェクトでは、処理の流れを「ステートグラフ」という形式で設計しています。ステートグラフとは、複数の処理ステップをそれぞれ「ノード（Node）」という単位で表し、それらをつないでワークフローを構成する仕組みです。各ノードは特定の役割を持っており、それぞれのノードを順番に実行することで、全体の処理が完成します。

以下に、このプロジェクトで使用している各ノードの役割や処理の詳細を詳しく紹介します。

各ノード（処理ステップ）の詳細説明

Node 1: データのアップロード（upload_data）

【目的】

ユーザーに、分析対象となる画像ファイルと会話履歴（テキストファイル）の場所を指定してもらい、それらをシステムに読み込みます。すでにデータがある場合、さらにデータを追加するか確認します。

【具体的な処理】

• 画像と会話履歴のファイルパス（場所）をユーザーから入力。
• 入力がない場合、デフォルトのパスを使用。
• 指定された画像ファイルが正常に開けるか、テキストファイルが読めるかを確認。
• 読み込んだデータを次の処理に渡すために、stateという辞書型オブジェクトに格納します。

Node 2: 過去の分析情報を利用（use_existing_info）

【目的】

過去に実施した分析結果が保存されている場合、その情報を現在の分析に追加で利用するかどうかをユーザーに確認します。

【具体的な処理】

• 現在のフォルダに過去の分析結果（JSONファイル）があるか検索。
• 過去の結果があれば、ユーザーに利用するかどうか尋ねる。
• ユーザーが使用を希望した場合、選択したファイルの内容を読み取り、以降の分析に活用。

Node 3: 分析要件チェック（check_requirements）

【目的】

読み込んだ画像と会話データを統合し、Googleの生成AI（Geminiモデル）を使って、分析に必要な最低限のデータが揃っているかを確認します。

【具体的な処理】

• 画像と会話のデータをもとに、AIに概要的な分析を依頼。
• AIから、「全体的な感情指数（総合スコア）」「信頼度」「評価理由」「追加で推奨する情報」などを取得。
• 信頼度が一定以下（例: 70点未満）の場合は、データ追加を要求するよう設計されています。

Node 4: 画像分析（analyze_image）

【目的】

読み込んだ各画像をGoogle Gemini AIで詳しく分析し、関係性や感情を数値で評価します。

【具体的な処理】

• 各画像について「距離感」「アイコンタクト」「表情」「身体的接触」の指標を数値化。
• 分析結果を後のレポート作成のためにリストに保管します。

Node 5: 会話分析（analyze_conversation）

【目的】

入力された会話履歴を分析し、感情傾向や関係性の質などを数値化して評価します。

【具体的な処理】

• 会話の中の「ポジティブな感情の割合」「ネガティブな感情の割合」「対話の主導権」「価値観の一致度」「関係性の温かさ」「毒性（不健全なやりとり）の可能性」などを算出。
• 算出したデータは次のレポート作成に活用されます。

Node 6: 総合レポート作成（generate_report）

【目的】

画像と会話の分析結果を統合し、AIが最終的な関係性の総合評価をレポートとして作成します。

【具体的な処理】

• 画像・会話それぞれの指標を平均化して総合スコアを算出。
• グラフを作成して画像と会話の分析結果を視覚的に提示。
• 全情報をもとに、Google Gemini AIに総合評価と具体的な改善提案を依頼。
• 詳細でわかりやすいレポートとして出力します。

Node 7: レポートへの追加情報の指示（guidance）

【目的】

作成されたレポートを表示し、ユーザーが追加情報による分析の改善を望むか確認します。

【具体的な処理】

• ユーザーが追加情報を求める場合、次の分析改善処理（Node 8）へ進む。
• 必要なければ、最終保存処理（Node 9）に進みます。

Node 8: 分析改善（refine_analysis）

【目的】

ユーザーが新たに提供した追加データ（新しい画像や会話）を元のデータに統合し、再度より精密な分析を行います。

【具体的な処理】

• 新たに追加されたデータを既存データに統合。
• 追加情報の重要度を高めて再分析を実施。
• より正確で詳細な最終レポートを生成します。

Node 9: レポート保存（save_report）

【目的】

最終的な分析レポートを、タイムスタンプ付きのJSON形式でローカルに保存します。

【具体的な処理】

• 現在の日時でファイル名を決定。
• JSON形式で分析レポートを保存し、将来の閲覧や利用に備えます。

# Node 1: upload_data
def upload_data(state: dict) -> dict:
    # Default file paths.
    default_image_path = "/kaggle/input/sample-pictures/sadcouple.png"
    default_conv_path = "/kaggle/input/sampletxt/angrychat.txt"
    default_new_conv_path = "/kaggle/input/sampletxt/coldchat.txt"
    default_new_image_path = "/kaggle/input/sample-pictures/coldcouple.png"
    
    # If data already exists, ask whether to append.
    if state.get("image_path") is not None and state.get("conv_path") is not None:
        if not isinstance(state["image_path"], list):
            state["image_path"] = [state["image_path"]]
        if not isinstance(state["conv_path"], list):
            state["conv_path"] = [state["conv_path"]]
            
        append_choice = input("Data already exists. Do you want to append new data? (yes/no): ").strip().lower()
        if append_choice == "yes":
            new_conv_path = input("Enter the path for additional conversation text (or press Enter for default or copy the datapath from dataset): ").strip()
            if not new_conv_path:
                new_conv_path = default_new_conv_path
            try:
                with open(new_conv_path, "r", encoding="utf-8") as f:
                    new_data = f.read()
                print("Additional conversation data read successfully.")
            except Exception as e:
                print("Error reading additional conversation data:", e)
            
            new_image_path = input("Enter the path for additional image (or press Enter for default or copy the datapath from dataset): ").strip()
            if not new_image_path:
                new_image_path = default_new_image_path
            try:
                with PILImage.open(new_image_path) as img:
                    img.verify()
                print("Additional image read successfully.")
            except Exception as e:
                print("Error reading additional image:", e)
    
            # Append the new data
            state["conv_data"] += "\nAdditional Data:\n" + new_data
            state["conv_path"].append(new_conv_path)
            state["image_path"].append(new_image_path)
        else:
            print("Keeping original data without appending.")
    else:
        user_upload_image = input("Enter the image file path (or press Enter for default or copy the datapath from dataset): ").strip()
        if not user_upload_image:
            user_upload_image = default_image_path
        
        user_upload_conv = input("Enter the conversation text file path (or press Enter for default or copy the datapath from dataset): ").strip()
        if not user_upload_conv:
            user_upload_conv = default_conv_path
        
        try:
            with open(user_upload_conv, "r", encoding="utf-8") as f:
                txtdata = f.read()
            print("Conversation data read successfully.")
        except Exception as e:
            print("Error reading conversation data:", e)
            txtdata = ""
        
        try:
            with PILImage.open(user_upload_image) as img:
                img.verify()
            print("Image loaded successfully.")
        except Exception as e:
            print("Error loading image:", e)
        
        state["conv_data"] = txtdata
        state["image_path"] = [user_upload_image]
        state["conv_path"] = [user_upload_conv]
        
        if not txtdata.strip():
            supplement = input("Conversation text is empty. Please provide missing data: ").strip()
            state["previous_data"] = supplement  # Initialize or append as needed.
    
    #print("Uploaded conversation data:")
    #print(state["conv_data"])
    print("Image file paths:", state["image_path"])
    print("Conversation file paths:", state["conv_path"])
    return state

# Node 2: use_existing_info
def use_existing_info(state: dict) -> dict:
    report_files = glob.glob("analysis_report_*.json")
    if not report_files:
        print("Step2 - No existing reports found; skipping existing info step.")
        state["previous_data"] = None
        return state
    
    answer = input("Existing report file detected. Use previous info for further analysis? (yes/no): ").strip().lower()
    if answer != "yes":
        state["previous_data"] = None
        print("Not using previous info.")
    else:
        print("Using existing info. Choose one of the following report files:")
        for idx, file in enumerate(report_files):
            print(f"{idx+1}. {file}")
        try:
            choice = int(input("Enter the report number: "))
            if 1 <= choice <= len(report_files):
                selected_file = report_files[choice - 1]
                with open(selected_file, "r", encoding="utf-8") as f:
                    previous_data = json.load(f)
                state["previous_data"] = previous_data
                print(f"Selected {selected_file} as previous info.")
            else:
                print("Choice out of range; not using previous info.")
                state["previous_data"] = None
        except Exception as e:
            print("Input error; not using previous info.", e)
            state["previous_data"] = None
    return state

# For print text color
BLUE = "\033[94m"
RED = "\033[91m"
GREEN = "\033[92m"
RESET = "\033[0m"


# Node 3: check_requirements
def check_requirements(state: dict) -> dict:
    photo_prompt = """
    Please carefully analyze this image and describe:
    - The physical proximity between the subjects.
    - Facial expressions.
    - Eye contact.
    - Any physical touch.
    Provide a brief textual description.
    """
    image_list = state.get("image_path", [])
    all_photo_description = ""
    if image_list:
        for img_path in image_list:
            with PILImage.open(img_path) as img:
                photo_response = client.models.generate_content(
                    model='gemini-2.0-flash',
                    contents=[img, photo_prompt]
                )
            try:
                photo_desc = photo_response.text.strip()
            except Exception:
                photo_desc = "Image description is not available."
            all_photo_description += f"[{img_path}]:\n{photo_desc}\n\n"
    else:
        all_photo_description = "No image data available."
    
    conv_text = state.get("conv_data", "")
    previous_text = state.get("previous_data", "")
    if isinstance(previous_text, dict):
        previous_text = json.dumps(previous_text, ensure_ascii=False)
    
    all_text = "Image Descriptions:\n" + all_photo_description + "\n\nConversation Text:\n" + conv_text
    if previous_text:
        all_text += "\n\nSupplemental Data:\n" + previous_text
    
    overall_prompt = (
        "Based on the following description, provide an overall emotional index (1-100), a confidence score (1-100), a one-sentence rationale, and a suggestion for additional input.\n"
        "Return in pure JSON format as follows:\n"
        '{\n'
        '  "comprehensive_emotional_index": number,\n'
        '  "confidence_score": number,\n'
        '  "rating_reason": "summary sentence",\n'
        '  "supplement_suggestion": "additional info suggestion"\n'
        '}\n'
        "Description:\n" + all_text
    )
    
    output_config = types.GenerateContentConfig(
        temperature=0.0,
        response_mime_type="application/json",
        response_schema=SummaryRating,
    )
    
    #print("Combined description for analysis:\n", all_text)
    overall_response = client.models.generate_content(
        model="gemini-2.0-flash",
        contents=[overall_prompt],
        config=output_config
    )
    #print("Original response:", overall_response)
    
    overall_data = extract_json_from_response(overall_response.text)
    print(f"\n{BLUE}===== Checking the Quality of the Uploaded Document ====={RESET}\n")

    comprehensive_emotional_index = overall_data.get("comprehensive_emotional_index", "N/A")
    confidence_score = overall_data.get("confidence_score", "N/A")
    rating_reason = overall_data.get("rating_reason", "No reason provided.")
    supplement_suggestion = overall_data.get("supplement_suggestion", "No suggestion provided.")
    
    print(f"{BLUE}【Quick Comprehensive Relationship Analysis】{RESET}")
    print(f"{comprehensive_emotional_index}\n")
    
    print(f"{BLUE}【Confidence Score】{RESET}")
    print(f"{GREEN}{confidence_score}{RESET} (70 or above is acceptable)\n")
    
    print(f"{BLUE}【Rating Reason】{RESET}")
    print(f"{rating_reason}\n")
    
    print(f"{BLUE}【Supplement Suggestion】{RESET}")
    print(f"{supplement_suggestion}\n")
    
    # Decide whether the requirements are met based on the confidence_score
    if overall_data.get("confidence_score", 0) < 70:
        suggestion = overall_data.get("supplement_suggestion", "Please provide additional data to improve confidence score.")
        print(f"Confidence score ({overall_data.get('confidence_score')}) is below 70; please provide additional data. Suggestion: {suggestion}")
        state["requirements_met"] = False
    else:
        state["requirements_met"] = True

    # Save the overall rating data into state for further use
    state["overall_rating"] = overall_data
    return state

# Node 4: analyze_image
def analyze_image(state: dict) -> dict:
    image_paths = state.get("image_path", [])
    prompt = (
        "Analyze this image with the following requirements:\n"
        "- Provide a brief description of the image.\n"
        "- Assign a proximity_score (0 to 1).\n"
        "- Assign an eye_contact_score (0 to 1).\n"
        "- Assign a facial_expression_score (0 to 1).\n"
        "- Assign a body_touch_score (0 to 1).\n"
        "- Provide a short rationale.\n"
        "Return the result as JSON conforming to the provided schema."
    )
    results = []
    for img_path in image_paths:
        try:
            with PILImage.open(img_path) as img:
                output_config = types.GenerateContentConfig(
                    temperature=0.0,
                    response_mime_type="application/json",
                    response_schema=ImageAnalysisResponse,
                )
                photo_response = client.models.generate_content(
                    model='gemini-2.0-flash',
                    contents=[img, prompt],
                    config=output_config
                )
                image_result = json.loads(photo_response.text)
                results.append(image_result)
        except Exception as e:
            results.append({
               "description": "No analysis result available.",
               "proximity_score": 0.5,
               "eye_contact_score": 0.5,
               "facial_expression_score": 0.5,
               "body_touch_score": 0.5,
               "reason": f"Analysis failed: {str(e)}"
            })
    state["image_analysis"] = results
    return state

# Node 5: analyze_conversation
def analyze_conversation(state: dict) -> dict:
    conv_text = state.get("conv_data", "")
    previous_text = state.get("previous_data", "")
    if isinstance(previous_text, dict):
        previous_text = json.dumps(previous_text, ensure_ascii=False)
    all_text = "Conversation Text:\n" + conv_text
    if previous_text:
        all_text += "\n\nSupplemental Data:\n" + previous_text

    text_prompt = (
        "Analyze the following conversation text. Evaluate:\n"
        "1. Positive and negative sentiment ratios.\n"
        "2. Initiative (who speaks first more often), producing an initiative_score (0-1).\n"
        "3. Value alignment regarding major values, producing value_alignment_score (0-1).\n"
        "4. Relationship warmth (0-1).\n"
        "5. Probability of toxic behavior (0-1, lower means less likely).\n"
        "Return in JSON format as follows:\n"
        '{\n'
        '  "sentiment_summary": {"positive_ratio": 0.7, "negative_ratio": 0.1},\n'
        '  "initiative_score": 0.6,\n'
        '  "value_alignment_score": 0.8,\n'
        '  "relationship_warmth_score": 0.85,\n'
        '  "toxicity_probability": 0.15,\n'
        '  "reason": "Analysis rationale."\n'
        '}\n'
        "Conversation text:\n" + all_text
    )
    
    output_config = types.GenerateContentConfig(
        temperature=0.0,
        response_mime_type="application/json",
        response_schema=ConversationAnalysisResponse,
    )
    conversation_response = client.models.generate_content(
        model="gemini-2.0-flash",
        contents=[text_prompt],
        config=output_config
    )
    conversation_result = json.loads(conversation_response.text)
    state["conversation_analysis"] = conversation_result
    return state

# Node 6: generate_report
def generate_report(state: dict) -> dict:
    # ---------------------------
    # Retrieve image analysis results (list of dicts)
    image_analysis_list = state.get("image_analysis", [])
    if not image_analysis_list:
        avg_proximity = avg_eye_contact = avg_facial = avg_body_touch = 0.5
        image_reason = "No image analysis data available."
    else:
        proximity_scores = [img.get("proximity_score", 0.5) for img in image_analysis_list]
        eye_contact_scores = [img.get("eye_contact_score", 0.5) for img in image_analysis_list]
        facial_scores = [img.get("facial_expression_score", 0.5) for img in image_analysis_list]
        body_touch_scores = [img.get("body_touch_score", 0.5) for img in image_analysis_list]
        
        avg_proximity = sum(proximity_scores) / len(proximity_scores)
        avg_eye_contact = sum(eye_contact_scores) / len(eye_contact_scores)
        avg_facial = sum(facial_scores) / len(facial_scores)
        avg_body_touch = sum(body_touch_scores) / len(body_touch_scores)
        
        reasons = [img.get("reason", "") for img in image_analysis_list if img.get("reason")]
        image_reason = "；".join(reasons) if reasons else "No image analysis rationale available."
    
    image_composite = (avg_proximity + avg_eye_contact + avg_facial + avg_body_touch) / 4

    # ---------------------------
    # Retrieve conversation analysis results (dictionary)
    conv_data = state.get("conversation_analysis", {})
    sentiment = conv_data.get("sentiment_summary", {})
    positive_ratio = sentiment.get("positive_ratio", 0.5)
    initiative_score = conv_data.get("initiative_score", 0.5)
    value_alignment_score = conv_data.get("value_alignment_score", 0.5)
    relationship_warmth_score = conv_data.get("relationship_warmth_score", 0.5)
    toxicity_probability = conv_data.get("toxicity_probability", 0.5)
    conv_reason = conv_data.get("reason", "No conversation analysis rationale available.")
    
    conv_composite = (initiative_score + value_alignment_score + relationship_warmth_score + (1 - toxicity_probability) + positive_ratio) / 5

    # ---------------------------
    # Calculate overall composite score (40% image, 60% conversation)
    overall_composite = 0.4 * image_composite + 0.6 * conv_composite
    overall_rating = round(overall_composite * 100)
    
    base_composite_reason = f"From image analysis: {image_reason}; From conversation analysis: {conv_reason}."

    # ---------------------------
    # Prepare metrics for plotting.
    image_metrics = ["Proximity", "Eye Contact", "Facial Expression", "Body Touch"]
    image_values = [avg_proximity, avg_eye_contact, avg_facial, avg_body_touch]
    
    conv_metrics = ["Initiative", "Value Alignment", "Relationship Warmth", "Positivity", "1 - Toxicity"]
    conv_values = [initiative_score, value_alignment_score, relationship_warmth_score, positive_ratio, 1 - toxicity_probability]
    
    # Unified chart function call.
    chart_filename = display_chart(image_metrics, image_values, conv_metrics, conv_values, overall_rating)
    
    # ---------------------------
    # Combine conversation text and supplemental data.
    conv_text = state.get("conv_data", "")
    previous_text = state.get("previous_data", "")
    all_text = "Conversation Text:\n" + conv_text
    if previous_text:
        if not isinstance(previous_text, str):
            previous_text = json.dumps(previous_text, ensure_ascii=False, indent=2)
        all_text += "\n\nSupplemental Data:\n" + previous_text

    # ---------------------------
    # Build prompt to call LLM for composite report.
    prompt = (
        "You are an expert psychologist and relationship counselor. Please use plain and clear language to analyze the data "
        "provided below and generate:\n"
        "1. A 'composite_reason': a brief, bullet-point summary of the key observations from the image analysis and conversation analysis.\n"
        "2. A 'detailed_report': a detailed explanation of the current relationship status, highlighting strengths, issues, and practical recommendations for improvement.\n\n"
        "【Image Analysis Rationale】\n"
        f"{image_reason}\n\n"
        "【Conversation Analysis Rationale】\n"
        f"{conv_reason}\n\n"
        "【Image Analysis Data】\n"
        f"{json.dumps(image_analysis_list, ensure_ascii=False, indent=2)}\n\n"
        "【Conversation Data】\n"
        f"{all_text}\n\n"
        "Return the result in the following JSON format without any extra text:\n"
        '{\n'
        '  "composite_reason": "Your summary here",\n'
        '  "detailed_report": "Your detailed analysis and recommendations here"\n'
        '}'
    )
    
    output_config = types.GenerateContentConfig(
        temperature=0.0,
        response_mime_type="application/json",
        response_schema=CompositeReport,
    )
    
    llm_response = client.models.generate_content(
        model="gemini-2.0-flash",
        contents=[prompt],
        config=output_config
    )
    
    try:
        composite_data = json.loads(llm_response.text)
    except Exception as e:
        composite_data = {
            "composite_reason": base_composite_reason,
            "detailed_report": (
                f"Preliminary Analysis: Overall rating is {overall_rating} out of 100. "
                "The data suggests communication and emotional connection issues. "
                "It is recommended that both parties work on open communication, rebuild trust, and consider professional counseling if needed."
            )
        }
    
    composite_reason_final = composite_data.get("composite_reason", base_composite_reason)
    detailed_report_final = composite_data.get("detailed_report", "")
    
    # ---------------------------
    # Assemble the final comprehensive report.
    final_insights = detailed_report_final 
    
    report = {
        "image_analysis": image_analysis_list,
        "conversation_analysis": conv_data,
        "previous_data": state.get("previous_data"),
        "overall_composite_score": overall_rating,
        "composite_reason": composite_reason_final,
        "chart_file": chart_filename,
        "detailed_report": final_insights
    }
    
    state["report"] = report
    return state


# Node 7: guidance
def guidance(state: dict) -> dict:
    report = state.get("report", {})
    print("\n===== Composite Analysis Report =====\n")
   
    # Overall score
    overall_score = report.get("overall_composite_score", "N/A")
    print(f"{BLUE}【Overall Emotional Rating】{RESET}")
    print(f"Your overall relationship rating is {RED}{overall_score}{RESET} out of 100.\n")
    
    # Key observations
    composite_reason = report.get("composite_reason", "No summary available.")
    print(f"{BLUE}【Summary of Key Observations】{RESET}")
    print(f"{composite_reason}\n")
    
    # Detailed report
    detailed_report = report.get("detailed_report", "No detailed report available.")
    print(f"{BLUE}【Detailed Analysis & Recommendations】{RESET}")
    print(f"{detailed_report}\n")
    
    # Chart file
    chart_file = report.get("chart_file", "No chart file.")
    print(f"{BLUE}【Chart File】{RESET}")
    print(f"{chart_file}\n")
    
    need_more = input("Would you like to add new file data (photo or text) for further report refinement? (yes/no): ").strip().lower()
    if need_more == "yes":
        state["branch"] = "refined"
    else:
        state["branch"] = "final"
    return state


# Node 8: refine analysis
def refine_analysis(state: dict) -> dict:
    """
    Incorporate new file data (photo and text) provided by the client,
    update supplemental data, and re-calculate image and conversation analyses based on the new inputs.
    Then, call the LLM to generate an updated composite report.
    The final output report follows the same format as Nodes 5-7, with branch set to "final".
    """
    new_data_str = ""
    
    # Process new photo file (optional)
    new_photo_path = input("Enter new photo file path (or press Enter if not applicable, or copy the file path from your dataset): ").strip()
    if new_photo_path:
        new_image_analysis = analyze_image_for_file(new_photo_path)
        image_analysis_list = state.get("image_analysis", [])
        image_analysis_list.append(new_image_analysis)
        state["image_analysis"] = image_analysis_list
        new_data_str += f"New photo file provided and analyzed: {new_photo_path}. "
    
    # Process new text file for conversation (optional)
    new_text_path = input("Enter new text file path (or press Enter if not applicable, or copy the file path from your dataset): ").strip()
    if new_text_path:
        try:
            with open(new_text_path, "r", encoding="utf-8") as f:
                new_text = f.read()
            new_data_str += f"New text data provided from {new_text_path}:\n{new_text}\n"
            # Reanalyze conversation using the new text data.
            new_conv_analysis = analyze_conversation_for_text(new_text)
            state["conversation_analysis"] = new_conv_analysis
            # Update the raw conversation text to include new text.
            conv_text = state.get("conv_data", "")
            state["conv_data"] = conv_text + "\n\n" + new_text
        except Exception as e:
            print("Error reading new text file:", e)
    
    # Update supplemental data with new file information.
    if new_data_str:
        original_previous_data = state.get("previous_data", "")
        combined_data = (original_previous_data + "\n*** Additional Data (High Weight) ***\n" + new_data_str) if original_previous_data else new_data_str
        state["previous_data"] = combined_data
    
    # Set branch to "final" so the flow ends with the final report.
    state["branch"] = "final"
    
    # --- Recalculate key metrics using updated data ---
    # For image analysis:
    image_analysis_list = state.get("image_analysis", [])
    if image_analysis_list:
        proximity_scores = [img.get("proximity_score", 0.5) for img in image_analysis_list]
        eye_contact_scores = [img.get("eye_contact_score", 0.5) for img in image_analysis_list]
        facial_scores = [img.get("facial_expression_score", 0.5) for img in image_analysis_list]
        body_touch_scores = [img.get("body_touch_score", 0.5) for img in image_analysis_list]
        avg_proximity = sum(proximity_scores) / len(proximity_scores)
        avg_eye_contact = sum(eye_contact_scores) / len(eye_contact_scores)
        avg_facial = sum(facial_scores) / len(facial_scores)
        avg_body_touch = sum(body_touch_scores) / len(body_touch_scores)
    else:
        avg_proximity = avg_eye_contact = avg_facial = avg_body_touch = 0.5
    
    image_composite = (avg_proximity + avg_eye_contact + avg_facial + avg_body_touch) / 4
    
    # For conversation analysis:
    conv_data = state.get("conversation_analysis", {})
    sentiment = conv_data.get("sentiment_summary", {})
    positive_ratio = sentiment.get("positive_ratio", 0.5)
    initiative_score = conv_data.get("initiative_score", 0.5)
    value_alignment_score = conv_data.get("value_alignment_score", 0.5)
    relationship_warmth_score = conv_data.get("relationship_warmth_score", 0.5)
    toxicity_probability = conv_data.get("toxicity_probability", 0.5)
    conv_composite = (initiative_score + value_alignment_score + relationship_warmth_score + (1 - toxicity_probability) + positive_ratio) / 5
    
    overall_composite = 0.4 * image_composite + 0.6 * conv_composite
    overall_rating = round(overall_composite * 100)
    
    # Prepare base composite reason (using prior report if available)
    report = state.get("report", {})
    base_composite_reason = report.get("composite_reason", "Original composite observations unavailable.")
    
    # Retrieve conversation raw text and supplemental data.
    conv_text = state.get("conv_data", "")
    previous_data = state.get("previous_data", "")
    all_text = "Conversation Text:\n" + conv_text
    if previous_data:
        if not isinstance(previous_data, str):
            previous_data = json.dumps(previous_data, ensure_ascii=False, indent=2)
        all_text += "\n\nSupplemental Data:\n" + previous_data

    # --- Build prompt for LLM to generate updated composite report ---
    new_prompt = (
        "You are an expert psychologist and relationship counselor. Based on the updated data provided below, please generate an updated composite report. "
        "The newly provided file data should be given higher weight in the evaluation. Your response must include:\n"
        "1. 'composite_reason': A concise bullet-point summary of key observations derived from the image and conversation analyses, highlighting any new insights due to the additional data.\n"
        "2. 'detailed_report': A comprehensive analysis of the current relationship status, including strengths, issues, and specific, actionable recommendations for improvement.\n\n"
        "【Original Composite Observations】\n"
        f"{base_composite_reason}\n\n"
        "【Image Analysis Data】\n"
        f"{json.dumps(image_analysis_list, ensure_ascii=False, indent=2)}\n\n"
        "【Conversation Data】\n"
        f"{all_text}\n\n"
        "Return your response in JSON format as follows (with no extra text):\n"
        '{\n'
        '  "composite_reason": "Your revised composite reason",\n'
        '  "detailed_report": "Your revised detailed report"\n'
        '}'
    )
    
    output_config = types.GenerateContentConfig(
        temperature=0.0,
        response_mime_type="application/json",
        response_schema=CompositeReport,
    )
    
    llm_response = client.models.generate_content(
        model="gemini-2.0-flash",
        contents=[new_prompt],
        config=output_config
    )
    
    try:
        new_composite_data = json.loads(llm_response.text)
    except Exception as e:
        new_composite_data = {
            "composite_reason": base_composite_reason,
            "detailed_report": (
                f"Preliminary Analysis: The overall rating remains {overall_rating} out of 100. Further professional consultation is advised."
            )
        }
    
    refined_composite_reason = new_composite_data.get("composite_reason", base_composite_reason)
    refined_detailed_report = new_composite_data.get("detailed_report", "")
    
    final_insights = refined_detailed_report 
    
    final_report = {
        "image_analysis": image_analysis_list,
        "conversation_analysis": conv_data,
        "previous_data": state.get("previous_data"),
        "overall_composite_score": overall_rating,
        "composite_reason": refined_composite_reason,
        "chart_file": report.get("chart_file", "N/A"),
        "detailed_report": final_insights
    }
    
    state["report"] = final_report
    
    # Recompute metrics for chart display.
    if image_analysis_list:
        proximity_scores = [img.get("proximity_score", 0.5) for img in image_analysis_list]
        eye_contact_scores = [img.get("eye_contact_score", 0.5) for img in image_analysis_list]
        facial_scores = [img.get("facial_expression_score", 0.5) for img in image_analysis_list]
        body_touch_scores = [img.get("body_touch_score", 0.5) for img in image_analysis_list]
        avg_proximity = sum(proximity_scores) / len(proximity_scores)
        avg_eye_contact = sum(eye_contact_scores) / len(eye_contact_scores)
        avg_facial = sum(facial_scores) / len(facial_scores)
        avg_body_touch = sum(body_touch_scores) / len(body_touch_scores)
    else:
        avg_proximity = avg_eye_contact = avg_facial = avg_body_touch = 0.5
    
    image_metrics = ["Proximity", "Eye Contact", "Facial Expression", "Body Touch"]
    image_values = [avg_proximity, avg_eye_contact, avg_facial, avg_body_touch]
    
    initiative_score = conv_data.get("initiative_score", 0.5)
    value_alignment_score = conv_data.get("value_alignment_score", 0.5)
    relationship_warmth_score = conv_data.get("relationship_warmth_score", 0.5)
    sentiment = conv_data.get("sentiment_summary", {})
    positive_ratio = sentiment.get("positive_ratio", 0.5)
    toxicity_probability = conv_data.get("toxicity_probability", 0.5)
    conv_metrics = ["Initiative", "Value Alignment", "Relationship Warmth", "Positivity", "1 - Toxicity"]
    conv_values = [initiative_score, value_alignment_score, relationship_warmth_score, positive_ratio, 1 - toxicity_probability]
    
    chart_filename = display_chart(image_metrics, image_values, conv_metrics, conv_values, overall_rating)

    # After updating state["report"], print it out elegantly:
    report = state.get("report", {})
    
    print("\n===== Refined Composite Analysis Report =====\n")
    
    # Overall score
    overall_score = report.get("overall_composite_score", "N/A")
    print(f"{BLUE}【Overall Emotional Rating】{RESET}")
    print(f"Your overall relationship rating is {RED}{overall_score}{RESET} out of 100.\n")
    
    # Key observations
    composite_reason = report.get("composite_reason", "No summary available.")
    print(f"{BLUE}【Summary of Key Observations】{RESET}")
    print(f"{composite_reason}\n")
    
    # Detailed report
    detailed_report = report.get("detailed_report", "No detailed report available.")
    print(f"{BLUE}【Detailed Analysis & Recommendations】{RESET}")
    print(f"{detailed_report}\n")
    
    # Chart file
    chart_file = report.get("chart_file", "No chart file.")
    print(f"{BLUE}【Chart File】{RESET}")
    print(f"{chart_file}\n")
    
    state["report"]["chart_file"] = chart_filename
    
    return state

# Node 9: save_report
def save_report(state: dict) -> dict:
    report = state.get("report", {})
    filename = f"analysis_report_{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
    try:
        with open(filename, "w", encoding="utf-8") as f:
            json.dump(report, f, ensure_ascii=False, indent=2)
        print(f"Final Report saved to {filename}")
    except Exception as e:
        print("Failed to save report:", e)
    return state

7. ワークフロー制御用分岐関数

信頼性スコアに応じて、追加データを要求するか、分析を進めるかを判断します。

# Correct branch_requirements function
def branch_requirements(state: dict) -> Literal["analyze_image", "upload_data"]:
    return "analyze_image" if state.get("requirements_met") else "upload_data"

# Branching based on guidance outcome:
def branch_guidance(state: dict) -> Literal["refine_analysis", "save_report"]:
    return "refine_analysis" if state.get("branch") == "refined" else "save_report"

8. 状態グラフ（StateGraph）の構築

各処理をノードとして登録し、条件分岐で制御しながら順次処理を進めます。

def build_state_graph():
    graph_builder = StateGraph(dict)
    graph_builder.add_node("upload_data", upload_data)
    graph_builder.add_node("use_existing_info", use_existing_info)
    graph_builder.add_node("check_requirements", check_requirements)
    graph_builder.add_node("analyze_image", analyze_image)
    graph_builder.add_node("analyze_conversation", analyze_conversation)
    graph_builder.add_node("generate_report", generate_report)
    graph_builder.add_node("guidance", guidance)
    graph_builder.add_node("refine_analysis", refine_analysis)
    graph_builder.add_node("save_report", save_report)
    
    graph_builder.add_edge(START, "upload_data")
    graph_builder.add_edge("upload_data", "use_existing_info")
    graph_builder.add_edge("use_existing_info", "check_requirements")
    graph_builder.add_conditional_edges("check_requirements", branch_requirements)
    graph_builder.add_edge("analyze_image", "analyze_conversation")
    graph_builder.add_edge("analyze_conversation", "generate_report")
    graph_builder.add_edge("generate_report", "guidance")
    graph_builder.add_conditional_edges("guidance", branch_guidance)
    graph_builder.add_edge("refine_analysis", "save_report")
    graph_builder.add_edge("save_report", END)
    
    return graph_builder.compile()

9. グラフの視覚化と実行

Mermaidを用いてワークフロー全体を図示し、分析プロセスの透明性を高めます。また、グラフを実行して最終的な分析レポートを生成します。

このプロジェクトを通じて、Gen AI技術がどのように人間関係の深層分析に役立つかを具体的に示すことができました。

from IPython.display import Image, display

# Build the state graph from the previously defined nodes and edges.
graph = build_state_graph()

===== コンポジット分析レポート =====

【全体感情評価】
あなたたちの全体的な関係評価は 100点中38点 です。

【主な観察結果】
主な観察内容は以下の通りです：

• 感情的な断絶：画像では身体的な近さは見られるものの、アイコンタクトがなく、真剣な表情をしており、感情的な距離が示唆されています。
• 期待の不満：エミリーは、約束が破られたことや自分が大切にされていないと感じることに失望を表明しています。
• 防御的態度と責任転嫁：アレックスは防御的になり、エミリーを非難して口論を悪化させています。
• コミュニケーションの崩壊：会話から、アレックスがエミリーの話を聞かず、エミリーが理解されていないと感じているパターンが明らかになりました。
• 関係の悪化：疲弊し、感謝されていないと感じるため、二人は距離を置くことを検討しています。

【詳細分析と推奨事項】
現在の関係状況：
現在、二人の関係は非常に緊張しており、重大な課題に直面しています。画像分析では、身体的な近さに反して感情的なつながりが欠如していることが示されています。会話分析では、期待が満たされないこと、防御的態度、コミュニケーションの問題が明らかとなり、恨みや疲労感が高まっています。

強み：

• 関係改善への意欲：アレックスもエミリーも、関係を改善したいという意欲を表明しています。アレックスは努力を誓い、エミリーもそれを信じたいと望んでいます。
• 互いの重要性の認識：アレックスはエミリーが自分にとって大切な存在であると述べており、根底には愛情があることが伺えます。

問題点：

• 約束違反と期待の不満：アレックスが約束を守れないことで、エミリーは大切にされていないと感じています。
• 防御的なコミュニケーション：アレックスの防御的態度と責任転嫁が口論をエスカレートさせ、解決を困難にしています。
• 傾聴の欠如：エミリーは自分の意見が聞かれていないと感じています。
• 感情的な距離：アイコンタクトの欠如と真剣な表情から、感情的な断絶が推察されます。
• 口論の悪化：会話パターンから、口論がエスカレートし、問題が解決されない傾向が見られます。

推奨事項：

• コミュニケーションスキルの向上：互いのニーズや感情を非難せずに表現し、積極的に傾聴し、共感を持って対応することを心がけましょう。カップルセラピーに参加し、効果的なコミュニケーション技術を学ぶのも有効です。
• 信頼の再構築：アレックスは約束を確実に守り、誠実な行動を積み重ねることでエミリーの信頼を取り戻す必要があります。小さな一歩の積み重ねが大きな効果を生みます。
• 質の高い時間を優先する：感情的なつながりを取り戻すために、二人で楽しめる活動を計画し、特別な時間を共有しましょう。
• 根本的な問題に取り組む：コミュニケーションの問題や期待の不満の根本原因を探りましょう。セラピーは、これらの課題に安全な環境で向き合い、健全な対処法を学ぶ場を提供してくれます。
• 共感を実践する：互いの立場や感情を理解し、認め合うことで、防御的態度を減らし、支え合う関係を築くことができます。
• 専門家の支援を求める：努力しても問題が解決しない場合は、カップルセラピストに相談することを検討してください。専門家のサポートを受けながら、課題に向き合い、関係を改善していきましょう。

読者の皆さんへ：よくある質問とプロジェクトについての補足説明

今回のプロジェクトについて、読者の皆さんが感じるかもしれない疑問を、Q&A形式でまとめました。

Q1：このプロジェクトは具体的に何を解決するために作られたのですか？

このプロジェクトでは、カップルや夫婦が抱える「自分たちの関係性やコミュニケーションの問題」をAI技術を活用して分析し、改善のきっかけを与えることを目的にしています。

例えば、

• 「最近パートナーとうまく会話ができていない」
• 「写真に写る二人の関係性が冷めているように感じる」

といった漠然とした悩みを持つ人に対して、AIが客観的な分析を提供することで、関係改善への具体的な第一歩を提案します。

Q2：AI技術をどのように使って、この問題を解決しているのですか？

私のプロジェクトでは、Googleが提供する最先端の生成AI（Gemini）を使い、次の2つの分析を実施しています。

• 画像分析：
  カップルの写真をAIが分析し、「身体の距離感」「アイコンタクト」「表情」「触れ合い方」などを数値化します。
• 会話履歴分析：
  二人の会話履歴を分析し、「ポジティブ/ネガティブな感情比率」「コミュニケーションの主導権」「お互いの価値観の一致度」「関係性の温かさ」などを数値で評価します。

これらの分析結果を総合してAIが関係性を評価し、さらに改善するための具体的なアドバイスをレポート形式で提示します。

実際の分析処理の流れを示すコードの一例も記事内に記載しました。
（例えば、画像を読み込む処理や、AIが分析結果を返す様子をコードで紹介しています。）

Q3：今回のプロジェクトで使ったAI技術には、どんな課題や限界がありますか？

AI技術には次のような課題や限界があります。

• AIが分析をするためのデータ（写真や会話）が不十分だったり質が低い場合、正確な評価が難しくなります。
• 人間の感情や関係性は非常に複雑で微妙なニュアンスが多いため、AIが完全に理解・評価することには限界があります。
• AIの分析結果はあくまで参考であり、最終的には人間自身の判断や専門家の支援が必要になるケースも多くあります。

Q4：今後、このプロジェクトをどのように活用・発展させていく予定ですか？

このプロジェクトをもとに、誰でも気軽に利用できるスマホアプリとして開発していく予定です。

将来的には、

• 個人が自宅で簡単に自分の関係性を分析し、改善する方法を学べるようにしたい
• カウンセラーやセラピストなど、専門家がカップルのサポートを行う際の補助ツールとして活用してもらいたい

と考えています。

技術がさらに進歩することで、AIによる関係性分析がもっと正確かつ深いレベルで可能になることを期待しています。

Nagi

アプリ開発フリーランス

XInstagram

現在海外在住。AIとアプリ開発、自由な働き方に関する最新情報を皆様にお届けすべく、日々挑戦中です。 趣味はアプリ制作、読書、カフェ巡り。副業・フリーランス生活についても発信中！