会話履歴

2025/01/30 2025/01/30

RunnableWithMessageHistoryによるrunnableのラップ

あるrunnableであるsome_runnableがあるとき，会話履歴を組み込むには次の形にする．

from langchain_core.runnables.history import RunnableWithMessageHistory

with_message_history = RunnableWithMessageHistory(
    runnable=some_runnable,
    get_session_history=get_session_history,
    input_messages_key="input",
    history_messages_key="history",
    output_messages_key="output",
)

with_message_history.invoke(
    {"input": input},
    config={"configurable": {"session_id": session_id}}
)

RunnableWithMessageHistoryは引数にとったrunnableをラップする造りである．

プロンプト

promptを引数に持ったsome_runnableでは，promptを呼び出すときにhistoryを組み入れる．

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "日本語で応答するアシスタントです。"),
        MessagesPlaceholder(variable_name="history"),
        ("human", "{input}"),
    ]
)

get_session_history

from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

store = {}

def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

全履歴をとり続けるとトークンを浪費するので，ChatMessageHistoryのところは改変の必要があるかもしれない．

コメント・シェア

SQLエージェントのサンプル

2025/01/28 2025/01/28

SQLiteのサンプル．
まず，データベースに接続を行う

from langchain_community.utilities import SQLDatabase
from sqlalchemy import create_engine

engine = create_engine("sqlite:///データベース.db")
db = SQLDatabase(engine=engine)

ここで，llm = ChatOpenAI(model = model, temperature = temperature)が与えられているとして，

from langchain_community.agent_toolkits import create_sql_agent

agent_executor = create_sql_agent(llm=llm, db=db, agent_type="tool-calling", verbose=True)
agent_executor.invoke({"input": "質問文"})

とすると，SQLを生成して推論を行う．

コメント・シェア

FewShot

2025/01/28 2025/01/28

FewShotでは以下のような様式で若干の応答例を与える．

from langchain_core.prompts import FewShotChatMessagePromptTemplate, ChatPromptTemplate

example_prompt = ChatPromptTemplate.from_messages(
    [
        ("human", {input}),
        ("ai", {output})
    ]
)

examples = [
    {"input": "入力例1", "output": "出力例1"},
    {"input": "入力例2", "output": "出力例2"},
]

few_shot_prompt = FewShotChatMessagePromptTemplate(
    example_prompt=example_prompt,
    examples=examples,
)

以上の準備の上で，実際の問いかけを行うプロンプトで次のようにする．

prompt = ChatPromptTemplate.from_messages(
    [
        few_shot_prompt,
        # 以下ZeroShotと同様
    ]
)

コメント・シェア

LangGraphのサンプル

2025/01/04 2025/01/04

準備

import operator
from typing import Annotated, Any
from langchain_core.messages import SystemMessage, HumanMessage, BaseMessage

class State(BaseModel):
    """ステートの雛形"""
    query: str = Field(default = "", description = "ユーザーの質問")
    messages: Annotated[list[BaseMessage], operator.add] = Field(default = [])

def state_llm(state: State) -> dict[str, Any]:
    """ノードである状態の一つ"""
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.4)
    message = llm.invoke(state.messages)
    return {"messages": [message]}

from langgraph.graph import StateGraph

graph = StateGraph(State)

ノードの追加

graph.add_node("ノードの名前", function_name_as_node)

エッジの追加

graph.add_edge("ソースとなるノード名", "ターゲットとなるノード名")

始点と終点

graph.set_entry_point("ノード名")
# または
# from langgraph.graph import START
# graph.add_edge(START, "ノード名")

from langgraph.graph import END

graph.add_edge("ソースとなるノード名", END)

分岐

graph.add_conditional_edges(
    "ノード名",
    lambda state: ブール値,
    {True: "ノードT", False: "ノードF"}
)

コンパイルと実行

compiled_graph = graph.compile()

initial_state = State(query = "クエリ")
result = compiled.invoke(initial_state)

コメント・シェア

チェックポイントのサンプル

2025/01/04 2025/01/04

from langgraph.graph import StateGraph
from langgraph.checkpoint.memory import MemorySaver

graph = StateGraph(State)

# STARTからENDまでのグラフを作成

checkpointer = MemorySaver()
compiled_graph = graph.compile(checkpointer = checkpointer)

config = {"configurable": {"thread_id": "student-1"}}
query_1 = State(query = "クエリ1")
response_1 = compiled_graph.invoke(query_1, config)

#config = {"configurable": {"thread_id": "student-2"}}
query_2 = State(query="クエリ2")
response_2 = compiled_graph.invoke(query_2, config)

