生成 AI は研究室の外へ飛び出し、今やビジネス現場の常識を塗り替えています。SAPは全速力でその波に乗っています。このブログシリーズでは、SAP AI Core の既定モデルを最速で呼び出し、実務で使える AI エージェントへ拡張する“秒速ハンズオン”をお届けします。
お知らせ
You can find the English version here.
📖 本シリーズで学べること
SAP AI Core 上でカスタム AI Agent を “秒速” で動かす方法LangChain・Google 検索ツール・RAGを使った実装AI Agent を REST API 化し、SAPUI5/Fiori の UI に載せ、Cloud Foundryにデプロイする手順
学習時間
各章は 10–15 分 で読める＆手を動かせるを予定しています。
🗺️ 連載ロードマップ
Part 0 プロローグPart 1 環境構築: SAP AI CoreとAI LaunchpadPart 2 LangChain でチャットモデルを構築Part 3 Agentツール: Google 検索を統合Part 4 RAG 基礎 ①: HANA Cloud VectorEngineと埋め込み処理Part 5 RAG 基礎 ②: Retriever Toolの構築Part 6 AI Agent を REST API 化 [現在のブログ]Part 7 SAPUI5 でチャットUI を構築Part 8 CloudFoundry にデプロイ
注記
続編は順次公開予定です。
この記事がお役に立ったら、ぜひ Kudos を押していただけると励みになります。 「ここをもっと詳しく知りたい」「別のテーマも取り上げてほしい」など、ご要望があればお気軽にコメントください！
AI Agent を REST API 化
1｜はじめに
本章では、これまで Notebook 上で開発してきた AI Agent を REST API として公開し、後で Cloud Foundry（CF） にデプロイできる形にまとめます。API化 することで、以下のメリットがあります:
再利用性：CLI や別のアプリケーション、SAP UI5 のフロントエンドなど、あらゆるクライアントから HTTPリクエスト で呼び出せるスケーラビリティ：プロセスを Webサーバ に常駐させることで複数リクエストに対応できる運用性：CF上での自動スケールやログ管理、認証制御など、クラウド環境の標準的な仕組みを活用できる
本章では、FastAPI を用いて簡単な WebAPI を実装し、LangChain ベースの AI Agent をエンドポイント化します。
2 | 事前準備
BTP サブアカウントSAP AI Core インスタンスSAP AI LaunchpadのサブスクリプションPython 3.13環境 & pipVSCodeやBASなどのIDE
Trial 環境の注意
Trial の HANA Cloud インスタンスは 毎晩自動停止 されます。日をまたぐ作業の場合は翌日インスタンスを再起動してください。
3 | フォルダ構成とライブラリ準備
CF へデプロイすることを見据え、プロジェクト用のフォルダを新しく用意します。以降説明するコードや設定ファイルは、全てこのフォルダ内で管理します。
# フォルダ構成
my-ai-agent-api/
├── main.py
├── requirements.txt
├── .env
└── （デプロイに必要なファイルを追加予定）
CF にデプロイすることを見据えて requirements.txt で、必要な Python パッケージを明示しておきます。CF ではアプリケーション起動時、requirements.txt にしたがってライブラリを自動インストールします。今回新しく記載した fastapi や uvicorn、gunicorn は Webサーバ として必須のライブラリです。
以下にようにライブラリを列挙します。
# generative-ai-hub-sdk の公式ドキュメントで指定されている依存関係群
generative-ai-hub-sdk
ai_core_sdk>=2.5.7
pydantic==2.9.2
openai>=1.56.0
langchain~=0.3.0
langgraph==0.3.30
langchain-community~=0.3.0
langchain-openai>=0.2.14
langchain-google-vertexai==2.0.1
langchain-google-community==2.0.7
langchain-aws==0.2.9
google-cloud-aiplatform==1.61.0
boto3==1.35.76
# SAP HANA Cloud VectorSearch 用
hdbcli==2.24.24
langchain-hana==0.1.0
# .env ファイルを読み込むため
python-dotenv==1.1.0
# FastAPI／ASGI サーバー
fastapi==0.109.0
gunicorn
uvicorn[standard]==0.27.0
ローカルで動作確認するために、新たに仮想環境を用意しましょう。 Part 5 までに使用した仮想環境を流用しても構いませんが、ここでは念のため新規で作成します。
cd my-ai-agent-api
python -m venv .venv
source .venv/bin/activate
pip install –upgrade pip
pip install -r requirements.txt
同様に、環境変数ファイル（.env）も用意しましょう。
# SAP AI Core 認証情報
AICORE_CLIENT_ID = “<YOUR_AICORE_CLIENT_ID>”
AICORE_CLIENT_SECRET = “<YOUR_AICORE_CLIENT_SECRET>”
AICORE_AUTH_URL = “https://<your-region>.authentication.<your-region>.hana.ondemand.com”
AICORE_BASE_URL = “https://api.ai.prod.<your-region>.aws.ml.hana.ondemand.com”
AICORE_RESOURCE_GROUP = “<YOUR_RESOURCE_GROUP>”
DEVELOPMENT_ID = “<YOUR_OPENAI_DEPLOYMENT_ID>”
# HANA Cloud 接続情報
HANA_DB_ADDRESS = “<YOUR_HANA_DB_HOST>.hana.trial-<region>.hanacloud.ondemand.com”
HANA_DB_PORT = <YOUR_HANA_DB_PORT> # 例: 443
HANA_DB_USER = “<YOUR_HANA_DB_USER>”
HANA_DB_PASSWORD = “<YOUR_HANA_DB_PASSWORD>”
# Google カスタム検索 API
GOOGLE_CSE_ID = “<YOUR_GOOGLE_CSE_ID>”
GOOGLE_API_KEY = “<YOUR_GOOGLE_API_KEY>”
これで、必要なパッケージのインストールと環境変数の用意が完了、API 開発の準備が整いました！
4 | main.py の作成
ここからは main.py ファイルを作成し、FastAPI を使って AI Agent を呼び出すエンドポイントを実装していきます。Part5 で使っていたノートブックのセルをほとんどそのまま組み合わせて作成しますが、AI Agent がどのように考え、何のアクションを取ったのかチャット画面で確認できるように、いくつか修正を加えます。
具体的には、以下の流れで構造化されたレスポンスを返すようにします:
環境変数の読み込み（.env）FastAPI アプリケーションの初期化LangChain / Google 検索・HANA Cloud Retriever Tool の作成AI Agent の初期化（AgentType を CHAT_ZERO_SHOT_REACT_DESCRIPTION）ログから Thought / Action / Observation を取り出す関数の定義AI Agent からのレスポンスを変換/agent/chat エンドポイントの実装import os
import re
import uvicorn
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from dotenv import load_dotenv
# LangChain関連のインポート
from gen_ai_hub.proxy.langchain.openai import ChatOpenAI
from langchain.tools import Tool
from langchain.tools.retriever import create_retriever_tool
from langchain_google_community import GoogleSearchAPIWrapper
from langchain.agents import initialize_agent, AgentType
# HANA関連のインポート
from langchain_hana import HanaInternalEmbeddings
from langchain_hana import HanaDB
from hdbcli import dbapi
# 環境変数を読み込む
load_dotenv(verbose=True)
# FastAPIアプリケーションを初期化
app = FastAPI()
# CORS設定
app.add_middleware(
CORSMiddleware,
allow_origins=[“*”],
allow_credentials=True,
allow_methods=[“*”],
allow_headers=[“*”],
)
# リクエストモデル
class QueryRequest(BaseModel):
query: str
# ChatOpenAIを初期化
chat_llm = ChatOpenAI(deployment_id=os.getenv(“DEVELOPMENT_ID”))
# Google検索ツールを設定
search = GoogleSearchAPIWrapper(k=5)
google_tool = Tool.from_function(
name=”google_search”,
description=”Search Google and return the first results”,
func=search.run
)
# embeddingsを初期化
embeddings = HanaInternalEmbeddings(
internal_embedding_model_id=”SAP_NEB.20240715″
)
# HANA Cloudと接続
connection = dbapi.connect(
address=os.getenv(“HANA_DB_ADDRESS”),
port=os.getenv(“HANA_DB_PORT”),
user=os.getenv(“HANA_DB_USER”),
password=os.getenv(“HANA_DB_PASSWORD”),
sslValidateCertificate=False,
autocommit=True,
)
# データベースの立ち上げ
db = HanaDB(
embedding=embeddings,
connection=connection,
table_name=”TEST_TABLE”
)
# retrieverツールを設定
retriever_tool = create_retriever_tool(
retriever=db.as_retriever(),
name=”hana_vectorengine”,
description=(
“Use this tool to search internal SAP documents stored “
“in HANA Cloud Vector Engine when the user asks company-specific questions.”
),
)
# エージェントを初期化
agent = initialize_agent(
tools=[google_tool, retriever_tool],
llm=chat_llm,
agent=AgentType.CHAT_ZERO_SHOT_REACT_DESCRIPTION,
verbose=True,
return_intermediate_steps=True,
)
def extract_thought_from_log(log):
if not log or not isinstance(log, str):
return “”
# 正規表現を使用して “Thought” 部分を抽出
patterns = [
r’Thought:s*(.*?)(?:nAction:|$)’,
r’^(.*?)(?:nAction:|$)’,
# 日本語対応
r’思考:s*(.*?)(?:nアクション:|$)’,
r’^(.*?)(?:nアクション:|$)’
]
# 複数のパターンを試行
for pattern in patterns:
match = re.search(pattern, log, re.DOTALL | re.IGNORECASE)
if match:
thought = match.group(1).strip()
# 空文字列や改行のみの場合はスキップ
if thought and not thought.isspace():
return thought
# マッチしない場合は元のログを返す
return log.strip()
def transform_response(raw_response):
if not isinstance(raw_response, dict):
return {“output”: “”, “intermediate_steps”: []}
output = raw_response.get(‘output’, ”)
intermediate_steps = raw_response.get(‘intermediate_steps’, [])
if not isinstance(intermediate_steps, list):
intermediate_steps = []
structured_steps = []
for idx, step in enumerate(intermediate_steps, 1):
try:
# ステップが辞書形式の場合（既に構造化されている場合）
if isinstance(step, dict):
structured_steps.append(step)
continue
# ステップがタプル形式の場合
if isinstance(step, tuple) and len(step) == 2:
action, observation = step
# ログからThoughtを抽出
thought = “”
if hasattr(action, ‘log’) and action.log:
thought = extract_thought_from_log(action.log)
# 構造化されたステップを作成
structured_step = {
“step_no”: idx,
“thought”: thought,
“action”: getattr(action, ‘tool’, ‘unknown’),
“action_input”: getattr(action, ‘tool_input’, {}),
“observation”: observation if observation is not None else “”
}
structured_steps.append(structured_step)
except Exception:
# エラーが発生した場合はスキップ
continue
return {
“output”: output,
“intermediate_steps”: structured_steps
}
@app.post(“/agent/chat”)
async def chat(request: QueryRequest):
“””エージェントにクエリを実行し、構造化されたレスポンスを返す”””
# エージェントを実行
raw_response = agent.invoke({“input”: request.query})
# レスポンスを変換
structured_response = transform_response(raw_response)
return structured_response
if __name__ == “__main__”:
uvicorn.run(app, host=”0.0.0.0″, port=8000)
このセルで何をしているか
AgentType.CHAT_ZERO_SHOT_REACT_DESCRIPTION: この設定を指定することで、AI Agent が「Thought → Action → Observation」という形で内部ステップを出力します。（ return_intermediate_steps=True を指定しないと response に含まれないので注意）extract_thought_from_log(): AI Agent の出力ログ（英語・日本語）から「Thought」の部分を抽出する、自作の正規表現関数です。transform_response(): (action, observation)形式のタプルを { step_no, thought, action, action_input, observation } という辞書形式に変換します。
AgentType.CHAT_ZERO_SHOT_REACT_DESCRIPTION の注意
このシリーズでは、チャットUI 上に AI Agent の思考プロセスを表示するため、AgentType.CHAT_ZERO_SHOT_REACT_DESCRIPTION を採用します。ただし、この AgentType は “ツールを使わない質問” に対してエラーを返すことがあります。現在、より良い対処方法を検討中ですので、暫定的にこの点にご留意ください。
5 | 動作を確認する
それでは、用意した Python ファイルを実行し、作成した API が意図通りに動作することを確認しましょう。最初にローカル環境で FastAPI アプリケーションを起動し、API ドキュメント画面を見ながら実際にエンドポイントへリクエストを飛ばしてみます。その後、帰ってきた JSON が「output」と「intermediate_steps」の構造を満たしているかチェックします。
仮想環境に入っていることを確認しつつ、ターミナルで以下のコマンドを実行し、FastAPI アプリケーションをローカルサーバとして立ち上げます。
gunicorn -w 1 -k uvicorn.workers.UvicornWorker main:app –bind 0.0.0.0:${PORT:-8000}
起動が完了すると、「Application startup complete」というログが表示されます。Business Application Studio（BAS）で実行している場合は以下のような画面が確認できます。
ポップアップで出てきた右下の「Open in a New Tab」を押し、URLに/docsを追加して下さい。 FastAPI が自動生成した Swagger UI が表示され、/agent/chat のエンドポイントに対してどのようなリクエストを投げられるか、インタラクティブに確認できます。
『POST – /agent/chat』のアコーディオンメニューを開き、「Try it out」をクリックします。Request ボディに JSON 形式で { “query”: “任意の質問文” } を入力し、送信ボタンを押すと API が即座に応答します。
画面の最下部のように HTTPステータスコード”200″ が返り、Response Body が「output」と「intermediate_steps」の構造を満たしていることを確認できたら成功です！
6｜チャレンジ – テキストファイルアップロード機能を追加しよう
ユーザが新しく用意したテキストファイルをアップロードし、HANA Cloud VectorEngine に格納されるエンドポイントを追加しましょう。この機能を用意しておくことで、ユーザはチャット画面からドキュメントを AI Agent に渡すことができます。
以下のコードを、main.py のインポート部やモデル定義、エンドポイント定義の適切な位置にまとめて挿入してみましょう。具体的な行番号やセクションはお任せしますが、インポートやモデルのそれぞれに対応するように挿入するか、あるいはファイルアップロード専用のブロックとしてまとめると良いかもしれませんね。
# ファイルアップロード用のインポート
import shutil
import tempfile
from fastapi import File, UploadFile, HTTPException
from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import CharacterTextSplitter
# レスポンスモデル
class UploadResponse(BaseModel):
message: str
filename: str
chunks_created: int
@app.post(“/agent/upload”)
async def upload_file(file: UploadFile = File(…)):
“””テキストファイルをアップロードしてHANAに埋め込みデータとして格納”””
# ファイル形式のチェック
if not file.filename.endswith(‘.txt’):
raise HTTPException(
status_code=400,
detail=”txt形式のファイルのみアップロード可能です。”
)
try:
# 一時ファイルとして保存
with tempfile.NamedTemporaryFile(delete=False, suffix=’.txt’) as tmp_file:
shutil.copyfileobj(file.file, tmp_file)
tmp_file_path = tmp_file.name
# テキストファイルを読み込み
text_documents = TextLoader(
tmp_file_path
).load()
# テキストを適切なチャンクに分割
text_splitter = CharacterTextSplitter(
chunk_size = 50,
chunk_overlap = 0,
)
text_chunks = text_splitter.split_documents(text_documents)
# HANAベクターデータベースに格納
db.add_documents(text_chunks)
# 一時ファイルを削除
os.unlink(tmp_file_path)
return UploadResponse(
message=”ファイルが正常にアップロードされ、埋め込み処理が完了しました。”,
filename=file.filename,
chunks_created=len(text_chunks)
)
except Exception as e:
raise HTTPException(
status_code=500,
detail=f”ファイル処理中にエラーが発生しました: {str(e)}”
)
ポイントは、FastAPI で受け取った UploadFile をそのまま扱うのではなく、必ず tempfile.NamedTemporaryFile にコピーしてパスを取得し、そのパスを TextLoader に渡す点です。こうすることで、ファイルの中身を安全にローカルディスクに保存しつつ、その後チャンク分割してHANA Cloud に送ることができます。
コードを追加したら、先ほどと同様の手順で動作確認を行います。
アップロード後、HANA Cloud Database Explorer を開き、テーブル “TEST_TABLE” を確認します。アップロードしたテキストがチャンクに分割されて新規レコードとして追加されており、各レコードに対してベクトルが生成されているか確認します。
これらが正しく反映されていれば、ファイルアップロード機能の実装は完了です！
7｜次回予告
Part 7 SAPUI5 でチャットUI を構築
Part 7 では、これまでに作成した AI Agent API を呼び出すフロントエンドとして、SAPUI5 ベースのチャット UI を組み立てます。お楽しみに！
免責事項
本ブログに記載された見解および意見はすべて私個人のものであり、私の個人的な立場で発信しています。SAP は本ブログの内容について一切の責任を負いません。
生成 AI は研究室の外へ飛び出し、今やビジネス現場の常識を塗り替えています。SAPは全速力でその波に乗っています。このブログシリーズでは、SAP AI Core の既定モデルを最速で呼び出し、実務で使える AI エージェントへ拡張する“秒速ハンズオン”をお届けします。お知らせYou can find the English version here. 📖 本シリーズで学べることSAP AI Core 上でカスタム AI Agent を “秒速” で動かす方法LangChain・Google 検索ツール・RAGを使った実装AI Agent を REST API 化し、SAPUI5/Fiori の UI に載せ、Cloud Foundryにデプロイする手順学習時間各章は 10–15 分 で読める＆手を動かせるを予定しています。 🗺️ 連載ロードマップPart 0 プロローグPart 1 環境構築: SAP AI CoreとAI LaunchpadPart 2 LangChain でチャットモデルを構築Part 3 Agentツール: Google 検索を統合Part 4 RAG 基礎 ①: HANA Cloud VectorEngineと埋め込み処理Part 5 RAG 基礎 ②: Retriever Toolの構築Part 6 AI Agent を REST API 化 [現在のブログ]Part 7 SAPUI5 でチャットUI を構築Part 8 CloudFoundry にデプロイ注記続編は順次公開予定です。この記事がお役に立ったら、ぜひ Kudos を押していただけると励みになります。 「ここをもっと詳しく知りたい」「別のテーマも取り上げてほしい」など、ご要望があればお気軽にコメントください！AI Agent を REST API 化1｜はじめに本章では、これまで Notebook 上で開発してきた AI Agent を REST API として公開し、後で Cloud Foundry（CF） にデプロイできる形にまとめます。API化 することで、以下のメリットがあります:再利用性：CLI や別のアプリケーション、SAP UI5 のフロントエンドなど、あらゆるクライアントから HTTPリクエスト で呼び出せるスケーラビリティ：プロセスを Webサーバ に常駐させることで複数リクエストに対応できる運用性：CF上での自動スケールやログ管理、認証制御など、クラウド環境の標準的な仕組みを活用できる本章では、FastAPI を用いて簡単な WebAPI を実装し、LangChain ベースの AI Agent をエンドポイント化します。 2 | 事前準備BTP サブアカウントSAP AI Core インスタンスSAP AI LaunchpadのサブスクリプションPython 3.13環境 & pipVSCodeやBASなどのIDETrial 環境の注意Trial の HANA Cloud インスタンスは 毎晩自動停止 されます。日をまたぐ作業の場合は翌日インスタンスを再起動してください。 3 | フォルダ構成とライブラリ準備CF へデプロイすることを見据え、プロジェクト用のフォルダを新しく用意します。以降説明するコードや設定ファイルは、全てこのフォルダ内で管理します。# フォルダ構成
my-ai-agent-api/
├── main.py
├── requirements.txt
├── .env
└── （デプロイに必要なファイルを追加予定） CF にデプロイすることを見据えて requirements.txt で、必要な Python パッケージを明示しておきます。CF ではアプリケーション起動時、requirements.txt にしたがってライブラリを自動インストールします。今回新しく記載した fastapi や uvicorn、gunicorn は Webサーバ として必須のライブラリです。以下にようにライブラリを列挙します。# generative-ai-hub-sdk の公式ドキュメントで指定されている依存関係群
generative-ai-hub-sdk
ai_core_sdk>=2.5.7
pydantic==2.9.2
openai>=1.56.0
langchain~=0.3.0
langgraph==0.3.30
langchain-community~=0.3.0
langchain-openai>=0.2.14
langchain-google-vertexai==2.0.1
langchain-google-community==2.0.7
langchain-aws==0.2.9
google-cloud-aiplatform==1.61.0
boto3==1.35.76
# SAP HANA Cloud VectorSearch 用
hdbcli==2.24.24
langchain-hana==0.1.0
# .env ファイルを読み込むため
python-dotenv==1.1.0
# FastAPI／ASGI サーバー
fastapi==0.109.0
gunicorn
uvicorn[standard]==0.27.0 ローカルで動作確認するために、新たに仮想環境を用意しましょう。 Part 5 までに使用した仮想環境を流用しても構いませんが、ここでは念のため新規で作成します。cd my-ai-agent-api
python -m venv .venv
source .venv/bin/activate
pip install –upgrade pip
pip install -r requirements.txt 同様に、環境変数ファイル（.env）も用意しましょう。# SAP AI Core 認証情報
AICORE_CLIENT_ID = “<YOUR_AICORE_CLIENT_ID>”
AICORE_CLIENT_SECRET = “<YOUR_AICORE_CLIENT_SECRET>”
AICORE_AUTH_URL = “https://<your-region>.authentication.<your-region>.hana.ondemand.com”
AICORE_BASE_URL = “https://api.ai.prod.<your-region>.aws.ml.hana.ondemand.com”
AICORE_RESOURCE_GROUP = “<YOUR_RESOURCE_GROUP>”
DEVELOPMENT_ID = “<YOUR_OPENAI_DEPLOYMENT_ID>”
# HANA Cloud 接続情報
HANA_DB_ADDRESS = “<YOUR_HANA_DB_HOST>.hana.trial-<region>.hanacloud.ondemand.com”
HANA_DB_PORT = <YOUR_HANA_DB_PORT> # 例: 443
HANA_DB_USER = “<YOUR_HANA_DB_USER>”
HANA_DB_PASSWORD = “<YOUR_HANA_DB_PASSWORD>”
# Google カスタム検索 API
GOOGLE_CSE_ID = “<YOUR_GOOGLE_CSE_ID>”
GOOGLE_API_KEY = “<YOUR_GOOGLE_API_KEY>”これで、必要なパッケージのインストールと環境変数の用意が完了、API 開発の準備が整いました！ 4 | main.py の作成ここからは main.py ファイルを作成し、FastAPI を使って AI Agent を呼び出すエンドポイントを実装していきます。Part5 で使っていたノートブックのセルをほとんどそのまま組み合わせて作成しますが、AI Agent がどのように考え、何のアクションを取ったのかチャット画面で確認できるように、いくつか修正を加えます。具体的には、以下の流れで構造化されたレスポンスを返すようにします:環境変数の読み込み（.env）FastAPI アプリケーションの初期化LangChain / Google 検索・HANA Cloud Retriever Tool の作成AI Agent の初期化（AgentType を CHAT_ZERO_SHOT_REACT_DESCRIPTION）ログから Thought / Action / Observation を取り出す関数の定義AI Agent からのレスポンスを変換/agent/chat エンドポイントの実装import os
import re
import uvicorn
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from dotenv import load_dotenv
# LangChain関連のインポート
from gen_ai_hub.proxy.langchain.openai import ChatOpenAI
from langchain.tools import Tool
from langchain.tools.retriever import create_retriever_tool
from langchain_google_community import GoogleSearchAPIWrapper
from langchain.agents import initialize_agent, AgentType
# HANA関連のインポート
from langchain_hana import HanaInternalEmbeddings
from langchain_hana import HanaDB
from hdbcli import dbapi
# 環境変数を読み込む
load_dotenv(verbose=True)
# FastAPIアプリケーションを初期化
app = FastAPI()
# CORS設定
app.add_middleware(
CORSMiddleware,
allow_origins=[“*”],
allow_credentials=True,
allow_methods=[“*”],
allow_headers=[“*”],
)
# リクエストモデル
class QueryRequest(BaseModel):
query: str
# ChatOpenAIを初期化
chat_llm = ChatOpenAI(deployment_id=os.getenv(“DEVELOPMENT_ID”))
# Google検索ツールを設定
search = GoogleSearchAPIWrapper(k=5)
google_tool = Tool.from_function(
name=”google_search”,
description=”Search Google and return the first results”,
func=search.run
)
# embeddingsを初期化
embeddings = HanaInternalEmbeddings(
internal_embedding_model_id=”SAP_NEB.20240715″
)
# HANA Cloudと接続
connection = dbapi.connect(
address=os.getenv(“HANA_DB_ADDRESS”),
port=os.getenv(“HANA_DB_PORT”),
user=os.getenv(“HANA_DB_USER”),
password=os.getenv(“HANA_DB_PASSWORD”),
sslValidateCertificate=False,
autocommit=True,
)
# データベースの立ち上げ
db = HanaDB(
embedding=embeddings,
connection=connection,
table_name=”TEST_TABLE”
)
# retrieverツールを設定
retriever_tool = create_retriever_tool(
retriever=db.as_retriever(),
name=”hana_vectorengine”,
description=(
“Use this tool to search internal SAP documents stored “
“in HANA Cloud Vector Engine when the user asks company-specific questions.”
),
)
# エージェントを初期化
agent = initialize_agent(
tools=[google_tool, retriever_tool],
llm=chat_llm,
agent=AgentType.CHAT_ZERO_SHOT_REACT_DESCRIPTION,
verbose=True,
return_intermediate_steps=True,
)
def extract_thought_from_log(log):
if not log or not isinstance(log, str):
return “”
# 正規表現を使用して “Thought” 部分を抽出
patterns = [
r’Thought:s*(.*?)(?:nAction:|$)’,
r’^(.*?)(?:nAction:|$)’,
# 日本語対応
r’思考:s*(.*?)(?:nアクション:|$)’,
r’^(.*?)(?:nアクション:|$)’
]
# 複数のパターンを試行
for pattern in patterns:
match = re.search(pattern, log, re.DOTALL | re.IGNORECASE)
if match:
thought = match.group(1).strip()
# 空文字列や改行のみの場合はスキップ
if thought and not thought.isspace():
return thought
# マッチしない場合は元のログを返す
return log.strip()
def transform_response(raw_response):
if not isinstance(raw_response, dict):
return {“output”: “”, “intermediate_steps”: []}
output = raw_response.get(‘output’, ”)
intermediate_steps = raw_response.get(‘intermediate_steps’, [])
if not isinstance(intermediate_steps, list):
intermediate_steps = []
structured_steps = []
for idx, step in enumerate(intermediate_steps, 1):
try:
# ステップが辞書形式の場合（既に構造化されている場合）
if isinstance(step, dict):
structured_steps.append(step)
continue
# ステップがタプル形式の場合
if isinstance(step, tuple) and len(step) == 2:
action, observation = step
# ログからThoughtを抽出
thought = “”
if hasattr(action, ‘log’) and action.log:
thought = extract_thought_from_log(action.log)
# 構造化されたステップを作成
structured_step = {
“step_no”: idx,
“thought”: thought,
“action”: getattr(action, ‘tool’, ‘unknown’),
“action_input”: getattr(action, ‘tool_input’, {}),
“observation”: observation if observation is not None else “”
}
structured_steps.append(structured_step)
except Exception:
# エラーが発生した場合はスキップ
continue
return {
“output”: output,
“intermediate_steps”: structured_steps
}
@app.post(“/agent/chat”)
async def chat(request: QueryRequest):
“””エージェントにクエリを実行し、構造化されたレスポンスを返す”””
# エージェントを実行
raw_response = agent.invoke({“input”: request.query})
# レスポンスを変換
structured_response = transform_response(raw_response)
return structured_response
if __name__ == “__main__”:
uvicorn.run(app, host=”0.0.0.0″, port=8000)このセルで何をしているかAgentType.CHAT_ZERO_SHOT_REACT_DESCRIPTION: この設定を指定することで、AI Agent が「Thought → Action → Observation」という形で内部ステップを出力します。（ return_intermediate_steps=True を指定しないと response に含まれないので注意）extract_thought_from_log(): AI Agent の出力ログ（英語・日本語）から「Thought」の部分を抽出する、自作の正規表現関数です。transform_response(): (action, observation)形式のタプルを { step_no, thought, action, action_input, observation } という辞書形式に変換します。 AgentType.CHAT_ZERO_SHOT_REACT_DESCRIPTION の注意このシリーズでは、チャットUI 上に AI Agent の思考プロセスを表示するため、AgentType.CHAT_ZERO_SHOT_REACT_DESCRIPTION を採用します。ただし、この AgentType は “ツールを使わない質問” に対してエラーを返すことがあります。現在、より良い対処方法を検討中ですので、暫定的にこの点にご留意ください。 5 | 動作を確認するそれでは、用意した Python ファイルを実行し、作成した API が意図通りに動作することを確認しましょう。最初にローカル環境で FastAPI アプリケーションを起動し、API ドキュメント画面を見ながら実際にエンドポイントへリクエストを飛ばしてみます。その後、帰ってきた JSON が「output」と「intermediate_steps」の構造を満たしているかチェックします。仮想環境に入っていることを確認しつつ、ターミナルで以下のコマンドを実行し、FastAPI アプリケーションをローカルサーバとして立ち上げます。gunicorn -w 1 -k uvicorn.workers.UvicornWorker main:app –bind 0.0.0.0:${PORT:-8000}起動が完了すると、「Application startup complete」というログが表示されます。Business Application Studio（BAS）で実行している場合は以下のような画面が確認できます。ポップアップで出てきた右下の「Open in a New Tab」を押し、URLに/docsを追加して下さい。 FastAPI が自動生成した Swagger UI が表示され、/agent/chat のエンドポイントに対してどのようなリクエストを投げられるか、インタラクティブに確認できます。『POST – /agent/chat』のアコーディオンメニューを開き、「Try it out」をクリックします。Request ボディに JSON 形式で { “query”: “任意の質問文” } を入力し、送信ボタンを押すと API が即座に応答します。画面の最下部のように HTTPステータスコード”200″ が返り、Response Body が「output」と「intermediate_steps」の構造を満たしていることを確認できたら成功です！ 6｜チャレンジ – テキストファイルアップロード機能を追加しようユーザが新しく用意したテキストファイルをアップロードし、HANA Cloud VectorEngine に格納されるエンドポイントを追加しましょう。この機能を用意しておくことで、ユーザはチャット画面からドキュメントを AI Agent に渡すことができます。以下のコードを、main.py のインポート部やモデル定義、エンドポイント定義の適切な位置にまとめて挿入してみましょう。具体的な行番号やセクションはお任せしますが、インポートやモデルのそれぞれに対応するように挿入するか、あるいはファイルアップロード専用のブロックとしてまとめると良いかもしれませんね。# ファイルアップロード用のインポート
import shutil
import tempfile
from fastapi import File, UploadFile, HTTPException
from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import CharacterTextSplitter
# レスポンスモデル
class UploadResponse(BaseModel):
message: str
filename: str
chunks_created: int
@app.post(“/agent/upload”)
async def upload_file(file: UploadFile = File(…)):
“””テキストファイルをアップロードしてHANAに埋め込みデータとして格納”””
# ファイル形式のチェック
if not file.filename.endswith(‘.txt’):
raise HTTPException(
status_code=400,
detail=”txt形式のファイルのみアップロード可能です。”
)
try:
# 一時ファイルとして保存
with tempfile.NamedTemporaryFile(delete=False, suffix=’.txt’) as tmp_file:
shutil.copyfileobj(file.file, tmp_file)
tmp_file_path = tmp_file.name
# テキストファイルを読み込み
text_documents = TextLoader(
tmp_file_path
).load()
# テキストを適切なチャンクに分割
text_splitter = CharacterTextSplitter(
chunk_size = 50,
chunk_overlap = 0,
)
text_chunks = text_splitter.split_documents(text_documents)
# HANAベクターデータベースに格納
db.add_documents(text_chunks)
# 一時ファイルを削除
os.unlink(tmp_file_path)
return UploadResponse(
message=”ファイルが正常にアップロードされ、埋め込み処理が完了しました。”,
filename=file.filename,
chunks_created=len(text_chunks)
)
except Exception as e:
raise HTTPException(
status_code=500,
detail=f”ファイル処理中にエラーが発生しました: {str(e)}”
)ポイントは、FastAPI で受け取った UploadFile をそのまま扱うのではなく、必ず tempfile.NamedTemporaryFile にコピーしてパスを取得し、そのパスを TextLoader に渡す点です。こうすることで、ファイルの中身を安全にローカルディスクに保存しつつ、その後チャンク分割してHANA Cloud に送ることができます。コードを追加したら、先ほどと同様の手順で動作確認を行います。 アップロード後、HANA Cloud Database Explorer を開き、テーブル “TEST_TABLE” を確認します。アップロードしたテキストがチャンクに分割されて新規レコードとして追加されており、各レコードに対してベクトルが生成されているか確認します。これらが正しく反映されていれば、ファイルアップロード機能の実装は完了です！ 7｜次回予告Part 7 SAPUI5 でチャットUI を構築Part 7 では、これまでに作成した AI Agent API を呼び出すフロントエンドとして、SAPUI5 ベースのチャット UI を組み立てます。お楽しみに！ 免責事項本ブログに記載された見解および意見はすべて私個人のものであり、私の個人的な立場で発信しています。SAP は本ブログの内容について一切の責任を負いません。 Read More Technology Blog Posts by SAP articles
#SAPCHANNEL