序章従来のキーワードベースのコンテキストコンテンツ配置では、皮肉や一見するとわからない関連性といったニュアンスを見逃すことが多く、広告を掲載するのに最適な場所を見つけるのは非常に困難です。このブログでは、Databricks 上に構築された AI Agent が、これらの制限を乗り越え、非常にニュアンスに富んだ、深い文脈に基づいたコンテンツ配置を実現する方法を紹介します。ここでは、映画やテレビの脚本を例に、コンテンツが最も大きな影響を与える具体的なシーンや瞬間を把握し、これを実現する方法を探っていきます。この特定の例に焦点を当てていますが、このコンセプトは、テレビの脚本、音声スクリプト(ポッドキャストなど)、ニュース記事、ブログなど、より広範なメディアデータのカタログに一般化できます。あるいは、これをプログラマティック広告向けに応用することも考えられます。その場合、入力データとして広告コンテンツのコーパス、関連メタデータ、掲載情報を使用し、エージェントがダイレクトプログラマティックまたは広告サーバー経由の配置を最適化するための適切なタグ付けを生成します。ソリューションの概要このソリューションは、Agent Framework、Vector Search、Unity Catalog、MLflow 3.0 による Agent Evaluation など、Databricks の AI Agent ツールにおける最新の技術を活用しています。下の図は、アーキテクチャの概要を示したものです。データソース: クラウドストレージや外部システムに保存された映画の脚本やメディアコンテンツデータの前処理: 非構造化テキストを取り込み、解析、クレンジング、チャンク化します。次に、処理されたテキストチャンクから埋め込みを作成し、レトリーバーツールとして使用できるように Databricks Vector Store でインデックス化します。エージェントの開発: コンテンツ配置エージェントは、Unity Catalog Function、LangGraph、MLflow、および任意のLLM (この例ではClaudeモデルを使用) でラップされたベクトル検索レトリーバーツールを活用します。エージェント評価: エージェントの品質は、LLM 判定、カスタム判定、人間によるフィードバック、反復的な開発ループを通じて継続的に改善されますエージェントのデプロイ: エージェントフレームワークは、エージェントを Databricks のモデルサービングエンドポイントにデプロイし、AI Gateway を通じてガバナンス、セキュリティ保護、モニタリングが行われます。アプリの使用法: Databricks Apps、カスタムアプリ、または従来の広告技術スタックを介してエージェントをエンドユーザーに公開し、継続的な品質向上のためにすべてのユーザーフィードバックとログを Databricks に記録します。実用的な観点から言うと、このソリューションにより、広告販売者は説明に基づいて、広告コンテンツを配置するのに最適な場所をコンテンツコーパス内で自然言語で問い合わせることができます。そこでこの例では、データセットに大量の映画の脚本が含まれているとして、エージェントに「ペットフードの広告はどこに配置できますか?」と質問したとします。広告はボウルから餌を食べているビーグルの画像です」と続けると、エージェントはAir BudやMarley & Meといった有名な犬の映画から、特定のシーンを返すことが期待されます。以下は、当社のエージェントによる実際の例です。ソリューションの概要を理解したところで、エージェントを構築するためのデータ準備の方法を詳しく見ていきましょう。データの前処理コンテキストに応じた配置のための映画データの前処理エージェントにレトリーバーツール (検索拡張生成 (RAG) と呼ばれる手法) を追加する場合、高品質を実現するためにはデータ処理パイプラインが非常に重要なステップとなります。この例では、堅牢な非構造化データパイプラインを構築するためのベストプラクティスに従います。これには通常、4つのステップが含まれます:解析チャンキング埋め込みインデックス作成このソリューションで使用するデータセットには、1200本分の映画の脚本全文が含まれており、これらを個別のテキストファイルとして保存します。広告コンテンツを最も文脈的に関連性の高い方法で配置するため、私たちの前処理戦略は、映画そのものではなく、映画の特定のシーンを推奨することです。カスタムシーン解析まず、生のトランスクリプトを解析し、標準的な脚本の書式(例:「INT」、「EXT」など)をシーンの区切り文字として使用して、各脚本ファイルを個々のシーンに分割します。これにより、関連するメタデータ(例:タイトル、シーン番号、シーンの場所)を抽出してデータセットを充実させ、Deltaテーブルに生のトランスクリプトとともに保存できます。scene_header_pattern = r"\b(INT|EXT|INTERIOR|EXTERIOR)(\.|\s*\/\s*\.?\s*(EXT|INT|INTERIOR|EXTERIOR)\.?|\.)?(\s|$)" scene_header_regex = re.compile(scene_header_pattern,re.IGNORECASE | re.MULTILINE) @udf("struct<scene_count: int, scenes: array<struct<header: string, scene_number: int, text: string>>>") def extract_scenes(script_text): scenes = [] matches = list(scene_header_regex.finditer(script_text)) # Extract text between headers for i, match in enumerate(matches): print(match) header = match.group().strip()start = match.end()# End of the header line end = matches[i+1].start()if i < len(matches)-1 else len(script_text) scene_text = script_text[start:end].strip()#.replace("\r\n", "").replace("\t","") scene_text_cleaned = re.sub(r"\s+"," ", scene_text).strip()# scene_text = script_text[start:end].replace("\r\n","").replace("\t","").strip()scenes.append((header,scene_text_cleaned)) return {"scene_count": len(scenes), "scenes": [{"header": header, "scene_number": i+1, "text": scene_text} for i, (header, scene_text) in enumerate(scenes)]}シーンを意識した固定長チャンキング戦略次に、クレンジングされたシーンデータに対して固定長のチャンキング戦略を実装し、このユースケースでは取得してもあまり価値がないため、短いシーンは除外します。@pandas_udf(ArrayType(StringType()), PandasUDFType.SCALAR) def chunk_scenes_pandas_udf(scenes: pd.Series) -> pd.Series: # Fixed chunk sizes directly within the UDF's scope min_chunk_size = 50 max_chunk_size = 500 results = [] for scene in scenes: if scene is None: results.append([])continue scene_chunks = text_splitter.split_text(scene)chunks = [] previous_chunk = "" for c in scene_chunks: current_chunk_stripped = c.strip() if not current_chunk_stripped and not previous_chunk: continue # Ensure proper concatenation before encoding for combined length check combined_text = (previous_chunk + "\n" + current_chunk_stripped).strip() if previous_chunk else current_chunk_stripped encoded_len = len(tokenizer.encode(combined_text))if encoded_len <= max_chunk_size / 2: # Keep original logic for max_chunk_size previous_chunk = combined_text # Update previous_chunk with the combined text else: # Process the accumulated previous_chunk split_prev = text_splitter.split_text(previous_chunk.strip()) if split_prev: chunks.extend(split_prev)# Start new previous_chunk with the current chunk previous_chunk = current_chunk_stripped # Start new previous_chunk if previous_chunk: split_prev = text_splitter.split_text(previous_chunk.strip()) if split_prev: chunks.extend(split_prev)# Filter chunks by min_chunk_size filtered_chunks = [c for c in chunks if len(tokenizer.encode(c)) > min_chunk_size] results.append(filtered_chunks)return pd.Series(results)注:当初は固定長のチャンク(これは脚本全体よりは良かったでしょう)を検討しましたが、シーンの区切り文字で分割することで、レスポンスの関連性が大幅に向上しました。Vector Searchリトリーバーの作成次に、組み込みの Delta-Sync と Databricks が管理する埋め込みを活用し、シーンレベルのデータを Vector Search Index に読み込むことで、デプロイと使用を容易にします。つまり、スクリプト データベースが更新されると、対応する Vector Search Index もデータの更新に合わせて更新されます。下の図は、1本の映画(『恋のからさわぎ』)がシーンごとに分割されている例を示しています。ベクトル検索を使用すると、キーワードが完全に一致しなくても、エージェントは広告コンテンツの説明と意味的に類似したシーンを見つけることができます。可用性とガバナンスに優れたVector Searchインデックスの作成は簡単で、エンドポイント、ソーステーブル、埋め込みモデル、Unity Catalogのロケーションを定義するために数行のコードを記述するだけです。この例でのインデックス作成については、以下のコードを参照してください。from databricks.vector_search.client import VectorSearchClient from databricks.sdk import WorkspaceClient import databricks.sdk.service.catalog as c CATALOG = "media_advertising" SCHEMA = "contextual_advertising" # The table we'd like to index source_table_fullname = f"{CATALOG}.{SCHEMA}.movie_scripts_content"# Where we want to store our index in Unity Catalog vs_index_fullname = f"{CATALOG}.{SCHEMA}.movie_scripts_content_vs"VECTOR_SEARCH_ENDPOINT_NAME = "mads_demo_vs_endpoint" vsc = VectorSearchClient() # Create the endpoint - only necessary if the endpoint doesn't exist yet!if not endpoint_exists(vsc, VECTOR_SEARCH_ENDPOINT_NAME): vsc.create_endpoint(name=VECTOR_SEARCH_ENDPOINT_NAME, endpoint_type="STANDARD") wait_for_vs_endpoint_to_be_ready(vsc, VECTOR_SEARCH_ENDPOINT_NAME) print(f"Endpoint named {VECTOR_SEARCH_ENDPOINT_NAME} is ready.")# Create the vector search indx if not index_exists(vsc, VECTOR_SEARCH_ENDPOINT_NAME, vs_index_fullname): print(f"Creating index {vs_index_fullname} on endpoint {VECTOR_SEARCH_ENDPOINT_NAME}...") vsc.create_delta_sync_index( endpoint_name=VECTOR_SEARCH_ENDPOINT_NAME, index_name=vs_index_fullname, source_table_name=source_table_fullname, pipeline_type="TRIGGERED", primary_key="unique_movie_scene_id", embedding_source_column='scene_text', #The column containing our text embedding_model_endpoint_name='databricks-gte-large-en' #The embedding endpoint used to create the embeddings ) #Let's wait for the index to be ready and all our embeddings to be created and indexed wait_for_index_to_be_ready(vsc, VECTOR_SEARCH_ENDPOINT_NAME, vs_index_fullname) else: #Trigger a sync to update our vs content with the new data saved in the table wait_for_index_to_be_ready(vsc, VECTOR_SEARCH_ENDPOINT_NAME, vs_index_fullname) vsc.get_index(VECTOR_SEARCH_ENDPOINT_NAME, vs_index_fullname).sync()print(f"index {vs_index_fullname} on table {source_table_fullname} is ready")データの準備が整ったので、コンテンツ配置エージェントの構築に進むことができます。エージェント開発DatabricksにおけるAgentic AIの中核的な原則は、LLMに必要なツールを備えさせることで、エンタープライズデータに対して効果的に推論させ、データインテリジェンスを解放することです。LLMにエンドツーエンドのプロセス全体を実行させるのではなく、特定のタスクをツールや関数にオフロードし、LLMをインテリジェントなプロセスオーケストレーターとして機能させます。これにより、ユーザーのセマンティックな意図の理解や問題解決方法の推論といった、LLMの強みに特化して活用できます。私たちのアプリケーションでは、ユーザーのリクエストに基づいて関連シーンを効率的に検索する手段として、ベクトル検索インデックスを使用します。理論的にはLLM自身の知識ベースを使って関連シーンを取得することも可能ですが、Vector Searchインデックスのアプローチは、Unity Catalog内のガバナンスが効いたエンタープライズデータからの取得を保証するため、より実用的で効率的かつ安全です。エージェントは、ユーザーからの問い合わせに対していつ、どのように関数を呼び出すかを判断するために、関数定義内のコメントを使用することに注意してください。以下のコードは、Vector Searchインデックスを標準のUnity Catalog SQL関数にラップし、エージェントの推論プロセスにとってアクセスしやすいツールにする方法を示しています。%sql CREATE OR REPLACE FUNCTION media_advertising.contextual_advertising.search_movie_scripts ( -- The agent uses this comment to determine how to generate the query string parameter. query STRING COMMENT 'The query string for searching the movie script database' ) RETURNS TABLE -- The agent uses this comment to determine when to call this tool. This describes the types of documents and information contained within the index. COMMENT 'Search for relevant script chunks from the movie scripts database that matches the intent of the user request including elements of the scene that would make it a good fit' RETURN SELECT scene_text as page_content, map('title', TRY_CAST(title AS STRING), 'scene_number', TRY_CAST(scene_number AS STRING), 'search_score', TRY_CAST(search_score AS STRING)) as metadata FROM vector_search( -- Specify your Vector Search index name here index => 'media_advertising.contextual_advertising.movie_scripts_content_vs', query_text => query, query_type => "hybrid", num_results => 5 )エージェントを定義しましたが、次は何をすべきでしょうか?エージェント評価: MLflow を使用したエージェントの品質測定チームがエージェントアプリケーションを本番環境に導入する上で最大の障害の1つは、エージェントの品質と有効性を測定する能力です。本番環境へのデプロイでは、主観的な「雰囲気」に基づく評価は許容されません。チームには、アプリケーションが期待どおりに機能していることを確認し、反復的な改善を導くための定量的な方法が必要です。これらすべての疑問は、製品チームと開発チームを夜も眠れなくさせるでしょう。そこで登場するのが、DatabricksのMLflow 3.0によるエージェント評価です。MLflow 3.0は、モデルのトレース、評価、モニタリング、プロンプトレジストリなど、エンドツーエンドのエージェント開発ライフサイクルを管理するための堅牢なツールスイートを提供します。Databricks の LLM Judges の概要評価機能により、組み込みのLLMジャッジを活用して、事前に定義されたメトリクスに対する品質を測定できます。しかし、私たちのような特殊なシナリオでは、カスタマイズされた評価がしばしば必要になります。Databricksは、ユーザーが自然言語でジャッジ基準を提供し、Databricksがジャッジインフラを管理する自然言語「ガイドライン」の定義、ユーザーがプロンプトとカスタム評価基準を提供するプロンプトベースのジャッジ、単純なヒューリスティックまたはユーザーが完全に定義するLLMジャッジでありうるカスタムスコアラーまで、さまざまなレベルのカスタマイズをサポートしています。このユースケースでは、レスポンス形式に関するカスタムガイドラインと、シーンの関連性を評価するためのプロンプトベースのカスタムジャッジの両方を使用し、コントロールとスケーラビリティの強力なバランスを実現しています。# Custom Judge Example - Using own prompt & LLM w = WorkspaceClient() client = w.serving_endpoints.get_open_ai_client() # Define the prompts for the Judge LLM. judge_system_prompt = """ You are an impartial AI assistant responsible for evaluating the quality of a response generated by another AI model. Your evaluation should be based on the original user query and the AI's response. The context of the conversations is an user looking to find the most relevant script for their advertising scenario Provide a quality score as an integer from 1 to 5 (1=Poor, 2=Fair, 3=Good, 4=Very Good, 5=Excellent) that reflects the scene relevance. A 5 should reflect a perfect fit, a 1 should reflect a very weak or completely incorrect fit. Also, provide a brief rationale for your score in under 200 tokens. Be a very harsh critic and give out 5s sparingly. ONLY evaluate the final answer given, do not evaluate the options provided from the vector search tool call process Your output MUST ONLY be a single valid JSON object with two keys: "score" (an integer) and "rationale" (a string). The format must conform to this format Example: {"score": 4, "rationale": "The scene returned fit the user request well but not perfectly and the format was correct."} """ judge_user_prompt = """ Please evaluate the AI's Response below based on the Original User Query. Original User Query: ```{user_query}``` AI's Response: ```{llm_response_from_app}``` Provide your evaluation strictly as a JSON object with "score" and "rationale" keys. """ @scorer def script_fit_custom_metric(inputs: dict[str, Any], outputs: str) -> Feedback: user_query = inputs["input"][-1]["content"] # Call the Judge LLM using the OpenAI SDK client. judge_llm_response_obj = client.chat.completions.create( model="databricks-meta-llama-3-3-70b-instruct", # This example uses Databricks hosted Llama - you can customize as needed messages=[ {"role": "system", "content": judge_system_prompt}, {"role": "user", "content": judge_user_prompt.format(user_query=user_query, llm_response_from_app=outputs)}, ], max_tokens=200, # Max tokens for the judge's rationale temperature=0.0, # For more deterministic judging ) judge_llm_output_text = judge_llm_response_obj.choices[0].message.content judge_output_json = json.loads(judge_llm_output_text) score = judge_output_json.get("score") rationale = judge_output_json.get("rationale") # Parse the Judge LLM's JSON output. judge_eval_json = json.loads(judge_llm_output_text) parsed_score = int(judge_eval_json["score"]) parsed_rationale = judge_eval_json["rationale"] return Feedback( value=parsed_score, rationale=parsed_rationale, # Set the source of the assessment to indicate the LLM judge used to generate the feedback source=AssessmentSource( source_type=AssessmentSourceType.LLM_JUDGE, source_id="databricks-meta-llama-3-3-70b-instruct", ) ) # Use this custom judge, built-in judges, and a custom guideline to call Mlflow evaluate with mlflow.start_run(run_name="Movie-Eval-LLM"): eval_results = mlflow.genai.evaluate( data=eval_df, predict_fn=lambda input: AGENT.predict({"input": input}), scorers=[ RelevanceToQuery(), Safety(), # You can have any number of guidelines. Guidelines( name="ReturnFormat", guidelines= "Assess whether output contains Movie, Scene Number, Scene Description, Scene Justification" ), script_fit_custom_metric ], )合成データの生成エージェント評価におけるもう 1 つの一般的な課題は、エージェントの構築時に評価の基準となるユーザーリクエストのグラウンドトゥルース (正解データ) がないことです。私たちの場合、考えられる顧客リクエストの十分なセットがなかったため、構築したエージェントの有効性を測定するために合成データを生成する必要もありました。私たちは、組み込みの `generate_evals_df` 関数を活用してこのタスクを実行し、顧客のリクエストに合致すると予想されるサンプルを生成するように指示します。この合成生成データを評価ジョブの入力として使用し、データセットをブートストラップします。これにより、顧客に提供する前に、エージェントのパフォーマンスを明確かつ定量的に把握できます。# Define the synthetic data generation question_guidelines = """ # User personas - An account executive who is responsible for contextual content placement within shows - An enterprise executive who is responsible for the P&L and wants to optimize content placement in shows # Example questions - When could I insert a commercial for a light hearted comedy movie we want to promote for next summer? - When could I insert a commercial for a boys and girls club non-profit ad campaign? # Additional Guidelines - Question should be succinct with the goal of optimizing the relevance of advertising within a script. - The question should be generic, use the documents as a generalized framework to ask questions about movies. - NEVER reference specific scenes or characters. The full application will be asking questions across multiple scripts at once, not a specific show. """ agent_description = """ The Agent is a RAG chatbot that aims to recommend the optimal placement for advertising within scripts. The scripts are movies, but they are still intending to air commercials even though that is traditionally associated with TV. The Agent has access to a movie metadata and genre, and its task is to answer the user's questions by retrieving the relevant script chunks from the corpus and synthesizing a helpful, accurate response of where it makes sense to insert the content placement. End users will be using this agent across many scripts at once, so questions will be generic across the full script database, rather than about specific shows. """ eval_df = generate_evals_df(docs=sample_df, num_evals=50, agent_description=agent_description, question_guidelines=question_guidelines) eval_df['inputs'] = [{'input': row['messages']} for row in eval_df['inputs']] # For Responses Agent display(eval_df)MLflow 評価データセットを使用して、エージェントの品質を定量的に判断するための評価ジョブを実行できます。このケースでは、組み込みのジャッジ(関連性と安全性)、エージェントが正しい形式でデータを返したかを評価するカスタムガイドライン、そしてユーザーのクエリに対して返されたシーンの品質を1~5段階で評価するプロンプトベースのカスタムジャッジを組み合わせて使用します。幸いなことに、LLMジャッジのフィードバックに基づくと、私たちのエージェントは非常に優れたパフォーマンスを発揮しているようです。MLflow 3では、トレースをさらに深く掘り下げて、モデルがどのように機能しているかを理解し、各応答の背後にあるジャッジの論理的根拠を把握することもできます。このような観測レベルの詳細は、エッジケースを掘り下げ、エージェントの定義に対応する変更を加え、それらの変更がパフォーマンスにどのような影響を与えるかを確認するのに非常に役立ちます。この迅速なイテレーションと開発のループは、高品質なエージェントを構築する上で非常に強力です。私たちはもはや手探りで進むのではなく、アプリケーションのパフォーマンスについて明確な定量的ビューを得られるようになりました。Databricks レビューアプリジャッジとしてのLLMは非常に有用で、スケーラビリティのためにはしばしば必要ですが、本番環境への移行に自信を持ち、エージェントの全体的なパフォーマンスを向上させるためには、対象分野の専門家によるフィードバックが必要となることがよくあります。対象分野の専門家は、エージェントプロセスを開発するAIエンジニアではないことが多いため、フィードバックを収集し、それを私たちの製品やジャッジに反映させる方法が必要です。Agent Framework 経由でデプロイされたエージェントに付属のレビューアプリは、この機能を標準で提供します。対象分野の専門家は、エージェントと自由形式でやり取りすることも、エンジニアが専門家に特定の例を評価するよう依頼するカスタムのラベリングセッションを作成することもできます。これは、エージェントが困難なケースでどのように機能するかを観察したり、エンドユーザーのリクエストを非常によく表している可能性のある一連のテストケースで「ユニットテスト」として使用したりするのに非常に役立ちます。このフィードバックは、肯定的か否定的かにかかわらず、評価データセットに直接統合され、下流のファインチューニングや自動化されたジャッジの改善に使用できる「ゴールドスタンダード」を作成します。エージェント評価は確かに困難で時間がかかり、パートナーチーム間の連携と投資を必要とします。これには、通常の職務要件の範囲外と見なされる可能性のある、対象分野の専門家の時間も含まれます。Databricksでは、評価をエージェントアプリケーション構築の基盤と見なしており、組織がエージェント開発プロセスのコアコンポーネントとして評価の重要性を認識することが不可欠です。Databricks Model ServingとMCPを使用したエージェントのデプロイDatabricks でエージェントを構築すると、バッチとリアルタイムの両方のユースケースに柔軟なデプロイオプションが提供されます。このシナリオでは、Databricks の モデルサービングを活用し、REST API 経由でダウンストリームと統合できる、スケーラブルで安全なリアルタイムエンドポイントを生成します。簡単な例として、カスタムのModel Context Protocol(MCP)サーバーとしても機能するDatabricksアプリを介してこれを公開します。これにより、このエージェントをDatabricksの外部でツールとして活用できるようになります。 コア機能の拡張として、画像からテキストへの変換機能をDatabricksアプリに統合できます。以下は、LLMが受信した画像を解析し、テキストキャプションを生成し、希望するターゲットオーディエンスを含めたカスタムリクエストをコンテンツ配置エージェントに送信する例です。このケースでは、マルチエージェントアーキテクチャを活用して、ペット広告画像ジェネレーターを使用して広告画像をパーソナライズし、配置を依頼しました。 このエージェントをカスタムMCPサーバーでラップすることにより、広告主、パブリッシャー、メディアプランナーにとって、既存のアドテクエコシステムへの統合オプションが拡張されます。 まとめこのAIエージェントは、スケーラブルでリアルタイム、かつ深くコンテキストに応じた配置エンジンを提供することで、単純なキーワードを超えて、はるかに高い広告関連性を実現し、広告主とパブリッシャーの双方にとってキャンペーンのパフォーマンスを直接向上させ、広告の無駄を削減します。Databricks上のAIエージェントについてさらに学ぶ: Databricks Lakehouse Platformでの大規模言語モデルとAIエージェントの構築およびデプロイに関する専用リソースをご覧ください。専門家に相談する: これを貴社のビジネスに適用する準備はできていますか?Databricksが次世代の広告ソリューションの構築とスケーリングをどのように支援できるかについては、私たちのチームにお問い合わせください。