ここで，コメントアウトしているコンフィグの切り替えがない場合，同一のスレッドIDなのでresponse_1とresponse_2では記憶が引き継がれた回答を得る．
もしアンコメントすれば，スレッドが別れるので，記憶は引き継がれない．

コメント・シェア

RAGの応用例

2025/01/01 2025/01/01

HyDE

質問に対するひとまずの回答をプロンプト | モデル | パーサで応答させ，これを仮説とみなす．
その質問をRunnablePassthroughで素通りさせたRunnableParallelを用意すれば，先の仮説チェーンを検索器に渡したコンテキストと当初の質問が対になり，仮説的思考が行われたと言える．

検索クエリの複数化

質問文を与えて検索クエリを複数生成するというチェーンを作る．
このチェーンを検索器.map()メソッドに連結すると，複数の検索結果を得る．

検索結果は何らかの方法で並び替えたり，絞り込むことも有用．

複数の検索器

異なる検索器を用意してルーティングを行ったり，類似度の評価方法が異なる検索器での結果を比較させることができる．

コメント・シェア

LangChainのプロンプトのサンプル

2024/12/30 2024/12/30

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages(
    [
        # {"キー": "バリュー"}で穴埋めするとメッセージ部分になる
        # ("system", "メッセージ"), ("user", "メッセージ")などが入る
    ]
)

prompt_value = prompt.invoke({"キー": "バリュー"})

model = ChatOpenAI(model="gpt-4o-mini", temperature = 0)
ai_message = model.invoke(prompt_value)
print(ai_message)

メッセージ部分

穴埋めされて変換された後のメッセージ部分は次のような形のイメージ

from langchain_core.messages import SystemMessage, HumanMessage, AIMessage

messages = [
    SystemMessage("文章"),
    HumanMessage("文章"),
    AIMessage("文章"),
    HumanMessage("文章"),
]

ストリーミング出力

printの行を次に差し替え．

for chunk in model.stream(messages):
    print(chunk.content, end = "", flush = True)

履歴を組み込む場合

MessagesPlaceholderをインポートして，メッセージに履歴となる部分を挿入する．
_

from langchain_core.prompts import MessagesPlaceholder

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "システムへのメッセージ"),
        MessagesPlaceholder("chat_history", optional = True),
        ("human", "{input}")
    ]
)

キーとしたchat_history でインボークする．

prompt_value = prompt.invoke(
    {
        "chat_history": [
            HumanMessage("文章"),
            AIMessage("応答"),
        ],
        "input": "質問文",
    }
)

コメント・シェア

LangChainを使用する

2024/12/30 2024/12/30

パイプによりチェーンを連結するサンプル

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_messages(
    # メッセージ
)

model = ChatOpenAI(model="gpt-4o-mini", temperature = 0)

chain = prompt | model | StrOutputParser()
output = chain.invoke(メッセージの穴埋め部)
print(output)

戻り値の型を定める

次のようにクラスを作って，それを埋めるチェーンを作ることができる．

from pydantic import BaseModel, Field

class SampleCls(BaseModel):
    arg1: list[str] = Field(descriptions = "変数1の説明")
    arg2: list[str] = Field(descriptions = "変数2の説明")

chain = prompt | model.with_structured_output(SampleCls)

コメント・シェア

RAGを使う

2024/12/30 2024/12/30

from langchain_community.document_loaders import GitLoader

def file_filter(file_path: str) -> bool:
    """拡張子txtのファイルのみを抽出するためのフィルタ"""
    return file_path.endswith(".txt")

loader = GitLoader(
    clone_url = "クローン元のURL",
    repo_path = "クローン先のパス",
    branch = "ブランチ名",
    file_filter = file_filter,
)

ここで，

docs = loader.load()

とするとフィルタで絞り込まれた全文書が返る．

ドキュメントの分割

from langchain_text_splitters import CharacterTextSplitter

text_splitter = CharacterTextSplitter(chunk_size = 1000, chunk_overlap = 0)
split_docs = text_splitter.split_documents(docs)

ドキュメントのベクトル化

from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
db = Chroma.from_documents(split_docs, embeddings)

ドキュメントの検索

retriever = db.as_retriever()
context_docs = retriever.invoke("クエリ")

以上で，クエリのベクトルの類似度が高い文脈を掬い上げる．

コメント・シェア

Runnable

2024/12/30 2024/12/30

チェーンの要素は抽象基底クラスRunnableを継承している．

並列で繋ぐものの1つをRunnablePassthroughにしておくと，入力値を次工程に引き継げます．
例えば，ある質問を入力したとき，RunnablePassthroughと検索を並列にすると，次工程へ元の質問文と検索結果をペアで渡すことができます．

渡すものと出力を両方保持して最終出力としたい場合は，RunnablePassthrough.assignメソッドが使えます．

