はじめに
自然言語で何かを動かすのは、私にとって、遊びの一つです。
FastMCPとOpenAI Agents SDKを使って、
・ユーザーと会話しながら処理を次々と実行してくれる
・自らデータ収集と分析を行い、提案や意見を言ってくれる
・操作質問に回答してくれる
「AIコンシェルジュ」機能（自作UI）を業務システムにくっつけて、遊んでみました。
※くっつけたのは、別の目的で作っていた、架空の農園の職員用の販売システム（デモ用）です。
▼＜処理の実行＞AIコンシェルジュが、ユーザーと会話しながら、次々と処理を実行
▼＜データ分析・提案＞AIコンシェルジュが、ユーザーと会話しながら、データ分析して提案
▼＜システムのサポート＞AIコンシェルジュが、システムの仕様や操作の質問に回答
全体構成
※MCP（Model Context Protocol）やFastMCPについては、MCP公式ドキュメント、FastMCP公式ドキュメントを参照してください。
遊んでみる
解説は後回しにして、早速遊んでみます。
このAIコンシェルジュは、既存のデモ用販売システム（ある架空の農園の職員が使用する販売システム）にくっついて、ユーザーと会話しながら、処理の実行をしたり、自らデータ収集と分析をして提案や意見を言ったり、システムの使い方の質問に答えたりします。
AIコンシェルジュがやること
-
ユーザーと会話しながら、処理の実行から提案や意見まで
自然言語での会話で、顧客情報や販売データの収集や集計、販売データの登録を行います。
また、自らデータ収集/分析し、提案や意見を言ってくれたりします。
-
ユーザーからのシステムの仕様や操作の質問に回答する
付属しているシステムの仕様や操作の質問が来たら、システムのマニュアルを読んで、回答します。
※以降、「あなた」（ユーザー）と「AIコンシェルジュ」とのやり取りは、見やすくするために、画面キャプチャを貼るだけではなく、文字列で書き起こします。
会話-1） ユーザーの依頼に沿って、AIコンシェルジュが処理を次々と実行
会話-1 | 実行日：2025/6/4（水）
▼あなた:
先月の松本さんと田中さんのそれぞれの売上合計を教えて。
▼AIコンシェルジュ:
先月の売上合計は次の通りです。
- **松本理恵さん**: 1,810円
- かぼちゃ4個: 600円
- トマト5個とナス5本: 850円
- かぼちゃ3個: 360円
- **田中一郎さん**: 3,810円
- キュウリ11本: 770円
- トマト8個: 640円
- ごぼう15本: 1,500円
- ナス5本: 500円
- トマト5個: 400円
▼あなた:
松本さんの詳細な販売データが見たい。
▼AIコンシェルジュ:
松本理恵さんの先月の販売データは以下の通りです。
1.**販売日: 2025年05月06日**
- 販売番号: S2025051315103
- 商品: かぼちゃ4個
- 価格: 600円
2.**販売日: 2025年05月13日**
- 販売番号: S20250518214043
- 商品: トマト5個とナス5本
- 価格: 850円
3.**販売日: 2025年05月20日**
- 販売番号: S20250528072333
- 商品: かぼちゃ3個
- 価格: 360円
▼あなた:
販売登録忘れてた。松本さんに、先週月曜に、ごぼうを10本、1200円で売ってた。
▼AIコンシェルジュ:
販売登録が完了しました。
- **販売日**: 2025年05月26日
- **販売番号**: S20250604132639
- **商品**: ごぼう10本
- **価格**: 1200円
松本さんへの販売データに追加されました。
▼あなた:
結局、松本さんの先月の売上合計はいくらになった？
▼AIコンシェルジュ:
松本理恵さんの先月の新しい売上合計は以下の通りです：
- かぼちゃ4個: 600円
- トマト5個とナス5本: 850円
- かぼちゃ3個: 360円
- ごぼう10本: 1200円
**合計: 3,010円**
AIコンシェルジュが、ユーザーからの要求を次々と処理しています。
「販売登録忘れてた。」という表現から、気を利かせて販売登録をしてくれました。
会話-2） ユーザーの求めに応じ、AIコンシェルジュがデータを収集・分析し販促を提案
会話-2 | 実行日：2025/6/6（金） | ※一部の改行や整形は筆者
▼あなた:
佐藤次郎さんへの売上を伸ばしたい。どんな野菜をおすすめすればいいか？
▼AIコンシェルジュ:
佐藤次郎さんの最近の購入履歴を分析しました。以下のような野菜を多く購入されています：
1. **ナス** - これが最も頻繁に購入されています。
特に5月には多くの本数を購入しており、顧客の好みがうかがえます。
2. **キュウリ** - 最初の購入があり、一定の需要があります。
3. **かぼちゃ** - 少数ですが購入されています。
4. **ごぼう** - 購入履歴がありますが、頻度は低いです。
### おすすめの野菜
- **ナス**
- **キュウリ**
これらを中心に、特にナスを提案して、他の野菜とのセット販売を考えるのも良いかもしれません。
また、季節によって新鮮な野菜を随時提案することで、佐藤さんの興味を引きやすくなるでしょう。
▼あなた:
そうなんだ。このお客さんの詳細な全販売データを見たい。
▼AIコンシェルジュ:
佐藤次郎さんの全販売データを以下にまとめます：
1. **2025年05月10日**
- 商品：キュウリ15本
- 価格：1,050円
- 販売ID：S20250515035710
2. **2025年05月14日**
- 商品：ナス8本
- 価格：720円
- 販売ID：S20250515110242
3. **2025年05月17日**
- 商品：かぼちゃ4個
- 価格：560円
- 販売ID：S20250519092049
4. **2025年05月20日**
- 商品：ナス10本
- 価格：900円
- 販売ID：S20250528232131
5. **2025年05月29日**
- 商品：ナス5本
- 価格：500円
- 販売ID：S20250531222853
6. **2025年06月02日**
- 商品：ナス7本
- 価格：800円
- 販売ID：S20250604224647
7. **2025年04月27日**
- 商品：ごぼう6本
- 価格：660円
- 販売ID：S20250605072006
### 合計販売数と金額
- **合計販売額**：4,190円
- **合計購入商品数**：合計51本（キュウリ、ナス、かぼちゃ、ごぼう）
このデータを基に、今後の営業戦略を考える際に役立てることができます。
何かさらに知りたいことがあれば教えてください！
佐藤次郎さんへの売り上げを伸ばしたいので、率直に、AIコンシェルジュに、何を売ればいいのか聞いてみました。
そこで、AIコンシェルジュは、自ら佐藤次郎さんの過去の販売データ全件を収集・分析し、「特にナスを提案して～」と、販促の提案してくれました。
会話-3） ユーザーのシステム操作の質問に、AIコンシェルジュが回答
会話-3
▼あなた:
販売登録画面の販売日は、売ったのが今日でも、わざわざ入れなきゃいけない？
▼AIコンシェルジュ:
販売登録画面では、販売日には最初から今日の日付が自動で入っています。
もし空欄であれば、今日の日付が設定されるため、基本的には手動で入力する必要はありません。
▼あなた:
顧客コードを間違って入れたらどうなる？
▼AIコンシェルジュ:
顧客コードを間違えて入力した場合、そのコードが存在しないと入力がクリアされます。
正しい顧客コードが入力されない限り、販売登録を進めることができません。
ユーザーからの販売システムの仕様や操作についての質問を受け、AIコンシェルジュが回答する、という会話です。
AIコンシェルジュは、質問の内容から、販売システムのマニュアルPDFを読んで、質問に回答しました。
仕組みと実装
仕組み
全体構成
FastMCPとOpenAI Agents SDKを使用しています。
Agentの内部のLLMはgpt-4o-miniです。
MCPサーバー
MCPクライアントからの要求に従い、
・DBの読み書きを行います。
・販売システムのマニュアルPDFを検索し、質問に対する回答を生成します。
前提
DBヘアクセスする頻度が高くない、という想定です。
頻繁にアクセスする場合は、違った実装もあるかもしれません。
MCPクライアントとのTransportに、標準入出力ではなくStreamable Httpを使用しています。
なので、MCPサーバーを事前に起動しておく必要があります。
双方同一マシン上にあるのでこんなの不要ですが、まあ遊びです。
アクセスするDB
ある架空の農園の業務用DBです。
顧客マスタのテーブルと、販売データのテーブルがあります。
これらのテーブルを、以下のFunction（ストアドプロシージャ）を使って操作します。
「get_customers」：顧客データの収集
「get_sales」：販売データの収集
「add_sale」：販売データの登録
ツール
MCPサーバーが備えるツールは以下の4つです。
| ツール名 | 機能概要 |
|---|---|
| get_customers | 顧客データの収集 |
| get_sales | 販売データの収集 |
| add_sale | 販売データの登録 |
| search_manual_sales | 販売システムのマニュアルPDFを検索して質問に対する回答を生成 |
実装コード
MCPサーバー ※一部省略
from fastmcp import FastMCP
import os
from typing import Annotated
from pydantic import Field
from typing import Dict, Any, Optional
import logging
from contextlib import asynccontextmanager
from dataclasses import dataclass
from typing import AsyncIterator
from datetime import datetime
import psycopg2
import psycopg2.extras
from docsearch import search_in_document # 自作のパッケージ。指定された文書内を検索して質問に回答。
# ロギング設定（出力フォーマット追加）
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [FAST-MCP-SERVER-SALES-HTTPSTR] %(message)s"
)
logger = logging.getLogger("fast-mcp-server-sales-httpstr")
# DB接続とカーソルをグローバル変数として保持
db_connection = None
db_cursor = None
# 文書内サーチクラスのインスタンス
search_document = None
@dataclass
class AppContext:
server_name: str = "MCP Server for the farmer sales system"
start_time: str = ""
@asynccontextmanager
async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
"""
ライフスパン。
MCPクライアント側のasync with Clientブロックの開始でtry、終了でfinallyが実行される。
"""
global db_connection
try:
# MCPクライアント側のasync with Clientブロックの開始
# ここでDB接続はしない。MCPクライアントからMCPサーバーへの要求には、DBとは関係ないものもある。
yield AppContext()
finally:
# MCPクライアント側のasync with Clientブロックの終了
# DB切断（DB接続があれば）はasync with Clientブロック終了の都度行う。
if db_connection is not None:
(省略)
mcp = FastMCP("Farmer_sales_system_server",
instructions="農園の販売システムのMCPサーバー",
lifespan=app_lifespan)
def get_db_connection():
'''有効なDB接続が無い場合のみ、新たにDB接続し、カーソルを作成'''
global db_connection, db_cursor
if db_connection is None or db_connection.closed:
（省略）
return db_connection, db_cursor
@mcp.tool()
async def get_customers(
code_customer: Annotated[str, Field(description="顧客マスタの検索条件の1つ。顧客コード。完全一致。オプション。", default=None)],
name_str_customer: Annotated[str, Field(description="顧客マスタの検索条件の1つ。顧客名の一部または全部。部分一致。オプション。", default=None)]
) -> Dict[str, Any]:
"""
販売システムにおいて、顧客マスタの検索を行い、顧客データを返す。（プロンプト例：顧客コード1234の名前と住所が知りたい。）（プロンプト例：佐藤さんというお客様の顧客データが見たい。）
"""
try:
# 有効なDB接続がもし無い場合に限り、新規作成
conn, cur = get_db_connection()
# DB内Function「get_customers」を使用
cur.execute(
"SELECT * FROM get_customers(%s, %s)",
(code_customer, name_str_customer)
)
# 結果を取得
customers = cur.fetchall()
# 戻り値の形式を整える
return {
"status": "success",
"data": {
"customers": customers,
"count": len(customers)
}
}
except Exception as e:
# エラーハンドリング
logger.error(f"顧客データ取得エラー: {e}")
return {
"status": "error",
"data": {
"error": str(e)
}
}
@mcp.tool()
async def get_sales(
code_customer: Annotated[str, Field(description="販売データの必須検索条件の1つ。検索したい販売データの販売相手である顧客のコードで絞りたい場合に使用。")],
code_sales: Annotated[str, Field(description="販売データの任意検索条件の1つ。検索したい販売データの販売番号で指定して検索したい場合に使用。", default=None)],
date_sales_begin_str: Annotated[str, Field(description="販売データの任意検索条件の1つ。検索したい販売データの販売日を期間で指定して検索したい場合に使用。その検索期間の開始年月日。YYYY-MM-DDの形式。", default=None)],
date_sales_end_str: Annotated[str, Field(description="販売データの任意検索条件の1つ。検索したい販売データの販売日を期間で指定して検索したい場合に使用。その検索期間の終了年月日。YYYY-MM-DDの形式。", default=None)]
) -> Dict[str, Any]:
"""
販売システムにおいて、販売データの検索を行い、検索結果を返す。（プロンプト例：顧客コード1234のお客様の販売実績を知りたい）（プロンプト例：顧客コード1234の先月の販売合計額は？）
"""
try:
# 有効なDB接続がもし無い場合に限り、新規作成
conn, cur = get_db_connection()
# 日付文字列が指定されている場合はISO 8601フォーマットに変換
if date_sales_begin_str:
try:
# 日付の解析とISO 8601形式への変換
date_obj = datetime.strptime(date_sales_begin_str, '%Y-%m-%d')
# ISO 8601形式（YYYY-MM-DD）に変換
date_sales_begin_str = date_obj.strftime('%Y-%m-%d')
except ValueError:
logger.error(f"販売日期間の開始日を日付型に変換できませんでした: {date_sales_begin_str}")
raise ValueError(f"販売日期間の開始日の形式が不正です。YYYY-MM-DD形式で指定してください: {date_sales_begin_str}")
if date_sales_end_str:
#logger.info(f"date_sales_end_str:{date_sales_end_str}")
try:
# 日付の解析とISO 8601形式への変換
date_obj = datetime.strptime(date_sales_end_str, '%Y-%m-%d')
# ISO 8601形式（YYYY-MM-DD）に変換
date_sales_end_str = date_obj.strftime('%Y-%m-%d')
except ValueError:
logger.error(f"販売日期間の終了日を日付型に変換できませんでした: {date_sales_end_str}")
raise ValueError(f"販売日期間の終了日の形式が不正です。YYYY-MM-DD形式で指定してください: {date_sales_end_str}")
# DB内Function「get_sales」を使用
cur.execute(
"SELECT * FROM get_sales(%s, %s, %s, %s)",
(code_customer, code_sales, date_sales_begin_str, date_sales_end_str)
)
# 結果を取得
sales_data = cur.fetchall()
# 結果が空でないことを確認
if not sales_data:
return {
"status": "error",
"data": {
"error": "販売データ取得に失敗しました"
}
}
# 最初のレコードのstatus_codeを確認
status_code = sales_data[0]['status_code']
status_message = sales_data[0]['status_message']
match status_code:
case 404:
# 顧客が存在しない場合（status_code = 404）
return {
"status": "error",
"data": {
"error": status_message
}
}
case 204:
# 検索条件に合うデータがない場合（status_code = 204）
return {
"status": "success",
"data": {
"sales": sales_data,
"count": 0
}
}
case 200:
# 検索条件に合うデータがある場合（status_code = 200）
return {
"status": "success",
"data": {
"sales": sales_data,
"count": len(sales_data)
}
}
case _:
# その他のエラー
return {
"status": "error",
"data": {
"error": f"予期しないステータスコード: {status_code}"
}
}
except Exception as e:
# エラーハンドリング
logger.error(f"販売データ取得エラー: {e}")
return {
"status": "error",
"data": {
"error": str(e)
}
}
@mcp.tool()
async def add_sale(
code_customer: Annotated[str, Field(description="販売相手の顧客のコード。必須。")],
products: Annotated[str, Field(description="販売した物。必須。（例：キュウリ15本とナス8個）")],
price: Annotated[int, Field(description="販売価格。通貨を除いた整数部分。必須。（例：2000）")],
date_sales_str: Annotated[str, Field(description="販売した年月日。YYYY-MM-DDの形式。オプション。", default=None)]
) -> Dict[str, Any]:
"""
販売システムにおいて、販売データの登録を行い、登録された販売データを返す。（プロンプト例：販売登録。キュウリ15本とナス8個を2000円で顧客コード1234のお客様に売った。）（プロンプト例：販売登録。昨日、キュウリ15本とナス8個を2000円で売った。顧客コードは1234。）
"""
try:
# 有効なDB接続がもし無い場合に限り、新規作成
conn, cur = get_db_connection()
# 日付文字列の処理
if date_sales_str:
try:
# 日付の解析とISO 8601形式への変換
date_obj = datetime.strptime(date_sales_str, '%Y-%m-%d')
# ISO 8601形式（YYYY-MM-DD）に変換
date_sales_str = date_obj.strftime('%Y-%m-%d')
except ValueError:
logger.error(f"販売日を日付型に変換できませんでした: {date_sales_str}")
raise ValueError(f"販売日の形式が不正です。YYYY-MM-DD形式で指定してください: {date_sales_str}")
else:
# 指定がない場合は現在の日付をISO 8601形式で設定
date_sales_str = datetime.now().strftime('%Y-%m-%d')
# DB内Function「add_sale」を使用
cur.execute(
"SELECT * FROM add_sale(%s, %s, %s, %s);",
(code_customer, products, price, date_sales_str)
)
# 結果を取得（1レコードのみ）
result = cur.fetchone()
# status_code に応じて戻り値を作成
if result['status_code'] == 404:
# 顧客が存在しない場合はエラーとして扱う
return {
"status": "error",
"data": {
"error": result['status_message']
}
}
else:
# 正常終了の場合（status_codeが201）
return {
"status": "success",
"data": {
"sales_registered": result
}
}
except Exception as e:
# エラーハンドリング
return {
"status": "error",
"data": {
"error": str(e)
}
}
@mcp.tool()
def search_manual_sales(
query: Annotated[str, Field(description="販売システムの仕様や操作法などの質問。必須。")]
) -> dict:
'''
販売システムの仕様や操作法などの質問に対応し、販売システムマニュアルを検索し、関連箇所を要約して返す。
'''
global search_document
try:
# search_in_documentクラスのインスタンスがNoneならインスタンス化
if search_document is None:
search_document = search_in_document()
# 環境変数からPDFパスを取得
pdf_path = os.getenv('SALES_MANUAL_PATH')
if not pdf_path:
return {
"status": "error",
"data": {
"error": "環境変数 'SALES_MANUAL_PATH' が設定されていません。"
}
}
# 指定された文書内の検索を実行
status_code, result = search_document.search_in_file(pdf_path, query)
# ステータスコードに応じて戻り値を設定
match status_code:
case 404: # PDFファイルが無い
return {
"status": "error",
"data": {
"error": "販売システムマニュアルのファイルがありません。"
}
}
case 201: # 検索結果無し
return {
"status": "success",
"data": {
"search_result": "販売システムマニュアルに記載はありませんでした。"
}
}
case 200: # 検索結果有り
return {
"status": "success",
"data": {
"search_result": result
}
}
case _: # その他のエラー
return {
"status": "error",
"data": {
"error": f"予期しないステータスコード: {status_code}"
}
}
except Exception as e:
return {
"status": "error",
"data": {
"error": str(e)
}
}
@mcp.prompt()
def system_message() -> str:
"""LLMに投入するシステムメッセージを返す"""
# 現在年月日と曜日の情報を取得
today = datetime.now()
iso_date = today.strftime("%Y-%m-%d")
weekday_ja = ["月曜日", "火曜日", "水曜日", "木曜日", "金曜日", "土曜日", "日曜日"]
weekday_str = weekday_ja[today.weekday()]
# システムメッセージを構築
system_text = (
"あなたは、ある農園の販売システムのAIエージェントです。この農園の販売システムでは、顧客に対し、野菜を販売しています。"
"ユーザーの依頼に応じて、顧客マスタの顧客データを収集したり、販売登録したり、販売データの収集をして下さい。"
"また、ユーザーから提案や意見を求められたら、積極的にこの販売システムのデータを収集し、それを基に回答して下さい。"
"ユーザーからのこの販売システムの仕様や操作法などの質問にも回答して下さい。"
"この販売システムには関係ないプロンプトにはツールを適用しないで、無視してください。"
f"\n週は、月曜日にはじまり、日曜日に終わります。"
f"\n現在の日付は {iso_date}（{weekday_str}）です。"
)
return system_text
if __name__ == "__main__":
# MCPサーバー起動。
# Streamable Httpなので、事前にこのMCPサーバーを起動させておく。
mcp.run(transport="streamable-http", host="127.0.0.1", port=8000)
MCPクライアント
UI「AIコンシェルジュ」画面から、ユーザーの自然言語の要求を受け取ります。
OpenAI Agents SDKを使用しているので、中のAgentが、自分が保持しているMCPサーバーツール群からその自然言語の要求に合うようなツールを適宜選択し、MCPサーバーに実行させます。
その実行結果をUI「AIコンシェルジュ」画面に渡します。
実装コード
MCPクライアント
import asyncio
import os
import logging
from agents import Agent, Runner, RunConfig
from agents.mcp import MCPServerStreamableHttp
from dotenv import load_dotenv
from datetime import datetime
# 環境変数の読み込み
load_dotenv()
# ロギング設定
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [FAST-MCP-CLIENT-SALES-HTTPSTR] %(message)s"
)
logger = logging.getLogger("fast-mcp-client-sales-httpstr")
LLM_MODEL_NAME ="gpt-4o-mini"
# ユーザーとAgentのやりとりの履歴
# OpenAI Agents SDKの履歴管理を活用
conversation_items = []
# Agentに投入する指示
# 現在年月日と曜日の情報を取得
today = datetime.now()
iso_date = today.strftime("%Y-%m-%d")
weekday_ja = ["月曜日", "火曜日", "水曜日", "木曜日", "金曜日", "土曜日", "日曜日"]
weekday_str = weekday_ja[today.weekday()]
# Agentに投入する指示を構築
AGENT_INSTRUCTIONS = (
"あなたは、ある農園の販売システムのAIエージェントです。この農園の販売システムでは、顧客に対し、野菜を販売しています。"
"ユーザーの依頼に応じて、顧客マスタの顧客データを収集したり、販売登録したり、販売データの収集をして下さい。"
"また、ユーザーから提案や意見を求められたら、積極的にこの販売システムのデータを収集し、それを基に回答して下さい。"
"ユーザーからのこの販売システムの仕様や操作法などの質問にも回答して下さい。"
"この販売システムには関係ないプロンプトにはツールを適用しないで、無視してください。"
f"\n週は、月曜日にはじまり、日曜日に終わります。"
f"\n現在の日付は {iso_date}（{weekday_str}）です。"
)
def process_user_prompt(user_input: str) -> str:
"""
ユーザーのプロンプトをAgentが処理。
UIとのやり取りはこの同期関数が行う。
"""
async def async_process():
global conversation_items
try:
async with MCPServerStreamableHttp(
params={
"url": "http://127.0.0.1:8000/mcp",
}
) as server:
await server.list_tools()
agent = Agent(
name="農園販売システムのAIエージェント",
instructions=AGENT_INSTRUCTIONS,
mcp_servers=[server],
model=LLM_MODEL_NAME
)
# 新しいユーザー入力を今までのやりとりに追加
current_input = conversation_items + [{"role": "user", "content": user_input}]
# tracing_disabled=Trueをしないと、Agentがcompleteした後でも、OpenAIのトレーシングシステムがデバッグ情報を収集・送信してしまい、終わってないように見える。
run_config = RunConfig(tracing_disabled=True)
# Agentに、新しいユーザー入力を追加した今までのやりとりcurrent_inputを投入し、回答してもらう
result = await Runner.run(agent, current_input, run_config=run_config)
# 今までのやりとりを抽出・保存（以降も会話のコンテキストとして使用）
conversation_items = result.to_input_list()
ret = result.final_output.strip()
# async with ブロックを抜け、MCPサーバーとのMCP通信セッションが自動的に遮断される
return ret
except Exception as e:
logger.error(f"プロンプト処理中にエラー発生: {type(e).__name__}: {e}")
return f"[MCP] エラーが発生しました: {str(e)}"
try:
return asyncio.run(async_process())
except Exception as e:
logger.error(f"asyncio実行エラー: {type(e).__name__}: {e}")
return f"[MCP] 実行エラー: {str(e)}"
AgentにInstructionsで現在年月日と週の定義を与える
Agentの内部のLLMは、「今がいつか」を知りません。
考えてみれば当たり前で、そんなの学習データには無いからです。
アプリケーションとしての「ChatGPT」は今がいつかを知っていますが、それはアプリケーション「ChatGPT」の話であって、API経由で扱うLLM「GPT」の話ではないです。
一方で、ユーザーの自然言語での要求には、ほぼ必ず、現時点からの相対表現が含まれます。
週と曜日を使用した相対表現も当然あり得ます。
例）「顧客コードxxxxのお客様の先月の販売データ全件を見せて。」
例）「販売登録して。顧客コードxxxxのお客様に、先週水曜日、～を1000円で販売した。」
具体的な年月日が使われる方が、むしろ少ないはずです。
このような相対表現をGPTが解釈して、ツールの年月日の引数を生成する場合、
・今は何年何月何日の何曜日か
・「週」は何曜日から何曜日までか
という情報がGPT自身に必要となります。
ですので、それらをInstructionsに含めて、Agent（の内部のLLMであるGPT）に提供します。
Agentに現在年月日と週の定義を与えるInstructionsを生成
# Agentに投入する指示
# 現在年月日と曜日の情報を取得
today = datetime.now()
iso_date = today.strftime("%Y-%m-%d")
weekday_ja = ["月曜日", "火曜日", "水曜日", "木曜日", "金曜日", "土曜日", "日曜日"]
weekday_str = weekday_ja[today.weekday()]
# Agentに投入する指示を構築
AGENT_INSTRUCTIONS = (
"あなたは、ある農園の販売システムのAIエージェントです。この農園の販売システムでは、顧客に対し、野菜を販売しています。"
"ユーザーの依頼に応じて、顧客マスタの顧客データを収集したり、販売登録したり、販売データの収集をして下さい。"
"また、ユーザーから提案や意見を求められたら、積極的にこの販売システムのデータを収集し、それを基に回答して下さい。"
"ユーザーからのこの販売システムの仕様や操作法などの質問にも回答して下さい。"
"この販売システムには関係ないプロンプトにはツールを適用しないで、無視してください。"
f"\n週は、月曜日にはじまり、日曜日に終わります。"
f"\n現在の日付は {iso_date}（{weekday_str}）です。"
)
ユーザーとAgentの複数ターンに渡る会話の実現
ユーザーとAgentとの会話は、通常、複数ターンに渡って行われます。
しかし、実装において何も考慮しないと、Agentとの会話は「1ターン」で消失します。
（それは当たり前で、そもそも1ターン毎にAgentを生成しているので。普通の実装はそうするはず。）
なので、1個前の会話を前提に発話しても、全く通じません。
Agentと複数ターンに渡る一貫した会話をするためには、基本的にはAgent呼び出し側が仕組みを作る必要があります。
具体的には、今までのユーザー入力とそれに対するAgentの回答が入ったlistの準備と、それをAgentに毎回投入することです。
そして、OpenAI Agents SDKには、それを手助けしてくれる仕組みがあります。
# Agentに、新しいユーザー入力を追加した今までのやりとりcurrent_inputを投入し、回答してもらう
result = await Runner.run(agent, current_input, run_config=run_config)
Agentからの回答resultは、RunResultクラスのインスタンスです。
RunResultクラスには、to_input_list()というメソッドがあります。
この戻り値は、このrunに投入された入力と、それに対する回答、及びそこに至るまでのAgentのツール呼び出しとその結果などが入ったlistです。要するに、そのlistにはこの1ターン分のやり取りとその経緯全てが入っているわけです。
そこで、ちょっと工夫して、
Agentから何らかの回答resultを得た時、to_input_list()でその1ターン分のやり取りlistを抽出・保持しておきます。
そして、新たにユーザー入力があり、新たにAgentに回答してもらいたい時は、Agent呼び出し側が保持しているそのlistにその新しいユーザー入力を追加し、そのlistをAgentに投入する、ということをします。
毎ターンこれを繰り返すことにより、Agentは、今までのユーザーとの会話とその経緯を全部把握したうえで、新たなユーザー入力に対処してくれます。