ITインフラの構築・運用フェーズで生成AIがどう役立つのかを解説する本連載。今回は、自然言語でログ調査ができるRAGアプリケーションの実装手順を紹介します。Fluent Bitによるログ収集からStreamlitによるチャット画面の実装まで、一連の開発プロセスをソースコードとともに解説します。
この記事は会員限定です。会員登録(無料)すると全てご覧いただけます。
ITインフラの構築・運用フェーズで生成AIがどう役立つのかを解説する本連載。前回は「膨大なログ調査も自然言語で」をテーマに、生成AIとRAG(Retrieval-Augmented Generation)を活用し、従来は多大な時間と労力を要していたログ調査を大幅に効率化するためのアーキテクチャを紹介しました。
今回は、そのアーキテクチャをベースに、どのようにアプリケーションへ落とし込んでいくのかに焦点を当てます。設計のポイントや実装手順を解説しながら、生成AIを運用現場で「使える形」にするための実践的な開発プロセスを紹介します。
本稿で紹介するアプリケーションのソースコードはGitHubリポジトリで公開しています。
第3回の記事でご紹介したアーキテクチャを簡単におさらいします。これをベースに、以下の図に示す各コンポーネントを順に実装していきます。
まず、各サーバに「Fluent Bit」を導入し、ログを収集してログ収集APIに送信します(1)。ログ収集APIでは、ログの本文と日時を分離し、本文はベクトル化して「ChromaDB」に保存、日時はメタデータとして管理します(2)。
インフラ運用担当者は、Streamlitで作成したチャット画面から自然言語で質問を入力します(3)。質問内容をベクトル化してデータベースを検索し、関連するログを取得します(4)。
最後に、取得したログを基にLLM(大規模言語モデル)が回答を生成し(5)、その結果をチャット画面に返します(6)。
これにより、運用担当者は大量のログを直接調べることなく、必要な情報を素早く把握できます。
今回紹介するアプリケーションのディレクトリ構成は以下の通りです。Fluent Bitの設定ファイルやログ収集用のスクリプト、チャットUI用のスクリプトなどが含まれています。
├── fluent-bit │ ├── apps.log # サンプルログファイル │ ├── docker-compose.yml # Fluent Bit用のDocker Composeファイル │ ├── fluent-bit.conf # Fluent Bitのメイン設定ファイル │ └── parsers.conf # Fluent Bitの解析設定ファイル ├── ingest.py # ログ収集APIの実装 ├── requirements.txt # 必要なPythonパッケージ一覧 └── ui.py # チャットUIの実装
各サーバからログを収集するためにFluent Bitを導入します。Fluent Bitは軽量なログ収集エージェントであり、多様な入力ソースからログを収集し、さまざまな出力先に転送できます。ここでは、Fluent Bitは公式のDockerイメージから起動することとして、fluent-bit/docker-compose.ymlファイルの主要な設定内容を紹介します。
services:
fluent-bit:
image: fluent/fluent-bit:4.2
container_name: fluent-bit
- ./fluent-bit.conf:/fluent-bit/etc/fluent-bit.conf:ro
- ./parsers.conf:/fluent-bit/etc/parsers.conf:ro
1行目左側の./fluent-bit.confはホスト(Composeファイルのあるディレクトリ)上のファイルです。1行目右側の/fluent-bit/etc/fluent-bit.confはコンテナ内のFluent Bitが参照する標準パスです。2行目も同様に、parsers.confをコンテナ内の標準パスにマウントしています。:roはread-only(読み取り専用)でマウントする指定です。
この構成にすることで、設定をホスト側で編集→コンテナ再起動で反映ができ、試行錯誤がしやすくなります。またroにすることで、コンテナ側から設定を誤って書き換える事故も防げます。
- ./apps.log:/var/log/apps.log:ro
Fluent Bitの入力側でtailプラグインを使って/var/log/apps.logを監視する想定のため、ホストのapps.logをそのパスにマウントしています。
extra_hosts: - "host.docker.internal:host-gateway"
今回の構成ではFluent Bitの出力先が「ログ収集API(FastAPI)=ホスト側で起動しているHTTPサーバ」になります。その場合、Fluent Bitコンテナからホストへアクセスできる必要があります。
host.docker.internalは「コンテナからホストへ到達するための特別なホスト名」としてよく使われます。ただしLinux環境ではhost.docker.internalが自動で解決されない場合があり、その対策としてextra_hostsを使って明示的に名前解決できるようにしています。host-gatewayはDockerが持つホストゲートウェイの特殊指定で、Docker Compose側でホストのゲートウェイIPを割り当ててくれます。
この設定があることで、Fluent Bit側の出力設定で下記のように書いた場合でも、コンテナ内で正しく解決され、ホスト上のlocalhost相当へHTTP送信できるようになります。
Host host.docker.internal
Fluent Bitのメイン設定ファイルがfluent-bit/fluent-bit.confです。今回の構成に必要な主要設定内容を紹介します。
[INPUT]
name tail
path /var/log/apps.log
parser simplelog
Tag app.logs
このセクションでは、Fluent Bitがどのログを、どの形式で読み取るかを指定しています。
[OUTPUT]
Name http
Match app.logs
Host host.docker.internal
Port 8000
URI /ingest
このOUTPUTでは、ログをHTTPリクエストとしてログ収集APIに送信する設定をしています。
次に、Fluent Bitの解析用パーサ設定ファイルであるfluent-bit/parsers.confの主要な内容を紹介します。このファイルの役割は、ログのフォーマットに基づいてログを解析し、必要なフィールド(ここでは日時と本文)を抽出することです。
[PARSER]
Name simplelog
Format regex
Regex ^\[(?<time>[^\]]+)\]\s(?<log>.*)$
Time_Key time
Time_Format %Y-%m-%d %H:%M:%S %z
[2024-01-13 11:00:00 +0000] Jan 13 11:00:00 server1 kernel: Out of memory: Kill process 1234 (myapp) score 1000 or sacrifice child
この結果、1行のログは以下のように分解されます。
これにより、ログに含まれるタイムスタンプを正確に内部時刻として扱えるようになります。
ログ収集APIやチャットUIを実装するために必要なPythonパッケージをインストールするためのrequirements.txtファイルの内容は以下の通りです。
fastapi==0.121.3 uvicorn==0.38.0 chromadb==1.3.5 streamlit==1.51.0 google-genai==1.52.0
それぞれのライブラリについて説明します。
ログ収集APIは、Fluent Bitから送信されたログを受け取り、ベクトル化してChromaDBに保存する役割を担います。以下に、ingest.pyファイルの主要な実装内容を示します。
chroma_db_client = chromadb.PersistentClient() collection = chroma_db_client.get_or_create_collection(name="logs")
ChromaDBの永続クライアント(PersistentClient)を作成し、logsというコレクションを用意します。ここにログ本文をドキュメントとして保存します。これを使用することで、データベースの内容がディスクに保存され、アプリケーションの再起動後もデータが保持されます。
app = FastAPI()
FastAPIアプリケーションのインスタンスを作成します。これにより、APIエンドポイントを定義してリクエストを処理できるようになります。
@app.post("/ingest")
async def receive_log(request: Request):
data = await request.json()
/ingestエンドポイントを定義し、POSTリクエストを受け取ります。Fluent Bitから送信されたログデータはJSON形式で受け取り、request.json()メソッドを使用してパースします。
受け取るログの形式は以下のようになっています。
[
{
"date": 1705132800,
"log": "Jan 13 11:00:00 server1 kernel: Out of memory: Kill process 1234 (myapp) score 1000 or sacrifice child"
},
{
"date": 1705132860,
"log": "Jan 13 11:01:00 server1 myapp[1234]: Application started successfully"
}
]
dateフィールドがログのタイムスタンプ(UNIX timestamp 形式)、logフィールドがログ本文です。
for entry in data:
ts = entry["date"] # ログのタイムスタンプ(UNIX timestamp 形式)を取得
log = entry["log"] # ログ本文を取得
Fluent Bitから送信された各ログエントリについて、タイムスタンプとログ本文を抽出します。タイムスタンプはUNIX timestamp形式で受け取ります。
metadatas = []
metadatas.append({"timestamp": ts})
ログのタイムスタンプはChromaDBのメタデータとして保存します。これにより、後段の検索処理で日時条件による絞り込みが可能になります。
unique_id = str(uuid.uuid4())
ChromaDBに登録するための一意なIDをUUIDで生成します。
# ChromaDBにログを追加
collection.add(
documents=[log],
ids=[unique_id],
metadatas=metadatas,
)
ログ本文をdocumentsとしてChromaDBに登録します。本文はベクトル化され、ベクトル検索の対象となります。
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000, log_level="debug")
このスクリプトを直接実行した場合に、Uvicornを使ってFastAPIアプリを起動します。host="0.0.0.0"とすることで、Dockerコンテナなど外部からのアクセスも可能になります。
チャットアプリケーションは、Streamlitを使って構築し、ユーザーからの自然言語の質問を受け付け、関連するログを検索して回答を生成します。以下に、ui.pyファイルの主要な実装内容を示します。
# Azure OpenAI クライアントの初期化 llm_client = genai.Client()
Geminiを利用するためのクライアントです。今回の例ではGoogleの生成AIサービスであるGeminiを使用していますが、他のLLMサービスを利用する場合は適宜変更してください。
chroma_db_client = chromadb.PersistentClient() collection = chroma_db_client.get_or_create_collection(name="logs")
ChromaDBの永続クライアントを作成し、logsコレクションを取得します。ログ収集APIで保存したログデータを検索するために使用します。
def extract_time_range_from_query(query: str):
system_prompt = """
あなたは日時抽出専用のアシスタントである。
ユーザーの日本語の質問文から、「検索したい時間範囲(開始と終了)」だけを抽出し、
次のJSON形式で出力せよ。
・・・以下略・・・
- JSON 以外の文字は一切出力してはならない。
""".strip()
ユーザーの日本語クエリから検索対象の日時範囲を抽出する関数です。ここでは、正規表現ではなく生成AIに日時抽出専用の役割を与える設計にしています。
ここで定義しているシステムプロンプトでは、ユーザーの質問文から「検索したい時間範囲(開始と終了)」だけを抽出し、JSON形式で出力するよう指示しています。これにより、複雑な自然言語の質問からも正確に日時範囲を特定できます。そしてここで抽出した日時範囲を、ChromaDBの検索時にメタデータフィルターとして利用します。このフィルターの適用により、指定された時間範囲内のログのみを対象に検索をするため、不要なログが混入するのを防げます。
resp = llm_client.models.generate_content(
model="gemini-2.5-flash",
config=types.GenerateContentConfig(
system_instruction=system_prompt),
contents=query
)
content = resp.text
Geminiのテキスト生成機能を使って、ユーザーの質問文から日時範囲を抽出します。system_instructionに先ほど定義したシステムプロンプトを渡し、contentsにユーザーの質問文を指定します。そして、変数contentには、Geminiが生成した日時範囲のJSON文字列が格納されます。
# 抽出した JSON から日時をパースして UNIX timestamp に変換
try:
data = json.loads(content)
start_str = data.get("start")
end_str = data.get("end")
start_ts = datetime.fromisoformat(start_str).timestamp() if start_str else None
end_ts = datetime.fromisoformat(end_str).timestamp() if end_str else None
return start_ts, end_ts
except Exception:
# 日付抽出に失敗したため、時間フィルタなしで検索
return None, None
抽出したJSON文字列をパースし、開始日時と終了日時をISO 8601形式からUNIX timestampに変換します。変換に成功した場合は、開始と終了のタイムスタンプをタプルで返します。もしパースに失敗した場合は、時間フィルターなしで検索するために(None, None)を返します。
def search_logs(query: str) -> str:
start_ts, end_ts = extract_time_range_from_query(query)
ログから関連するエントリを検索する関数です。まず、ユーザーの質問文から日時範囲を抽出します。先ほど定義したextract_time_range_from_query関数を呼び出し、開始と終了のタイムスタンプを取得します。
where_filter = None
if start_ts is not None and end_ts is not None:
where_filter = {
"$and": [
{"timestamp": {"$gte": start_ts}},
{"timestamp": {"$lte": end_ts}}
]
}
抽出した日時範囲に基づいて、ChromaDBのメタデータフィルターを作成します。開始と終了の両方が存在する場合、$and条件でtimestampが指定範囲内にあるログのみを対象とするフィルターを構築します。
results = collection.query(
query_texts=[query],
n_results=5, # 上位5件のみ
where=where_filter, # 日時フィルタを適用(取れなければ None のまま)
)
retrieved_logs = "\n".join(results["documents"][0]) if results["documents"] else ""
ChromaDBにベクトル検索を実行し、関連するログエントリを取得します。query_textsにユーザーの質問文を指定し、n_resultsで上位5件のみを取得します。whereパラメーターに先ほど作成した日時フィルターを適用します。検索結果からログ本文を抽出し、改行で結合して1つの文字列として返します。
prompt = f"""
以下はサーバーログの一部である。このログをもとに質問に答えよ。
ログ:
{retrieved_logs}
質問:
{query}
答えは簡潔にまとめてください。
"""
ユーザーの質問に対して回答を生成するためのプロンプトを作成します。取得したログエントリとユーザーの質問文を含め、ログをもとに質問に答えるよう指示しています。
response = llm_client.models.generate_content(
model="gemini-2.5-flash", contents=prompt
)
return response.text
Geminiのテキスト生成機能を使って、作成したプロンプトに基づいて回答を生成します。生成された回答テキストを返します。
if "history" not in st.session_state:
st.session_state["history"] = []
ここからは、Streamlitを使って画面を構築するためのコードが続きます。まず、チャット履歴を初期化しています。st.session_stateは、Streamlitのセッション状態を管理するためのオブジェクトで、セッション間でデータを共有するために使用されます。"history"というキーがst.session_stateに存在しない場合は、空のリストを初期化しています。
for message in st.session_state.history:
with st.chat_message(message["role"]):
st.write(message["content"])
チャット履歴を表示するためのコードです。st.session_state.historyに格納されているチャット履歴を取り出して、st.chat_messageを使って表示しています。message["role"]にはユーザーかチャットbotのどちらであるかを示すroleが格納され、message["content"]にはメッセージの内容が格納されています。
if prompt := st.chat_input("質問を入力してください"):
ユーザーが質問を入力した際の処理を記述しています。st.chat_inputを使って、ユーザーが質問を入力するための入力フィールドを表示しています。ユーザーが質問を入力すると、その内容がpromptに格納されます。
with st.chat_message("user"):
st.write(prompt)
ユーザーが入力した質問を表示するためのコードです。st.chat_messageを使って、ユーザーが質問を入力したことを示すuserのroleを指定しています。st.writeを使って、ユーザーが入力した質問を表示しています。
st.session_state.history.append({"role": "user", "content": prompt})
ユーザーの質問をチャット履歴に追加するために、st.session_state.history.append()を使って、st.session_state.historyにユーザーが入力した質問を追加しています。roleにはuserを、contentにはユーザーが入力した質問を指定しています。
response = search_logs(prompt)
ユーザーの質問に対して回答を生成するために、先ほど定義したsearch_logs関数を呼び出しています。promptにはユーザーが入力した質問が渡され、その質問に基づいて関連するログを検索し、回答を生成します。生成された回答はresponseに格納されます。
with st.chat_message("assistant"):
st.write(response)
生成した回答を表示するためのコードです。st.chat_messageを使って、回答を表示するためのassistantのロールを指定しています。st.writeを使って、生成した回答を表示しています。
st.session_state.history.append({"role": "assistant", "content": response})
生成した回答をチャット履歴に追加するためのコードです。st.session_state.history.append()を使って、st.session_state.historyに生成した回答を追加しています。roleにはassistantを、contentには生成した回答を指定しています。
最後に、今回紹介したアプリケーションの動作確認手順を説明します。
まず、GitHubリポジトリをクローンします。ターミナルで以下のコマンドを実行してください。
$ git clone https://github.com/noriyukitakei/log-qa-rag.git $ cd log-qa-rag
まず、Fluent Bitを起動します。fluent-bitディレクトリに移動し、以下のコマンドを実行してください。
$ cd fluent-bit $ docker compose up -d
Gemini APIキーを取得し、環境変数として設定します。リポジトリのルートディレクトリに.envファイルを作成し、以下の内容を追加してください。
GEMINI_API_KEY="取得したAPIキーをここに貼り付けてください"
Gemini APIキーは「Google AI Studio」から取得できます。以下のURLを参照してください。
必要なPythonパッケージをインストールします。ターミナルで以下のコマンドを実行してください。
$ pip install -r requirements.txt
次に、ログ収集APIを起動します。以下のコマンドを実行してください。
$ python ingest.py
apps.logに以下のようなログエントリを追加します。Fluent Bitが自動的にログを検知し、ログ収集APIに送信します。
[2026-01-13 11:45:22 +0000] kernel: Out of memory: Kill process 4455 (mysql) score 500 due to large temporary table usage [2026-01-13 11:45:23 +0000] dbserver: ERROR: MySQL killed due to memory pressure after large query [2026-01-13 11:45:23 +0000] dbserver: ERROR: Service restarted after OOM kill [2026-01-14 16:55:12 +0000] systemd: Started Web Service. [2026-01-14 17:05:42 +0000] kernel: Memory pressure detected. [2026-01-14 17:06:03 +0000] kernel: Out of memory: Kill process 2345 (node) score 900 or sacrifice child [2026-01-14 17:06:04 +0000] kernel: Killed process 2345 (node) total-vm:123456kB, anon-rss:98765kB, file-rss:1234kB [2026-01-14 17:06:05 +0000] webapp: ERROR: Service restarted after OOM kill [2026-01-14 17:12:17 +0000] kernel: Memory cgroup out of memory: Kill process 6789 (python) score 750 [2026-01-14 18:22:33 +0000] systemd: The system is running low on memory. [2026-01-15 02:44:11 +0000] kernel: Out of memory: Kill process 9876 (backup) score 700 due to huge file buffering [2026-01-15 02:44:12 +0000] backup: WARN: Backup aborted due to insufficient memory
チャットアプリケーションを起動します。以下のコマンドを実行してください。
$ streamlit run ui.py
ターミナルに以下のように表示されたら、ブラウザで指定されたURL(通常はhttp://localhost:8501)にアクセスしてください。
You can now view your Streamlit app in your browser. Local URL: http://localhost:8501 Network URL: http://192.168.x.x:8501
質問を入力するフィールドに、以下のような質問を入力して動作を確認してください。
2026年1月13日の11時頃に起きたOOM-Killerの原因を教えて。
すると、チャットアプリケーションが関連するログを検索し、回答を生成して表示します。例えば、以下のような回答が得られることが期待されます(図2参照)。
大規模なクエリによりMySQLが大量の一時テーブルを使用し、メモリ圧迫が発生したためです。
今回は、第3回でご紹介したRAGアーキテクチャをベースに、アプリケーション開発手順を具体的に解説しました。Fluent Bitを使ったログ収集から、FastAPIによるログ収集APIの実装、Streamlitを活用したチャットUIの構築まで、一連の流れを通じて、生成AIを活用したログ調査アプリケーションの開発プロセスを紹介しました。
これにより、ITインフラ運用担当者は膨大なログデータから必要な情報を迅速に抽出できるようになり、運用効率の大幅な向上が期待できます。
本連載は今回が最終回となります。「生成AI」というと、昨今ではバイブコーディング(Vibe Coding)や仕様駆動開発など、コード生成部分に注目が集まりがちです。しかし、その活用範囲はアプリケーション開発にとどまらず、まさに無限の可能性を秘めています。
本連載ではITインフラ構築・運用の現場に焦点を当ててきましたが、生成AIを活用することで、これまで人の手で行われてきたさまざまな業務プロセスを効率化・自動化することが可能です。今後、生成AIの進化とともに、私たちの働き方や業務の在り方も大きく変わっていくことでしょう。
私は、生成AIはインターネット登場以来の大きなパラダイムシフトをもたらす技術であると考えています。これを活用するかどうかによって、今後の業務効率や競争力に大きな差が生まれる可能性があります。本連載でご紹介した内容を参考にしながら、生成AIを活用した業務改善のアイデアをぜひ検討してみてください。
サイオステクノロジーに所属し、AzureやAI領域でのシステムインテグレーションや製品開発に取り組むエンジニア。Microsoft MVPとして認定され、YouTubeなどのメディアで「わかりみ深い」Azureの情報を日々発信している他、執筆活動(書籍『世界一やさしいRAG構築入門』)を通じて知識を広く共有している。
Copyright © ITmedia, Inc. All Rights Reserved.
編集部からのお知らせ