chain = 入力 | RunnablePassthrough.assign(answer = 出力用チェーン)

とすると，実行結果は出力用チェーンからの出力結果と入力をあわせた辞書を出力します．

コメント・シェア

openaiモジュールとtoolcall

2024/12/28 2024/12/28

tools = [
    {
    	"type": "function",
    	"function": {
    	    "name": "some_function",
    	    "description": "何に使う関数であるかを説明する文章",
    	    "parameters":{
    	        "type": "object",
    	        "properties": {
    	            "arg1": {
    	                "type": "string",
    	                "description": "引数1についての説明文",
    	            },
    	            "arg2": {
    	                "type": "string",
    	                "description": "引数2についての説明文",
    	            },
    	        },
    	        "required": ["arg1"],
    	    },
    	},
    }
]

実行してツール要望のレスポンスを得る

tools を加えたときのレスポンスをまず得る．

from openai import OpenAI

client = OpenAI()
model = "gpt-4o-mini"
messages = [{"role": "user", "content": "何らかの質問"}]

first_response = client.chat.completions.create(
    model = model,
    messages = messages,
    tools = tools,
)

ここで，first_response にてツールを使いたいtool_callsというレスポンスが返ってくるものとする．
このレスポンスをメッセージに加えておく．

first_response_message = first_response.choices[0].message
messages.append(first_response_message.to_dict())

ツール使用結果を渡す

次に，callされた関数群のレスポンスをメッセージに加える．

import json

available_functions = {
    "some_function": some_function,
}

for tool_call in response_message.tool_calls:
    function_name = tool_call.function.name
    function_to_call = available_functions[function_name]
    function_args = json.loads(tool_call.function.arguments)
    function_response = function_to_call(
        arg1 = function_args.get("arg1"),
        arg2 = function_args.get("arg2"),
    )
    
    messages.append(
        {
            "tool_call_id": tool_call.id,
            "role": "tool",
            "name": function_name,
            "content": function_response,
        }
    )

最終結果の取得

以上のtool使用後の結果を含んだメッセージを再度処理にかけてレスポンスを

second_response = client.chat.completions.create(
    model = model,
    messages = messages,
)

として得て，

print(second_response.to_json(indent=2))

などで出力するとよい．

コメント・シェア

openaiモジュールのチャットサンプル

2024/12/28 2024/12/28

from openai import OpenAI

client = OpenAI()
model = "gpt-4o-mini"

response = client.chat.completions.create(
    model = model,
    messages = [
        {"role": "system", "content": "AIがどのような役割を担っているか伝える文章を書く"},
        {"role": "user", "content": "ユーザーのメッセージ"},
        {"role": "assistant", "content": "AIの返答"},
        {"role": "user", "content": "ユーザーのメッセージ"},
    ],
)

print(response.choices[0].message.content)

チャットのストリーミング出力

次のようにstream = Trueとする．

response = client.chat.completions.create(
    model = model,
    messages = [
        # メッセージ
    ],
    stream = True
)

printは次のように変更する．

for r in response:
    content = r.choices[0].delta.content
    if content is not None:
        print(content, end = "", flush = True)

出力形式をJSONにする

次のようにresponse_format = {"type": "json_object"}とする．

response = client.chat.completions.create(
    model = model,
    messages = [
        # メッセージ
    ],
    response_format = {"type": "json_object"}
)

URLリンク先の画像を入力に用いる

次のように{"type": "image_url", "image_url": {"url": image_url}}を与える．

response = client.chat.completions.create(
    model = model,
    messages = [
        {
            "role": "user", 
            "content": [
                {"type": "text", "text": "画像を説明してください。"},
                {"type": "image_url", "image_url": {"url": image_url}},
            ],
        }
    ],
)

コメント・シェア

RunnableWithMessageHistoryによるrunnableのラップ

プロンプト

get_session_history

準備

ノードの追加

エッジの追加

始点と終点

分岐

コンパイルと実行

HyDE

検索クエリの複数化

複数の検索器

目次

基本構成

メッセージ部分

ストリーミング出力

履歴を組み込む場合

パイプによりチェーンを連結するサンプル

戻り値の型を定める

目次

ドキュメントのロード

ドキュメントの分割

ドキュメントのベクトル化

ドキュメントの検索

目次

チェーンの連結

関数をRunnableにする

チェーンの並列繋ぎ

RunnablePassthrough

目次

呼び出す予定の関数を定義

ツールの定義

実行してツール要望のレスポンスを得る

ツール使用結果を渡す

最終結果の取得

目次

単純なサンプル

チャットのストリーミング出力

出力形式をJSONにする

URLリンク先の画像を入力に用いる

ねこせんせい