📌
【LangChain】prompt、chain、structured_outputなどの書き方まとめ
chain
基本形
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from dotenv import load_dotenv
import os
load_dotenv()
#1. プロンプトの作成
template = """以下の料理のレシピを考えてください。
料理名:{dish}
"""
prompt = PromptTemplate.from_template(template)
#2. モデルを指定
llm = ChatOpenAI(model="gpt-4o-mini")
#3. アウトプットを指定
output_parser = StrOutputParser()
#4. chainを作成
chain = prompt | model | output_parser
#5. 実行
response = chain.invoke({"dish": "カレーライス"})
プロンプトをメッセージから作成
from langchain_core.messages import SystemMessage, AIMessage, HumanMessage
from langchain_core.prompts import PromptTemplate, ChatPromptTemplate
#1. プロンプトの作成
messages = [
SystemMessage("ユーザーが入力した料理のレシピを考えてください。"),
HumanMessage("{dish}"),
]
prompt = ChatPromptTemplate.from_messages(messages)
structured_outputで出力の型を指定する
from pydantic import BaseModel, Field
class Recipe:
ingredients: list[str] = Field(description="材料")
steps: list[str] = Field(description="手順")
chain = prompt | model.with_structured_output(Recipe)
chain.invoke({"dish": "カレーライス"})
ストリーミング出力
#2. モデルを指定
llm = ChatOpenAI(model="gpt-4o-mini", streaming=True)
#5. 実行
for token in chain.stream({"dish": "カレーライス"}):
print(token)
if token == "EOS":
break
結果
カ
レー
ライ
ス
の
レ
シ
ピ
を
以下
に
ご
紹介
します
ストリーミング出力(コールバックを指定)
import queue
from typing import Any, Optional
from langchain_core.callbacks import BaseCallbackHandler
#6 ストリームハンドラーの作成
class StreamHandler(BaseCallbackHandler):
def __init__(self):
self.text_queue: queue.Queue[Optional[str]] = queue.Queue()
self.current_chunk = ""
def on_llm_new_token(self, token: str, **kwargs: Any) -> None:
"""新しいトークンを受信するたびに呼ばれる"""
self.current_chunk += token
# 句読点や特定の文字で区切る
if any(char in self.current_chunk for char in ['。', '、', '!', '?']):
if self.current_chunk.strip(): # 空白文字のみの場合は無視
self.text_queue.put(self.current_chunk.strip())
self.current_chunk = ""
def on_llm_end(self, response, **kwargs) -> None:
"""生成完了時に呼ばれる"""
if self.current_chunk and self.current_chunk.strip():
self.text_queue.put(self.current_chunk.strip())
self.text_queue.put(None) # 終了マーカー
#7 ストリームハンドラーをインスタンス化
stream_handler = StreamHandler()
#8 コールバックハンドラーにストリームハンドラーを設定して実行
res = chain.invoke({"dish":"カレーライス"}, callback_handler=stream_handler)
Discussion