📌

【LangChain】prompt、chain、structured_outputなどの書き方まとめ

2025/01/20に公開

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