現在、数多くの AI 企業がエージェント型ビジネスインテリジェンスシステムを導入しており、複雑な SQL クエリや手動によるスプレッドシート解析を必要とせずに、データを即座に分析し包括的なインサイトを提供しています。例えば、RAG 搭載の営業分析システムは、「前四半期のベストセラー製品は何ですか？」といった質問に数時間ではなく数秒で答えることができます。
RAG 搭載の営業分析システムは、AI エージェントが自律的に推論し、意思決定を行い、外部ツールを連携させることで、データエコシステム全体にわたる自然言語クエリを可能にします。ユーザーはデータベーススキーマやスプレッドシートの数式を学ぶ必要はなく、エージェントがクエリを分解し、最適なデータソースを判断し、対象を絞った分析を実行し、適切な引用付きでデータに基づいた包括的な回答を提供します。
たとえば、異なる LLM やカスタムドキュメント形式を使用したり、エージェントの分析・動作方法を調整して特定のビジネス出力を生成することもできます。このチュートリアルでは、Novita AI の統一 LLM API、LangChain のエージェントフレームワーク、高度なドキュメント処理機能を使って、独自の RAG 搭載営業分析システムを構築する方法を学びます。クエリを最適なデータソースに自動的にルーティングし、実用的なビジネスインサイトを提供するシステムを作成します。
RAG 搭載営業分析システムとは？
RAG 搭載営業分析システムは、Retrieval-Augmented Generation（検索拡張生成）を使用して、大規模言語モデル（LLM）のパワーと実際のビジネスデータにアクセスできる検索レイヤーを組み合わせることで、小売販売に関する質問に対して正確で根拠のある回答を提供します。
このシステムは通常、以下の要素を含みます。
- AI エージェントがクエリを分析し、構造化データ分析（SQL）、ドキュメント分析、またはその両方のどれが必要かを判断します。
- エージェントはクエリを最も適切なデータソース（定量分析には SQL データベース、定性インサイトにはドキュメントストア）にインテリジェントにルーティングします。
- エージェントは、CSV ファイル分析用の Pandas エージェント、データベースクエリ用の SQL エージェント、ドキュメント検索用のベクターストアなど、さまざまなツールを使用して情報を収集します。これらのツールがないと、エージェントはワークフローで重要な処理を実行できません。
- エージェントは、数値データには統計分析、テキストドキュメントにはセマンティック検索など、専門技術を使って収集した情報を処理します。
- データ分析後、エージェントは要約、計算、適切な引用を含む構造化された回答を作成します。
必要なツール
構築パートに入る前に、必要なツールをセットアップしましょう。
Novita AI
RAG 搭載分析システムを構築するには、強力な大規模言語モデル（LLM）と埋め込みモデルへのアクセスが必要です。Novita AI は、手頃な価格で高性能な API を提供し、単一の統一インターフェースを通じて最新の LLM、埋め込みモデルなどへのアクセスを提供します。
LangChain
LangChain は、LLM を使ったアプリケーション構築のためのオープンソースフレームワークです。LangChain を使用すると、ステップバイステップで推論し、ツールを使用し、API と対話するエージェント型ワークフローを作成できます。営業分析システムでは、LangChain を利用して分析プロセスを構造化し、SQL エージェントやドキュメントプロセッサなどのツールを採用し、すべてのデータを構造化インサイトに統合します。
Streamlit
インタラクティブな Web インターフェースを構築するために Streamlit を使用します。高速プロトタイピングに最適で、最小限のコードでプロフェッショナルな見た目の UI を作成できるため、このチュートリアルを簡単に追うことができます。
FAISS
ベクターベースのドキュメント検索には、FAISS が高速な類似検索機能を提供し、ユーザークエリに基づいて関連するドキュメントチャンクを素早く見つけることができます。
SQLAlchemy & PyMySQL
これらは SQL データベース操作と MySQL 接続を処理し、システムが構造化ビジネスデータに直接クエリできるようにします。
pandas
CSV データの操作と分析に不可欠です。システムは pandas エージェントを使用して、表形式データに対して複雑な計算を実行します。
システムアーキテクチャの概要
RAG システムは、ユーザークエリをマルチステップのワークフローでインテリジェントに処理し、質問ごとに最適なデータソースと処理方法を自動的に決定します。ユーザーがデータの場所を知る必要はなく、システムが背後で最適なアプローチを判断します。
クエリ処理
エージェントによる処理の最初のステップはクエリ処理です。この場合、エージェントはユーザーが入力したクエリを検証し、そのクエリのコンテキストに合わせたカスタマイズされた処理計画を立てます。そして、質問が構造化データ、ドキュメント分析、またはその両方のどれに対して分析されるべきかを判断します。
データソース選択
ユーザーが質問を送信すると、システムはシンプルなアプローチを採用します。まず SQL データベースに問い合わせて問題を解決しようとします。データソース選択は、構造化データが定量的な質問に最も正確な回答を提供することが多いため、理にかなっています。SQL エージェントはクエリを迅速に実行し、適切な計算を含む正確な数値を返すことができます。
ドキュメント分析
SQL エージェントが空の結果を返したり、関連情報を見つけられなかったりすると、システムは自動的にドキュメント分析に切り替えます。ドキュメント処理のパスはファイルタイプによって異なります。CSV および Excel ファイルは複雑なデータ分析のために Pandas DataFrame に読み込まれ、PDF、Word ファイル、テキストドキュメントはセマンティック検索のためにチャンク化され、FAISS を使用してインデックス化されます。
ソース管理
提供された情報が信頼できる権威に遡れるように、すべてのソースを適切に引用することが不可欠です。ワークフローのこのステップは、最終レポートに提示された主張やデータを検証できるようにすることで、ユーザーとの信頼構築に役立ちます。
応答合成
すべての収集情報を結びつけるために、エージェントは分析されたすべてのデータソースからのインサイトを統合し、一貫性のある応答を形成します。ワークフローのこのフェーズがなければ、出力はまとまりのないレポートになる可能性があります。
RAG 搭載分析ワークフローの実装
ここまでで、RAG 搭載営業分析システムの概要を理解しました。次は実装です。開始するために、いくつかの主要なステップを実行します。
インストールとセットアップ
進める前に、RAG 分析システムの完全なファイル構造を確立しましょう。プロジェクトディレクトリに以下のファイルを作成する必要があります。
| rag-analytics-system/ ├── .env # 環境変数 ├── requirements.txt # Python 依存関係 ├── query_processor.py # メインの QueryProcessor クラス ├── main.py # Streamlit インターフェース |
依存関係のセットアップ
まず、requirements.txt ファイルに必要なすべての依存関係を記述します。
| langchain==0.3.26 langchain-openai==0.3.25 python-dotenv==1.1.1 SQLAlchemy==2.0.41 pandas==2.3.0 PyPDF2==3.0.1 faiss-cpu==1.11.0 PyMySQL==1.1.1 cryptography==45.0.4 langchain-experimental==0.3.4 streamlit==1.46.0 openpyxl==3.1.5 python-docx==1.2.0 |
次に、Novita AI にログインします。ログイン後、Manage API Keys ページに移動し、Add New Key ボタンをクリックして Key Name を入力します。
Novita AI はサインアップ後に無料クレジットを提供するため、構築や実験を開始する前にクレジットを購入する心配はありません。作成した Novita API キーを .env ファイルに追加します。
| NOVITA_API_KEY=“your_novita_api_key_here” |
次に、プロジェクトに必要なすべての依存関係をインストールします。
| # 仮想環境の作成 python -m venv rag_analytics_env source rag_analytics_env/bin/activate # Windows の場合: rag_analytics_env\Scripts\activate # 依存関係のインストール pip install -r requirements.txt |
クエリプロセッサの構築
RAG 分析ワークフロー全体は、QueryProcessor というモジュールクラスにカプセル化されています。このクラスは、クエリのルーティング、SQL 分析の実行、ドキュメントの処理、結果の統合、完全な分析ループの実行を管理します。
QueryProcessor は、分析システムの中央オーケストレーターとして機能します。データタイプごとに別々のツールを構築するのではなく、この統一アプローチにより、SQL データベース、CSV ファイル、非構造化ドキュメントのいずれであっても、クエリを最も適切な分析方法にインテリジェントにルーティングできます。
まず、query_processor.py ファイルに必要なライブラリをインポートします。これらのインポートにより、言語モデル統合、データベース接続、ドキュメント処理、ベクターストレージに必要なすべてのものが提供されます。
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain.agents import AgentExecutor, create_react_agent
from langchain_community.agent_toolkits.sql.base import create_sql_agent
from langchain_experimental.agents.agent_toolkits import create_csv_agent, create_pandas_dataframe_agent
from langchain_community.agent_toolkits.sql.toolkit import SQLDatabaseToolkit
from langchain.agents import AgentType
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import FAISS
from langchain_community.utilities import SQLDatabase
from langchain.prompts import PromptTemplate
from langchain_community.tools import ReadFileTool
from langchain.schema import Document
import os
from dotenv import load_dotenv
from sqlalchemy.engine import Engine
import glob
import pandas as pd
load_dotenv()
次に、Novita API 経由の LLM とさまざまなデータ処理ツールにアクセスできる分析エージェントを初期化します。初期化メソッドは、クエリ理解のための言語モデル、ドキュメント類似検索のための埋め込み、大規模ドキュメントを効率的に処理するためのテキスト分割など、必要なすべてのコアコンポーネントをセットアップします。
class QueryProcessor:
def __init__(self, documents_folder: str, sql_engine: Engine):
self.documents_folder = documents_folder
self.sql_engine = sql_engine
self.llm = ChatOpenAI(
model="google/gemma-3-27b-it",
temperature=0,
openai_api_key=os.getenv("NOVITA_API_KEY"),
openai_api_base="<https://api.novita.ai/v3/openai>",
default_headers={
"X-Model-Provider": "google"
}
)
self.embeddings = OpenAIEmbeddings(
model="baai/bge-m3",
openai_api_key=os.getenv("NOVITA_API_KEY"),
openai_api_base="<https://api.novita.ai/v3/openai>",
default_headers={
"X-Model-Provider": "baai"
}
)
self.text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200,
length_function=len,
)
self.sql_agent = None
self._prepare_data_sources()
コンストラクタはいくつかの主要コンポーネントを初期化します。LLM には温度 0 を使用して、創造的なバリエーションではなく一貫性のある事実に基づく応答を保証します。テキストスプリッターは、大きなドキュメントを処理する際にコンテキストの連続性を維持するために重複チャンクで構成されています。最後に、_prepare_data_sources() メソッドを呼び出して特殊エージェントをセットアップします。
マルチソースデータ統合の作成
次のステップは、さまざまなデータ処理エージェントを初期化するメソッドの作成です。各エージェントは、特定のデータタイプと分析タスクの処理を専門としており、さまざまな種類の質問に対して異なる専門家がいるようなものです。
def _prepare_data_sources(self):
"""SQL とドキュメントの両方のデータソースを準備"""
# SQL エージェントの準備
self._prepare_sql_agent()
# ドキュメントエージェントの準備
self._prepare_document_agent()
def _prepare_sql_agent(self):
"""SQL エージェントの初期化"""
# SQLAlchemy Engine を LangChain SQLDatabase に変換
db = SQLDatabase(self.sql_engine)
toolkit = SQLDatabaseToolkit(db=db, llm=self.llm)
self.sql_agent = create_sql_agent(
llm=self.llm,
toolkit=toolkit,
agent_type=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
verbose=True
)
def _prepare_document_agent(self):
"""ReadFileTool とベクターストアを使用してドキュメントエージェントを初期化"""
# サポートされているすべてのファイルを取得
supported_files = []
for ext in ['*.txt', '*.pdf', '*.docx', '*.xlsx', '*.xls', '*.csv']:
supported_files.extend(glob.glob(os.path.join(self.documents_folder, ext)))
print(f"\
{self.documents_folder} 内に {len(supported_files)} 個のサポートファイルが見つかりました")
if supported_files:
# ReadFileTool の作成
read_file_tool = ReadFileTool()
# ツールリストの作成
tools = [read_file_tool]
# react エージェント用のプロンプトテンプレートを作成
prompt = PromptTemplate.from_template("""
あなたはビジネスドキュメントに関する質問に答えることができる役立つアシスタントです。
以下のツールにアクセスできます。
{tools}
次の形式を使用してください。
Question: 回答すべき入力質問
Thought: 何をすべきかを常に考える
Action: 実行するアクション。{tool_names} のいずれか
Action Input: アクションへの入力
Observation: アクションの結果
... (この Thought/Action/Action Input/Observation は N 回繰り返すことができます)
Thought: 最終的な答えがわかった
Final Answer: 元の入力質問に対する最終的な答え
始めます。
Question: {input}
Thought: 関連ドキュメントを読んで答えを見つけるべき
{agent_scratchpad}
""")
agent = create_react_agent(self.llm, tools, prompt)
self.document_agent = AgentExecutor.from_agent_and_tools(
agent=agent,
tools=tools,
verbose=True
)
print("\
ドキュメントエージェントが正常に初期化されました")
else:
print("\
指定されたフォルダにドキュメントが見つかりませんでした。")
_prepare_sql_agent() メソッドは、データベースクエリ用の特殊エージェントを作成します。ZERO_SHOT_REACT_DESCRIPTION エージェントタイプを使用することで、エージェントは具体的な例がなくても SQL クエリについて推論できるため、異なるデータベーススキーマに適応できます。_prepare_document_agent() メソッドは、サポートされているファイルタイプをスキャンし、さまざまなドキュメント形式を読み取れる柔軟なエージェントを作成します。ReAct（Reasoning and Acting）プロンプトテンプレートは、人間のアナリストがドキュメント分析に取り組む方法と同様に、エージェントを構造化された思考プロセスに導きます。
ドキュメント処理とベクターストレージ
次に、ワークフローがさまざまなドキュメントタイプを処理する方法を処理する必要があります。このセクションでは、pandas エージェントを使用して CSV/Excel ファイルを処理するメソッドと、ベクターストレージを使用して非構造化ドキュメントを処理するメソッドを作成します。ここでの重要な洞察は、データタイプごとに異なる処理戦略が必要であるということです。
CSV_PROMPT_PREFIX = """
重要: 実際のデータが読み込まれた 'df' という pandas DataFrame を操作しています。
サンプルデータを作成したり、データをでっち上げたりしないでください。利用可能な実際の DataFrame 'df' のみを使用してください。
まず、DataFrame を以下のように探索します。
1. pandas 表示オプションを設定してすべての列を表示: pd.set_option('display.max_columns', None)
2. DataFrame の形状を確認: print(df.shape)
3. 列名を取得: print(df.columns.tolist())
4. データ型を確認: print(df.dtypes)
5. 最初の数行を表示: print(df.head())
6. その後、DataFrame 内の実際のデータを使用して質問に答えます。
"""
CSV_PROMPT_SUFFIX = """
- **重要**: DataFrame 内の実際のデータのみを使用してください。サンプルデータや架空のデータを作成しないでください。
- **常に** 最終回答を出す前に、別の方法で結果を検証してください。
- 次に、2つの方法の回答を振り返り、元の質問に正しく答えているか自問してください。
- 確信が持てない場合は、別の方法を試してください。
- 4桁以上の数字はカンマでフォーマットしてください。
- 試した方法が同じ結果にならない場合は、同じ結果になる2つの方法が得られるまで反映して再試行してください。
- それでも一貫した結果にたどり着けない場合は、答えがわからないと言ってください。
- 正しい答えが確信できる場合は、Markdown を使用して美しく徹底的な応答を作成してください。
- **答えをでっち上げたり、事前知識を使用したりせず、実行した計算結果のみを使用してください**。
- **常に**、最終回答の一部として、どのように答えにたどり着いたかを "説明:" で始まるセクションで説明してください。
- 説明では、最終回答に使用した列名に言及してください。
- 関連する DataFrame 操作とその結果を表示して作業を示してください。
"""
def _process_document_query(self, query: str) -> str:
"""CSV/Excel エージェント（表形式データ）と LLM（非構造化データ）を使用してドキュメントクエリを処理"""
try:
print(f"\
クエリを処理中: {query}")
# サポートされているすべてのファイルを取得
supported_files = []
for ext in ['*.txt', '*.pdf', '*.docx', '*.xlsx', '*.xls', '*.csv']:
supported_files.extend(glob.glob(os.path.join(self.documents_folder, ext)))
print(f"{self.documents_folder} 内に {len(supported_files)} 個のサポートファイルが見つかりました")
if not supported_files:
return "検索するドキュメントが見つかりませんでした。"
# 最初に CSV/Excel ファイルをチェック
csv_files = [f for f in supported_files if f.endswith('.csv')]
excel_files = [f for f in supported_files if f.endswith(('.xlsx', '.xls'))]
if csv_files:
csv_file = csv_files[0]
print(f"\
CSV ファイルを処理中: {csv_file}")
try:
# まず pandas DataFrame エージェントで試行
df = pd.read_csv(csv_file)
print(f"CSV が正常に読み込まれました。行数: {len(df)}、列: {df.columns.tolist()}")
# 改善された設定でエージェントを作成
print("pandas DataFrame エージェントを作成中...")
agent = create_pandas_dataframe_agent(
self.llm,
df,
verbose=True,
include_df_in_prompt=False, # 大きな DataFrame でのトークン制限を回避
allow_dangerous_code=True,
max_iterations=10,
handle_parsing_errors=True
)
# カスタムプロンプトでクエリを処理
print(f"エージェントでクエリを処理中: {query}")
# エージェントが実際の DataFrame を使用することを保証する改善されたプロンプト
prompt = f"""
以下の列を持つ {len(df)} 行の pandas DataFrame 'df' にアクセスできます: {df.columns.tolist()}
データの最初の数行:
{df.head().to_string()}
データ型:
{df.dtypes.to_string()}
{self.CSV_PROMPT_PREFIX}
Question: {query}
{self.CSV_PROMPT_SUFFIX}
"""
response = agent.invoke({"input": prompt})
print(f"エージェントの応答: {response}")
return response['output']
except Exception as e:
print(f"pandas DataFrame エージェントのエラー: {str(e)}")
print("代替 CSV エージェントを試行中...")
# CSV エージェントにフォールバック
try:
agent = create_csv_agent(
self.llm,
csv_file,
verbose=True,
allow_dangerous_code=True
)
enhanced_query = f"""
{self.CSV_PROMPT_PREFIX}
Question: {query}
{self.CSV_PROMPT_SUFFIX}
"""
response = agent.invoke({"input": enhanced_query})
return response['output']
except Exception as e2:
print(f"CSV ファイル処理エラー: {str(e2)}")
import traceback
print(f"完全なトレースバック: {traceback.format_exc()}")
return f"CSV ファイル処理エラー: {str(e2)}"
elif excel_files:
excel_file = excel_files[0]
print(f"\
Excel ファイルを処理中: {excel_file}")
try:
df = pd.read_excel(excel_file)
print(f"Excel が正常に読み込まれました。行数: {len(df)}、列: {df.columns.tolist()}")
# 改善された設定でエージェントを作成
print("pandas DataFrame エージェントを作成中...")
agent = create_pandas_dataframe_agent(
self.llm,
df,
verbose=True,
include_df_in_prompt=False, # 大きな DataFrame でのトークン制限を回避
allow_dangerous_code=True,
max_iterations=10,
handle_parsing_errors=True
)
# カスタムプロンプトでクエリを処理
print(f"エージェントでクエリを処理中: {query}")
# エージェントが実際の DataFrame を使用することを保証する改善されたプロンプト
prompt = f"""
以下の列を持つ {len(df)} 行の pandas DataFrame 'df' にアクセスできます: {df.columns.tolist()}
データの最初の数行:
{df.head().to_string()}
データ型:
{df.dtypes.to_string()}
{self.CSV_PROMPT_PREFIX}
Question: {query}
{self.CSV_PROMPT_SUFFIX}
"""
response = agent.invoke({"input": prompt})
print(f"エージェントの応答: {response}")
return response['output']
except Exception as e:
print(f"Excel ファイル処理エラー: {str(e)}")
import traceback
print(f"完全なトレースバック: {traceback.format_exc()}")
return f"Excel ファイル処理エラー: {str(e)}"
# 非構造化テキストファイルの場合
print("\
非構造化テキストファイルを処理中...")
all_content = []
for file_path in supported_files:
try:
if file_path.endswith('.txt'):
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
all_content.append(Document(page_content=content, metadata={"source": file_path}))
except Exception as e:
print(f"ファイル {file_path} の読み込みエラー: {str(e)}")
continue
if not all_content:
return "どのドキュメントも読み込めませんでした。"
# ベクターストアを使用して非構造化テキストを処理
print("ベクターストアでテキストを処理中...")
chunks = self.text_splitter.split_documents(all_content)
vector_store = FAISS.from_documents(chunks, self.embeddings)
relevant_docs = vector_store.similarity_search(query, k=3)
context = "\
\
".join([doc.page_content for doc in relevant_docs])
print(f"生成されたコンテキストの長さ: {len(context)}")
response = self.llm.invoke(
f"""以下のコンテキストに基づいて、質問に答えてください: {query}
\
Context:\
{context}\
\
Answer:"""
)
return response.content
except Exception as e:
print(f"ドキュメントクエリ処理エラー: {str(e)}")
import traceback
print(f"完全なトレースバック: {traceback.format_exc()}")
return f"ドキュメントクエリ処理エラー: {str(e)}"
CSV 処理セクションには、正確な分析を保証するための詳細なプロンプトエンジニアリングが含まれています。プレフィックスとサフィックスのプロンプトは、エージェントがデータをでっち上げたり、誤った結果を提供したりするのを防ぐために重要です。エージェントが複数の方法を試す検証手順は、注意深いアナリストが計算を再確認するのと同様に、精度を確保するのに役立ちます。pandas DataFrame エージェントを CSV エージェントよりも優先するのは、より堅牢なデータ処理と大規模データセットでのパフォーマンスが優れているためです。フォールバックメカニズムにより、1 つのアプローチが失敗した場合に、データを処理する代替手段が確保されます。それでは、_process_document_query メソッドを完成させて、Excel ファイルと非構造化ドキュメントを処理する部分を追加します。
elif excel_files:
excel_file = excel_files[0]
print(f"\
Excel ファイルを処理中: {excel_file}")
try:
df = pd.read_excel(excel_file)
print(f"Excel が正常に読み込まれました。行数: {len(df)}、列: {df.columns.tolist()}")
# 改善された設定でエージェントを作成
print("pandas DataFrame エージェントを作成中...")
agent = create_pandas_dataframe_agent(
self.llm,
df,
verbose=True,
include_df_in_prompt=False, # 大きな DataFrame でのトークン制限を回避
allow_dangerous_code=True,
max_iterations=10,
handle_parsing_errors=True
)
# カスタムプロンプトでクエリを処理
print(f"エージェントでクエリを処理中: {query}")
# エージェントが実際の DataFrame を使用することを保証する改善されたプロンプト
prompt = f"""
以下の列を持つ {len(df)} 行の pandas DataFrame 'df' にアクセスできます: {df.columns.tolist()}
データの最初の数行:
{df.head().to_string()}
データ型:
{df.dtypes.to_string()}
{self.CSV_PROMPT_PREFIX}
Question: {query}
{self.CSV_PROMPT_SUFFIX}
"""
response = agent.invoke({"input": prompt})
print(f"エージェントの応答: {response}")
return response['output']
except Exception as e:
print(f"Excel ファイル処理エラー: {str(e)}")
import traceback
print(f"完全なトレースバック: {traceback.format_exc()}")
return f"Excel ファイル処理エラー: {str(e)}"
# 非構造化テキストファイルの場合
print("\
非構造化テキストファイルを処理中...")
all_content = []
for file_path in supported_files:
try:
if file_path.endswith('.txt'):
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
all_content.append(Document(page_content=content, metadata={"source": file_path}))
except Exception as e:
print(f"ファイル {file_path} の読み込みエラー: {str(e)}")
continue
if not all_content:
return "どのドキュメントも読み込めませんでした。"
# ベクターストアを使用して非構造化テキストを処理
print("ベクターストアでテキストを処理中...")
chunks = self.text_splitter.split_documents(all_content)
vector_store = FAISS.from_documents(chunks, self.embeddings)
relevant_docs = vector_store.similarity_search(query, k=3)
context = "\
\
".join([doc.page_content for doc in relevant_docs])
print(f"生成されたコンテキストの長さ: {len(context)}")
response = self.llm.invoke(
f"""以下のコンテキストに基づいて、質問に答えてください: {query}
\
Context:\
{context}\
\
Answer:"""
)
return response.content
except Exception as e:
print(f"ドキュメントクエリ処理エラー: {str(e)}")
import traceback
print(f"完全なトレースバック: {traceback.format_exc()}")
return f"ドキュメントクエリ処理エラー: {str(e)}"
SQL データベース統合
ドキュメント処理を処理したので、SQL 分析とドキュメント分析のどちらを使用するかを決定するインテリジェントなクエリルーティングシステムを作成する必要があります。これがシステムの中核となるインテリジェンスです。つまり、特定の質問に対する回答が含まれる可能性が最も高いデータソースを判断します。
def process_query(self, query: str) -> str:
"""
エージェントを使用してクエリを処理し、SQL とドキュメントのどちらを使用するかをインテリジェントに判断
"""
# 最初に SQL エージェントを試行
try:
print("\
SQL エージェントを試行中...")
sql_result = self.sql_agent.run(query)
no_answer_phrases = [
"no results", "i don't know", "unknown", "not sure", "cannot answer", "don't have", "no data", "n/a"
]
if sql_result and not any(phrase in sql_result.lower() for phrase in no_answer_phrases) and sql_result.strip():
return f"SQL データベースから: {sql_result}"
else:
print("SQL エージェントが回答できませんでした。ドキュメントを試行中...")
except Exception as e:
print(f"SQL エージェントエラー: {str(e)}")
print("ドキュメントにフォールバック中...")
# SQL エージェントが失敗した場合、または結果を返さなかった場合、ドキュメント処理を試行
try:
print("\
ドキュメントを処理中...")
doc_result = self._process_document_query(query)
if doc_result:
return f"ドキュメントから: {doc_result}"
else:
print("ドキュメント処理が結果を返しませんでした")
except Exception as e:
print(f"ドキュメント処理エラー: {str(e)}")
return "SQL データベースまたはドキュメントに関連情報が見つかりませんでした。"
クエリルーティングロジックは優先順位システムに従います。SQL データベースは、通常、ビジネス上の質問に迅速かつ正確に回答できる構造化された定量的データを含んでいるため、最初に試行されます。SQL エージェントが不明確または否定的な応答を返した場合（no_answer_phrases リストで検出）、システムは自動的にドキュメント処理にフォールバックします。このアプローチは、人間のアナリストが最初に構造化データソースを確認し、データベースに必要な情報がない場合にドキュメントやレポートに頼る方法を模倣しています。
Streamlit インターフェースの作成
それでは、分析システムをビジネスユーザーがアクセスできるようにするユーザーインターフェースを構築しましょう。Streamlit インターフェースは、直感的なチャットベースのエクスペリエンスを提供し、技術的な複雑さを隠しながら、強力な分析機能を提供します。以下のコードスニペットを main.py ファイルに追加します。
import streamlit as st
from query_processor import QueryProcessor
import os
from dotenv import load_dotenv
from sqlalchemy import create_engine
# 環境変数の読み込み
load_dotenv()
#st.text_input
# セッション状態の初期化
if 'processor' not in st.session_state:
st.session_state.processor = None
if 'messages' not in st.session_state:
st.session_state.messages = []
# ページ設定
st.set_page_config(
page_title="Document Analysis Chatbot",
page_icon="🤖",
layout="wide"
)
# ボタンとタイトルのスタイリング用カスタム CSS
st.markdown("""
<style>
.stButton > button {
background-color: #23D57C;
color: white;
border: none;
border-radius: 8px;
padding: 0.5rem 1rem;
font-weight: 600;
transition: all 0.3s ease;
}
.stButton > button:hover {
background-color: #1fb36b;
box-shadow: 0 4px 8px rgba(35, 213, 124, 0.3);
transform: translateY(-2px);
}
.stButton > button:active {
background-color: #1a9960;
transform: translateY(0px);
}
h1 {
color: #23D57C !important;
font-weight: 700;
}
</style>
""", unsafe_allow_html=True)
# タイトルと説明
st.title("Document Analysis Chatbot using Novita")
この初期設定により、カスタムスタイルを備えたプロフェッショナルな外観のインターフェースが作成されます。セッション状態管理により、ユーザーがインターフェースを操作する際に会話履歴を失ったり、データソースを再初期化する必要がなくなります。カスタム CSS は視覚的なフィードバックを提供し、一貫したブランドエクスペリエンスを維持します。
最終システム統合
データソースの初期化とチャット機能を使用して Streamlit インターフェースを完成させましょう。この最終セクションでは、すべてのコンポーネントをユーザーフレンドリーなアプリケーションに統合し、ビジネスアナリストが技術的な専門知識を必要とせずに利用できるようにします。
# データソースが初期化されているか確認
if st.session_state.processor is None:
# データソース設定を中央に配置
st.markdown("<br><br>", unsafe_allow_html=True)
# 中央揃えの列を作成
col1, col2, col3 = st.columns([1, 2, 1])
with col2:
st.subheader("🚀 開始する")
st.write("ドキュメントとのチャットを開始するには、データソースを初期化してください")
# SQL データベース設定（非表示）
db_user = "root"
db_password = "1234cisco"
db_host = "localhost"
db_name = "retail_sales_db"
# ドキュメントフォルダ設定
st.write("**ドキュメントフォルダのパス:**")
documents_folder = st.text_input(
"Documents Folder Path",
placeholder="ドキュメントフォルダのパスを入力してください（例: docs）",
label_visibility="collapsed"
)
st.markdown("<br>", unsafe_allow_html=True)
# ボタンを中央に配置
button_col1, button_col2, button_col3 = st.columns([1, 1, 1])
with button_col2:
if st.button("データソースを初期化", use_container_width=True):
try:
# ドキュメントフォルダのパスを検証して解決
if not documents_folder:
st.error("ドキュメントフォルダのパスを指定してください")
else:
# 絶対パスに変換
abs_documents_folder = os.path.abspath(documents_folder)
if not os.path.exists(abs_documents_folder):
st.error(f"ドキュメントフォルダが見つかりません: {abs_documents_folder}")
elif not os.path.isdir(abs_documents_folder):
st.error(f"パスがディレクトリではありません: {abs_documents_folder}")
else:
# ダミーの SQL エンジンでクエリプロセッサを初期化
# SQL エンジンを作成
connection_string = f"mysql+pymysql://{db_user}:{db_password}@{db_host}/{db_name}"
sql_engine = create_engine(connection_string)
st.session_state.processor = QueryProcessor(abs_documents_folder, sql_engine)
st.success("データソースが正常に初期化されました！")
st.rerun()
except Exception as e:
st.error(f"データソースの初期化エラー: {str(e)}")
else:
# サイドバーにデータソースステータスを表示
with st.sidebar:
st.header("📊 データソース")
st.success("✅ データソースが初期化されました")
if st.button("データソースをリセット"):
st.session_state.processor = None
st.session_state.messages = []
st.rerun()
# メインチャットインターフェース
st.header("チャットインターフェース")
# チャットメッセージを表示
for message in st.session_state.messages:
with st.chat_message(message["role"]):
st.write(message["content"])
# チャット入力
if prompt := st.chat_input("ドキュメントについて質問する"):
# ユーザーメッセージをチャット履歴に追加
st.session_state.messages.append({"role": "user", "content": prompt})
with st.chat_message("user"):
st.write(prompt)
# クエリを処理
try:
response = st.session_state.processor.process_query(prompt)
# アシスタントの応答をチャット履歴に追加
st.session_state.messages.append({"role": "assistant", "content": response})
with st.chat_message("assistant"):
st.write(response)
except Exception as e:
st.error(f"クエリ処理エラー: {str(e)}")
# メッセージがある場合のみクリアチャットボタンを追加
if st.session_state.messages:
if st.button("チャットをクリア"):
st.session_state.messages = []
st.rerun()
初期化プロセスには、一般的なユーザーエラーを防ぐための堅牢なエラーハンドリングとパス検証が含まれています。サイドバーは明確なステータス情報を提供し、チャットインターフェースはユーザーが最新の AI アシスタントに期待する使い慣れたパターンに従っています。システムは会話履歴を維持し、ユーザーが以前の質問に基づいて構築し、自然な分析ワークフローを作成できるようにします。クリアチャット機能により、必要に応じて新しいセッションを開始できます。
分析システムの実行
RAG 搭載営業分析システムを構築したので、実際のビジネスクエリでどのように機能するかテストしてみましょう。
まず、data_generator.py ファイルを作成し、こちらの Python スクリプトからコードをコピーして、システムに提供するサンプルデータを生成し、次のコマンドでスクリプトを実行します。
| python data_generator.py |
これにより、sample_documents フォルダが作成され、以下のファイルが含まれます。
- large_sales_dataset.csv - 10,000 件の売上記録
- business_strategy_2024.txt - 戦略的ビジネスドキュメント
- sales_meeting_notes.txt - 会議メモとアクションアイテム
次に、Streamlit アプリケーションを起動します。
| streamlit run main.py |
アプリケーションがブラウザで開きます（http://localhost:8501）。
初期化後、ドキュメントのパスとして sample_documents を入力し、さまざまなビジネス質問をテストできます。
| # システムをテストするサンプルクエリ topic_1 = “2024 年のエレクトロニクスの総売上高はいくらですか？” topic_2 = “カスタマーエクスペリエンスに関する主要な戦略的イニシアチブは何ですか？” topic_3 = “最も高い総売上を上げた営業担当者は誰ですか？その金額はいくらですか？” |
結論
このチュートリアルを通じて、最新の AI がビジネスインテリジェンスワークフローをどのように変革できるかを示す RAG 搭載分析システムを構築しました。単一のインテリジェントインターフェースの下で複数のデータ処理アプローチを組み合わせることで、アナリストが日々直面するあらゆるビジネス質問に対応できるツールを作成しました。このシステムの主な強みは次のとおりです。
- 各質問に最適なデータソースを自動的に判断するインテリジェントクエリルーティング
- SQL データベース、CSV ファイル、Excel スプレッドシート、テキストドキュメントを処理するマルチフォーマットサポート
- フォールバックメカニズムと明確なエラーメッセージを提供する堅牢なエラーハンドリング
- 技術的専門知識を必要としないチャットベースの対話
- 信頼性の高い結果を保証する組み込みの検証と確認
このアプリケーションを拡張して、追加のソース、より高度な分析手法、カスタムビジネスロジックを組み込むことができます。この知識を活かして、ぜひ独自のエージェントを Novita AI で構築したり、既存のプロジェクトに統合してみてください。
Novita AI は、AI の野望を実現するオールインワンのクラウドプラットフォームです。統合 API、サーバーレス、GPU インスタンス — 必要なコスト効率の高いツールを提供します。インフラストラクチャを排除し、無料で開始して、AI のビジョンを現実にしましょう。
