エージェントにツールを持たせる

LLMでのツール呼び出し

from langchain_core.tools import tool

@tool
def mytool1():
    # ツール1の処理

@tool
def mytool2():
    # ツール2の処理

llm = ChatOpenAI()
llm_with_tool = llm.bind_tools([mytool1, mytool2])
response = llm_with_tool.invoke(クエリ)

message.append(response)

if response.tool_calls:
    for call in res.tool_calls:
	    val = # 処理
		messages.append(ToolMessage(val, tool_call_id=call["id"]))

response = llm_with_tool.invoke(message)

ツールをもったエージェントの作成

from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.tools import tool

tools = [mytool1, mytool2]
agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools)
response = agent_executor.invoke(クエリ)

コメント・シェア

プロンプトテンプレート

値の確定したプロンプト

from langchain_core.messages import SystemMessage, HumanMessage, AIMessage, ToolMessage

messages = [
    SystemMessage("あなたはしっかり者のAIアシスタントです。")
	HumanMessage("私はいまからあなたに質問をします。")
]

プロンプトテンプレート

プロンプトの一部を可変にするには、プロンプトテンプレートを使う。

from langchain_core.prompts import ChatPromptTemplate, SystemMessagePromptTemplate, MessagePlaceholder, HumanMessagePrompTemplate

ChatPromptTemplate.from_messages(
    [
	    SystemMessagePromptTemplate.from_template("I am a master in {category}."),
		MessagePlaceholder("chat_history")
	    HumanMessagePrompTemplate.from_template("My name is {name}."),
		AIMessage("Hello."),
	]
)

たとえば、キーchat_historyはinvokeして、随時バリューのリストにはメッセージをアペンドする

コメント・シェア

関数定義を委ねる

自然言語で関数の説明だけ与えることができる。

from pyaskit import function

@function(codable=False)
def myfunc(query: str, n: int) -> List[str]:
    """{{query}}というクエリに従って、{{n}}個の文字列を生成する"""

コメント・シェア

会話履歴

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エージェントのサンプル

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

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のサンプル

準備

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)

コメント・シェア

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

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の応用例

HyDE

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

検索クエリの複数化

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

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

複数の検索器

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

コメント・シェア

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

目次

  1. 基本構成
  2. メッセージ部分
  3. ストリーミング出力
  4. 履歴を組み込む場合

基本構成

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を使用する

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

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を使う

目次

  1. ドキュメントのロード
  2. ドキュメントの分割
  3. ドキュメントのベクトル化
  4. ドキュメントの検索

ドキュメントのロード

GitLoaderを使用する例．

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

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

目次

  1. チェーンの連結
  2. 関数をRunnableにする
  3. チェーンの並列繋ぎ
    1. RunnablePassthrough

チェーンの連結

連結部で入出力の型が揃っていれば，次のようにチェーン同士を直列繋ぎにできる．

chain_a = a1 | a2 | a3
chain_b = b1 | b2 | b3
chain = chain_a | chain_b

関数をRunnableにする

独自関数some_funcがあるとして，RunnableLambdaを用いた

from langchain_core.runnables import RunnableLambda

chain = a | b | c | RunnableLambda(some_func)

または，自動変換により単に

chain = a | b | c | some_func

としてチェーンに組み込める．

次のようにデコレータで書いておくことも可能である．

from langchain_core.runnables import chain

@chain
def some_func(args):
    return something

chain = a | b | c | some_func

チェーンの並列繋ぎ

RunnableParallelでチェーンを並列繋ぎにできる．

from langchain_core.runnables import RunnableParallel

chain = (
    RunnableParallel(
        {
    	    "aaa": chain_a,
            "bbb": chain_b,
        }
    )
    | ccc
    | ddd
    | eee
)

RunnablePassthrough

RunnablePassthroughのインポートは次のように行います．

from langchain_core.runnables import RunnablePassthrough

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

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

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

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

コメント・シェア

openaiモジュールとtoolcall

目次

  1. 呼び出す予定の関数を定義
  2. ツールの定義
  3. 実行してツール要望のレスポンスを得る
  4. ツール使用結果を渡す
  5. 最終結果の取得

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

def some_function(arg1, arg2):
    """ツールとして使う関数"""
    # 処理
    	return 戻り値

ツールの定義

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モジュールのチャットサンプル

目次

  1. 単純なサンプル
  2. チャットのストリーミング出力
  3. 出力形式をJSONにする
  4. URLリンク先の画像を入力に用いる

単純なサンプル

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}},
            ],
        }
    ],
)

コメント・シェア

  • page 1 of 1
著者の絵

ねこせんせい

One cat just leads to another.
(Ernest Hemingway)


色鉛筆塗り


校正室