LangChainでハマったこと、よく使う処理やパターン等をまとめます。（随時更新）
主な環境
- Python 3.11.8
- LangChain 0.1.14
OpenAIのVision APIを利用する
以下のようにHumanMessageにメッセージと画像URLのリストを渡せばOKです。
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
chat = ChatOpenAI(model="gpt-4-turbo")
url = "https://cdn-ak.f.st-hatena.com/images/fotolife/n/nana8a0a/20240323/20240323111544.png"
res = chat.invoke([
HumanMessage([
{"type": "text", "text": "何が写っていますか？"},
{"type": "image_url", "image_url": {"url": url}},
],
)])
print(res.content)
結果:
この画像は「AWS Certified SysOps Administrator - Associate」の認定証です。画像にはAWSのロゴと「training and certification」というテキスト、認定を受けた人の名前やID、スコア、有効期限などが書かれており、その人が試験に合格したことを示しています。具体的な試験のスコアや他の個人情報は黒塗りで隠されています。
複数の画像を渡して違いを調べてもらうこともできます。
url1 = "https://cdn-ak.f.st-hatena.com/images/fotolife/n/nana8a0a/20240323/20240323111544.png"
url2 = "https://cdn-ak.f.st-hatena.com/images/fotolife/n/nana8a0a/20240227/20240227083446.png"
res = chat.invoke([
HumanMessage([
{"type": "text", "text": "2つの画像の違いは？"},
{"type": "image_url", "image_url": {"url": url1}},
{"type": "image_url", "image_url": {"url": url2}},
],
)])
print(res.content)
結果:
２つの画像にはいくつかの違いがあります。
証明書のタイトル:
- 最初の画像のタイトルは「AWS Certified SysOps Administrator - Associate」です。
- 二番目の画像のタイトルは「AWS Certified Developer - Associate」です。
試験結果のスコア:
- 最初の画像ではスコアは813です。
- 二番目の画像ではスコアは850です。
試験日:
- 最初の画像の試験日は2024年3月14日です。
- 二番目の画像の試験日は2024年2月23日です。
参考
Web検索機能を使う（Tavily Search APIを使う）
LangChain公式のクイックスタートのAgent機能の説明にあったものです。Tavily Search APIを使います。
事前にtavily.comでアカウントを作ってAPIキーを取得して、環境変数TAVILY_API_KEYにセットしておく必要があります。
import langchain
langchain.verbose = True
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults
tools = [TavilySearchResults()]
chat = ChatOpenAI()
prompt = ChatPromptTemplate.from_messages([
MessagesPlaceholder("chat_history"),
MessagesPlaceholder("agent_scratchpad"),
])
agent = create_openai_tools_agent(chat, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, max_iterations=5)
res = agent_executor.invoke({"chat_history": [
SystemMessage("必ず関西弁で受け答えしてください。"),
AIMessage("おおきに。"),
HumanMessage("阪神の2023年のドラフト1位は誰？"),
]})
print(res["output"])
途中経過:
> Entering new AgentExecutor chain...
Invoking: `tavily_search_results_json` with `{'query': '2023年 阪神 ドラフト 1位'}`
[{'url': 'https://draft.npb.jp/draft/2023/draftlist_t.html', 'content': '富山GRNサンダーバーズ. 2位. 福島 圭音. 外野手. 白鴎大学. 2022. 日本野球機構（NPB）オフィシャルサイト。. プロ野球12球団の試合日程・結果や予告先発、ドラフト会議をはじめ、事業・振興に関する情報を掲載。. また、オールスター･ゲームや日本シリーズ ...'}, {'url': 'https://www.baseballchannel.jp/npb/tigers/draftkaigi2023/designated-player/breaking-news/list/', 'content': '「2023年プロ野球ドラフト会議 supported by リポビタンD」が、10月26日16時50分から行われる。プロ志望届を提出した高校生・大学生加え、社会人や独立リーグの候補者が運命の時を待つ。ここでは、阪神タイガースの指名選手一覧（ドラフト1位の抽選結果含む）を速報する。'}, {'url': 'https://www.draft-kaigi.jp/draftnews/2023draftnews/74893/', 'content': '2023年のドラフト会議は10月26日に行われ、支配下ドラフトが72人（昨年より＋3人）、育成ドラフトが50人（昨年より−7人）の、合わせて122人（昨年より-4人）が指名されました。 ... 青学大・西川史礁選手が初戦で2安打、12球団が視察しヤクルト・楽天・阪神 ...'}, {'url': 'https://www.youtube.com/watch?v=K_zO4e-m1yQ', 'content': '2023年のプロ野球ドラフト会議にて阪神タイガースから1位指名された下村海翔投手（青山学院大）の貴重な侍ジャパン選出時の映像。下村海翔 ...'}, {'url': 'https://hanshintigers.jp/news/topics/draft2023.html', 'content': 'プロ野球ドラフト会議2023速報. 2023年10月26日 更新. 26日 (木)、プロ野球の新人選手選択会議「プロ野球ドラフト会議 supported by リポビタンD」が行われ､阪神タイガースは下村海翔選手 (青山学院大)ら8選手を指名し交渉権を獲得しました｡.'}]2023年の阪神タイガースのドラフト1位指名は、下村海翔投手（青山学院大）やで。
> Finished chain.
結果:
2023年の阪神タイガースのドラフト1位指名は、下村海翔投手（青山学院大）やで。
参考
Agentから呼び出せる自前のToolを作る方法です。BaseToolを継承したクラスを作ります。
以下は雑にWebページの情報を抜き出すToolの実装例です。（HTMLファイルから本文を抜き出すためにtrafilaturaを使っています）
from langchain_core.messages import HumanMessage, SystemMessage
from langchain.pydantic_v1 import BaseModel, Field
from langchain.tools import BaseTool
from langchain.callbacks.manager import CallbackManagerForToolRun
from langchain_openai import ChatOpenAI
import logging as logger
import trafilatura
class ExtractWebPageInfoSchema(BaseModel):
url: str = Field(description="対象のWebページのURL")
prompt: str = Field(default="", description="ページから抜き出したい情報の指示文。例: 'このページの要約をしてください。'")
class ExtractWebPageInfo(BaseTool):
name = "extract_web_page_info"
description = "指定されたWebページから、指定された情報を抜き出します。"
args_schema: type[BaseModel] = ExtractWebPageInfoSchema
def _run(self,
url: str,
prompt: str = "",
run_manager: CallbackManagerForToolRun | None = None,
) -> str:
try:
logger.debug(f"web page 取得中... 指示: '{prompt}' url: {url}")
html = trafilatura.fetch_url(url)
content = trafilatura.extract(html)
logger.debug(f"取得完了 長さ: {len(content)} content: '{content}'")
if len(content) > 3000:
logger.debug("長すぎるので先頭一定数のみを入力します。")
content = content[:3000]
chat = ChatOpenAI()
logger.debug("chat api 問い合わせ中...")
res = chat.invoke([
SystemMessage("指示に従ってWebページの情報を抽出して要約してください。"),
HumanMessage("指示: " + prompt),
HumanMessage(f'Webページの内容:\n"""\n{content}\n"""'),
])
logger.debug(f"chat api 結果受信 {res}")
return res
except Exception as e:
logger.error(f"ExtractWebPageInfoでエラーが発生しました。{e}")
return str(e)
import langchain
langchain.verbose = True
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI
tools = [ExtractWebPageInfo()]
chat = ChatOpenAI(model="gpt-4")
prompt = ChatPromptTemplate.from_messages([
MessagesPlaceholder("chat_history"),
MessagesPlaceholder("agent_scratchpad"),
])
agent = create_openai_tools_agent(chat, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, max_iterations=5)
res = agent_executor.invoke({"chat_history": [
SystemMessage("必ず関西弁で受け答えしてください。"),
AIMessage("おおきに。なんの用や？"),
HumanMessage("このページの内容を100文字程度で教えて。 https://blog.778a0a.com/entry/2024/01/21/161800"),
]})
print(res["output"])
途中経過:
> Entering new AgentExecutor chain...
Invoking: `extract_web_page_info` with `{'url': 'https://blog.778a0a.com/entry/2024/01/21/161800', 'prompt': 'このページの内容を100文字程度で教えてください。'}`
web page 取得中... 指示: 'このページの内容を100文字程度で教えてください。' url: https://blog.778a0a.com/entry/2024/01/21/161800
取得完了 長さ: 1352 content: '『御成敗式目 鎌倉武士の法と生活』を読んだので感想です。...'
chat api 問い合わせ中...
chat api 結果受信 content='...'
content='『御成敗式目 鎌倉武士の法と生活』を読んだ感想。御成敗式目の制定経緯や武士社会への影響、吾妻鏡の評価など紹介。歴史書の物事評価には注意が必要。北条泰時の野心、後世の影響も考慮。鎌倉時代後期や室町時代についての書籍も読みたいと述べる。' response_metadata={'token_usage': {'completion_tokens': 158, 'prompt_tokens': 1562, 'total_tokens': 1720}, 'model_name': 'gpt-3.5-turbo', 'system_fingerprint': 'fp_c2295e73ad', 'finish_reason': 'stop', 'logprobs': None} ...
> Finished chain.
結果:
あのページは「御成敗式目 鎌倉武士の法と生活」についての感想や評価が書かれてるで。御成敗式目の制定経緯や武士社会への影響、吾妻鏡の評価や北条泰時の野心も語られててな。また、鎌倉時代後期や室町時代についての書籍にも興味を示してるんや。
AgentExecutorのおかげでTavilySearchResultsと組み合わせて良い感じのプロンプト文を入力すると、TavilySearchResultsで記事を検索してからExtractWebPageInfoで中身を読み取ってくれるので便利です。
from langchain_community.tools.tavily_search import TavilySearchResults
tools = [ExtractWebPageInfo(), TavilySearchResults()]
chat = ChatOpenAI(model="gpt-4")
prompt = ChatPromptTemplate.from_messages([
MessagesPlaceholder("chat_history"),
MessagesPlaceholder("agent_scratchpad"),
])
agent = create_openai_tools_agent(chat, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, max_iterations=5)
res = agent_executor.invoke({"chat_history": [
SystemMessage("必ず関西弁で受け答えしてください。"),
AIMessage("おおきに。なんの用や？"),
HumanMessage("778a0aの日記というブログのなかのパウロについての本の感想記事を探して100文字程度で要約して。"),
]})
途中経過:
> Entering new AgentExecutor chain...
Invoking: `tavily_search_results_json` with `{'query': '778a0a ブログ パウロについての本 感想'}`
[{'url': 'https://blog.778a0a.com/entry/2024/02/27/083624', 'content': '「AWS Certified Developer - Associate（以下DVA）」に合格しましたので感想です。 ...'}, {'url': 'https://ja.ligonier.org/blog/5-things-paul/', 'content': '使徒パウロについて知っておくべき五つのこと ...'}, {'url': 'https://bookmeter.com/books/11259434', 'content': '全54件中 1-40 件を表示. パウロ 十字架の使徒 (岩波新書) の 評価70％ ...'}, {'url': 'https://blog.778a0a.com/entry/2024/01/26/222537', 'content': '『パウロ 十字架の使徒』を読んだので感想です。 ...'}, {'url': 'https://ameji56.hatenablog.com/entry/2023/12/23/214100', 'content': '岩波新書の赤、青野太潮「パウロ」（2016年）は、初期キリスト教の使徒であり、新約聖書の著者の一人であったパウロに関する新書である。 ...'}]
Invoking: `extract_web_page_info` with `{'url': 'https://blog.778a0a.com/entry/2024/01/26/222537', 'prompt': 'このページの要約をしてください。'}`
web page 取得中... 指示: 'このページの要約をしてください。' url: https://blog.778a0a.com/entry/2024/01/26/222537
取得完了 長さ: 1451 content: '『パウロ 十字架の使徒』を読んだので感想です。...'
chat api 問い合わせ中...
chat api 結果受信 content='...'
content='『パウロ 十字架の使徒』は初期のキリスト教について詳しく知ることができる一冊で、パウロの活動や当時のキリスト教・ユダヤ教の状況について分かりやすく解説されています。初期のキリスト教では異民族への布教やユダヤ教的な律法主義に対する考え方に対立があり、その歴史的経緯が紹介されています。後半部分では贖罪論などについての議論も紹介されており、一般的なキリスト教の解釈との違いも示唆されています。読者は本書を通じてキリスト教の歴史や異端・非主流な宗派、ローマ帝国の歴史、イスラム教の歴史などについても興味を持ち、深く学ぶことを希望しています。' response_metadata={'token_usage': {'completion_tokens': 306, 'prompt_tokens': 1641, 'total_tokens': 1947}, 'model_name': 'gpt-3.5-turbo', 'system_fingerprint': 'fp_c2295e73ad', 'finish_reason': 'stop', 'logprobs': None} ...
結果:
『パウロ 十字架の使徒』についての感想記事やけど、初期のキリスト教について詳しく知れる本やったらしいで。パウロの活動や、当時のキリスト教・ユダヤ教の状況を分かりやすく説明してくれてるんやて。異民族への布教やユダヤ教的な律法主義に対する考え方の対立があった歴史的経緯も紹介されてるし、贖罪論についての議論も出てくるらしいで。キリスト教の歴史や異端・非主流な宗派、ローマ帝国の歴史、イスラム教の歴史などについても学びたくなる本やったみたいやな。
参考
Slackで自然言語でリマインダーが設定できると便利ですが、Slackのリマインダー機能はAPIやBotから操作できない（最新のAPIで使えなくなっている）ので、自前で作る必要があったので作りました。めちゃくちゃ雑な実装です。
基本的には、いつリマインドしたいかという自然言語のクエリーを受け取って、それを任意の日付型データに落とし込めればOKです。今回はその変換タスクはChatGPTにやってもらっています。定期実行には対応していないですが、クエリーをcron式に落とし込めればいいのかなあと思います。
import datetime
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
from langchain.pydantic_v1 import BaseModel, Field
from langchain.tools import BaseTool
from langchain.callbacks.manager import CallbackManagerForToolRun
from langchain_openai import ChatOpenAI
g_channel_id = None
g_user_id = None
g_current_time = None
class SetReminderSchema(BaseModel):
when: str = Field(description="when to remind")
message: str = Field(default="", description="リマインド時に伝えるメッセージ。ユーモア溢れた感じにする")
class SetReminder(BaseTool):
name = "set_reminder"
description = "Set a reminder. Use this tool if you want to set a reminder."
args_schema: type[BaseModel] = SetReminderSchema
def _run(self,
when: str,
message: str = "",
run_manager: CallbackManagerForToolRun | None = None,
) -> str:
try:
global g_channel_id, g_user_id, g_current_time
channel_id = g_channel_id
user_id = g_user_id
current_time = g_current_time
print(f"リマインダーを設定します。 時間: '{when}' メッセージ: {message}")
current_time_text = current_time.strftime("%Y-%m-%d %H:%M %A")
chat = ChatOpenAI(model="gpt-4")
example_date = datetime.datetime(2024, 4, 9, 15, 0, 0)
example_date_text = example_date.strftime("%Y-%m-%d %H:%M %A")
example_due_date = datetime.datetime(2024, 4, 10, 9, 0, 0)
example_due_date_text = example_due_date.strftime("%Y-%m-%d %H:%M")
print(f"例の時刻: {example_date_text} 求めたい時間: '明日'")
print(f"例の結果: {example_due_date_text}")
print(f"現在時刻: {current_time_text} 求めたい時間: '{when}'")
res = chat.invoke([
SystemMessage(
"次の求めたい時間を '%Y-%m-%d %H:%M' というフォーマットで返してください。" +
"'明日'や'来月'といった感じで特に時刻が明示されていない場合は午前9時としてください。" +
"\n【主な時刻】\n- 朝: 09:00\n- 夕方: 18:00\n- お昼休み: 12:00～13:00"
),
HumanMessage(f"例: 現在時刻: {example_date_text} 求めたい時間: '明日'"),
AIMessage(f"{example_due_date_text}"),
HumanMessage(f"現在時刻: {current_time_text} 求めたい時間: '{when}'"),
])
print(f"AIの応答: {res.content}")
format = "%Y-%m-%d %H:%M"
when_epoch = datetime.datetime.strptime(res.content, format).timestamp()
when_human_readable = datetime.datetime.fromtimestamp(when_epoch).strftime("%Y-%m-%d %A %H:%M")
en_weekdays = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"]
ja_weekdays = ["月", "火", "水", "木", "金", "土", "日"]
for en, ja in zip(en_weekdays, ja_weekdays):
when_human_readable = when_human_readable.replace(en, ja)
with open("reminders.txt", "a") as f:
obj = {
"when": when_epoch,
"message": message,
"channel_id": channel_id,
"user_id": user_id,
}
import json
f.write(json.dumps(obj, ensure_ascii=False) + "\n")
print(f"reminders.txtに追記しました。: {obj}")
return {"when": when, "message": message}
except Exception as e:
print(f"SetReminderでエラーが発生しました。{e}")
return "ERROR! " + str(e)
import langchain
langchain.verbose = True
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI
tools = [SetReminder()]
chat = ChatOpenAI(model="gpt-4")
prompt = ChatPromptTemplate.from_messages([
MessagesPlaceholder("chat_history"),
MessagesPlaceholder("agent_scratchpad"),
])
agent = create_openai_tools_agent(chat, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, max_iterations=5)
g_channel_id = "..."
g_user_id = "..."
g_current_time = datetime.datetime.now()
res = agent_executor.invoke({"chat_history": [
SystemMessage("必ず関西弁で受け答えしてください。"),
AIMessage("おおきに。なんの用や？"),
HumanMessage("30分後にリマインダーを設定して。"),
]})
print(res["output"])
途中経過:
> Entering new AgentExecutor chain...
Invoking: `set_reminder` with `{'when': '30 minutes later', 'message': 'おほん！30分たったで！なにかせなアカンことあったんやったら、そろそろ動き出すんやで！'}`
リマインダーを設定します。 時間: '30 minutes later' メッセージ: おほん！30分たったで！なにかせなアカンことあったんやったら、そろそろ動き出すんやで！
例の時刻: 2024-04-09 15:00 Tuesday 求めたい時間: '明日'
例の結果: 2024-04-10 09:00
現在時刻: 2024-04-13 13:15 Saturday 求めたい時間: '30 minutes later'
AIの応答: 2024-04-13 13:45
reminders.txtに追記しました。: {'when': 1712983500.0, 'message': 'おほん！30分たったで！なにかせなアカンことあったんやったら、そろそろ動き出すんやで！', 'channel_id': '...', 'user_id': '...'}
{'when': '30 minutes later', 'message': 'おほん！30分たったで！なにかせなアカンことあったんやったら、そろそろ動き出すんやで！'} ...
> Finished chain.
結果:
30分後にリマインダー設定したで！"おほん！30分たったで！なにかせなアカンことあったんやったら、そろそろ動き出すんやで！"っていうメッセージで教えるわ！
あとは、定期的にreminders.txtを見て、通知時刻を過ぎているものがあれば適宜通知するだけです。以下はscheduleを使った雑な実装例です。
import threading
import time
import schedule
import json
import datetime
from slack_bolt import App
app = App(token="...")
def job():
print("リマインダーを確認します。")
lines = []
print(f"reminder.txtを読み込ます。")
with open("reminders.txt", "r") as f:
for line in f:
lines.append(line)
print(f"reminder.txtを読み込みました。")
original_count = len(lines)
print(f"リマインダーの数 {len(lines)}")
for line in lines:
obj = json.loads(line)
when = obj["when"]
message = obj["message"]
channel_id = obj["channel_id"]
user_id = obj["user_id"]
when_human_readable = datetime.datetime.fromtimestamp(when).strftime("%Y-%m-%d %H:%M")
now = datetime.datetime.now().timestamp()
if now > when:
print(f"リマインダーを通知します。 {when_human_readable} {message}")
app.client.chat_postMessage(
icon_emoji="exclamation",
channel=channel_id,
text=f"<@{user_id}>\nリマインダーです。\n時間: {when_human_readable}\n内容: `{message}`")
lines.remove(line)
if original_count != len(lines):
print(f"reminder.txtを更新します。")
with open("reminders.txt", "w") as f:
for line in lines:
f.write(line)
print(f"reminder.txtを更新しました。")
def start_watch(use_another_thread: bool):
print("リマインダーを監視します。")
schedule.every(30).seconds.do(job)
if use_another_thread:
t = threading.Thread(target=watch)
t.start()
print("監視スレッドを開始しました。")
else:
watch()
def watch():
while True:
schedule.run_pending()
time.sleep(1)
if __name__ == "__main__":
start_watch(use_another_thread=False)
参考
以上です。
