PythonでLangChain書いてると、LCELの使い方で最初に悩みました。チュートリアルに書かれている内容から外れた場合、どう書けばいいのかわからなかったです。
下記チートシートを参考にLCELでのRunnableのつなげ方を理解しました。
環境
Python3.12.2 で以下のパッケージを使っています(Jupyterは省略)。
| Package | Version | 備考 |
|---|---|---|
| langchain | 0.2.6 | |
| langchain-core | 0.2.10 | |
| pygraphviz | 1.13 | フローの画像描写するなら必要 |
pygraphvizを使うためにbrew install graphvizでgraphvizをMacにインストールしています
あまり整理していない仮想環境なので必要なパッケージ漏れているかも
つなげ方
1. 単一
単一 Runnable なので接続はしないのですが、基本から始めます。
1.1. RunnableLambda(lambda)
最も簡単なRunnableLambdaでlambdaを使った場合です。実務であまり使わないような気もしますが、今回みたいな動きを確認する場合には重宝します。
from langchain_core.runnables import RunnableLambda
runnable = RunnableLambda(lambda x: x+1)
runnable.invoke(5)
当然、5+1 の6が出力されます。
出力
6
1.2. RunnableLambda(functin)
RunnableLambdaで関数を使った場合です。こっちの方がlambdaよりは実用的。
def length_function(text):
return len(text)
RunnableLambda(length_function).invoke("aaa")
"aaa"の長さの3が出力
出力
3
複数変数を渡したい場合には、辞書型にする。
def length_function2(dict):
return len(dict['var1']), len(dict['var2'])
RunnableLambda(length_function2).invoke({"var1": "aa", "var2": "aaa"})
出力
(2, 3)
1.3. RunnablePassthrough
RunnablePassthroughは、ただそのまま変数を渡すだけ。
最初使い方がよく理解できなかったが、後で実例示します。
from langchain_core.runnables import RunnablePassthrough
RunnablePassthrough().invoke("a")
出力
'a'
2. 直列
関数draw_asciiを使って、ChainをAsciiで出力します。ついでにChainの実行結果も出力します。
test_chain
#from IPython.display import Image
from langchain_core.runnables import RunnableSequence
def test_chain(chain: RunnableSequence, input: any) -> None:
result = chain.invoke(input)
print(result)
print(chain.get_graph().draw_ascii())
#display(Image(chain.get_graph().draw_png()))
上記の関数test_chain内でコメントアウトしている関数draw_pngを使うとこんな画像が出ます。たまに失敗する場合があり、理由調べるのも面倒だったので、当投稿ではdraw_asciiに1本化しています。
関数draw_mermaid_pngは、軽く試したけどエラーが出たので諦めました。大して調べてないですが、バグ？かと思いました。
2.1. シンプルな直列
|で Runnable をつなぎます。
from langchain_core.prompts import PromptTemplate
prompt_template = PromptTemplate.from_template("variable is {var1}")
chain = RunnableLambda(lambda x: x+1) | prompt_template
chain.invoke(5).text
test_chain(chain, 5)
出力
text='variable is 6'
+-------------+
| LambdaInput |
+-------------+
*
*
*
+-------------------------+
| Lambda(lambda x: x + 1) |
+-------------------------+
*
*
*
+----------------+
| PromptTemplate |
+----------------+
*
*
*
+----------------------+
| PromptTemplateOutput |
+----------------------+
2.2. itemgetter
直列で必要というよりは、次の Runnable に変数渡すための関数。辞書型のKeyの値を抽出して次の処理に渡します。下の例だとただ、Keyを切り替えているだけ。
Chainが複雑な場合に使います。
from operator import itemgetter
def concat_function(dict):
return dict['text1'] + ' : ' + dict['text2']
chain = {"text1": itemgetter("a"), "text2": itemgetter("b")} \
| RunnableLambda(concat_function)
print(chain.invoke({"a": "z", "b": "y"}))
test_chain(chain)
出力
z : y
+----------------------------+
| Parallel<text1,text2>Input |
+----------------------------+
***** *****
**** ****
*** ***
+-------------------------+ +-------------------------+
| Lambda(itemgetter('a')) | | Lambda(itemgetter('b')) |
+-------------------------+ +-------------------------+
***** *****
**** ****
*** ***
+-----------------------------+
| Parallel<text1,text2>Output |
+-----------------------------+
*
*
*
+-------------------------+
| Lambda(concat_function) |
+-------------------------+
*
*
*
+------------------------+
| concat_function_output |
+------------------------+
3. 並列
3.1. 辞書型
Chain組むときに辞書型にすると並列処理してくれます。
chain = {"var1": RunnableLambda(lambda x: str(x + 1)),
"var2": RunnableLambda(lambda x: str(x + 2))} \
| RunnableLambda(lambda x: x["var1"] + " : " + x["var2"])
test_chain(chain, 3)
出力
4 : 5
+--------------------------+
| Parallel<var1,var2>Input |
+--------------------------+
*** ***
** **
** **
+-------------+ +-------------+
| Lambda(...) | | Lambda(...) |
+-------------+ +-------------+
*** ***
** **
** **
+---------------------------+
| Parallel<var1,var2>Output |
+---------------------------+
*
*
*
+-------------+
| Lambda(...) |
+-------------+
*
*
*
+--------------+
| LambdaOutput |
+--------------+
並列確認のために、下記のChainを実行。5秒で処理が終わり、並列になっていることがわかります。
import time
chain = {"var1": RunnableLambda(lambda x: time.sleep(5)),
"var2": RunnableLambda(lambda x: time.sleep(5))} \
| RunnablePassthrough()
chain.invoke("")
3.2. RunnableParallel
RunnableParallelを使って並列化。なんとなく辞書型の方が好み。
rom langchain_core.runnables import RunnableParallel
runnable1 = RunnableLambda(lambda x: {"foo": x})
runnable2 = RunnableLambda(lambda x: x * 2)
chain = RunnableParallel(first=runnable1, second=runnable2)
test_chain(chain, 2)
出力
{'first': {'foo': 2}, 'second': 4}
+-----------------------------+
| Parallel<first,second>Input |
+-----------------------------+
**** ****
***** *****
*** ***
+------------------------------+ +-------------------------+
| Lambda(lambda x: {'foo': x}) | | Lambda(lambda x: x * 2) |
+------------------------------+ +-------------------------+
**** ****
***** *****
*** ***
+------------------------------+
| Parallel<first,second>Output |
+------------------------------+
3.3. InputとOutputのマージ
Inputに対して処理しないで次にそのまま渡すのと処理するのとで分ける。使いそうな気がする。
runnable = RunnableLambda(lambda x: x["a"] + 7)
chain = RunnablePassthrough.assign(b=runnable)
test_chain(chain, {"a": 10})
出力
{'a': 10, 'b': 17}
+------------------+
| Parallel<b>Input |
+------------------+
**** ****
*** ***
** **
+------------------------------+ +-------------+
| Lambda(lambda x: x['a'] + 7) | | Passthrough |
+------------------------------+ +-------------+
**** ****
*** ***
** **
+-------------------+
| Parallel<b>Output |
+-------------------+
4. 分岐
分岐。使ったことはないが、使いそうな気がする。
runnable1 = RunnableLambda(lambda x: {"a": x})
runnable2 = RunnableLambda(lambda x: [x] * 2)
chain = RunnableLambda(lambda x: runnable1 if x > 6 else runnable2)
print(chain.invoke(7))
print(chain.invoke(6))
print(chain.get_graph().draw_ascii())
フロー上は分岐が表現されないのが少し残念。
出力
{'a': 7}
[6, 6]
+-------------+
| LambdaInput |
+-------------+
*
*
*
+---------------------------------------+
| Lambda(lambda x: runnable1 if x > ... |
+---------------------------------------+
*
*
*
+--------------+
| LambdaOutput |
+--------------+
確認のためのサンプル
サンプルプログラムを作って、LCELの動きを確認します。
人物についてのQAタスクを処理します。WikipediaとDuckDuckGoから人物の情報を取得して最後に要約をします。
以下がその流れです。
+--------------------------+
| Parallel<wiki,ddgs>Input |
+--------------------------+
*** ****
**** ***
** ****
+-------------+ **
| Passthrough | *
+-------------+ *
* *
* *
* *
+--------------------+ *
| ChatPromptTemplate | *
+--------------------+ *
* *
* *
* *
+------------+ *
| ChatOpenAI | *
+------------+ *
* *
* *
* *
+---------------------+ *
| PydanticToolsParser | *
+---------------------+ *
* *
* *
* *
+-------------+ +---------------------------------------+
| Lambda(...) | | Lambda(lambda query: DDGS().chat(q... |
+-------------+** +---------------------------------------+
*** ***
**** ****
** **
+---------------------------+
| Parallel<wiki,ddgs>Output |
+---------------------------+
*
*
*
+--------------------+
| ChatPromptTemplate |
+--------------------+
*
*
*
+------------+
| ChatOpenAI |
+------------+
*
*
*
+------------------+
| ChatOpenAIOutput |
+------------------+
環境
Python3.12.2 で以下のパッケージを使っています(Jupyterは省略)。
| Package | Version | 備考 |
|---|---|---|
| duckduckgo_search | 6.1.5 | |
| langchain | 0.2.6 | |
| pygraphviz | 1.13 | フローの画像描写するなら必要 |
| langchain-core | 0.2.10 | |
| langchain-openai | 0.1.13 | |
| wikipedia | 1.4.0 |
プログラム
1. Package Import
from typing import Optional
from duckduckgo_search import DDGS
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage
from langchain_core.prompts import ChatPromptTemplate, HumanMessagePromptTemplate
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_core.runnables import RunnableLambda, RunnablePassthrough, RunnableSequence
from langchain_core.runnables.base import Other
from langchain_community.tools import WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper
2. Chain確認
Chainを確認するための関数です。
def test_chain(chain: RunnableSequence, input: any) -> Other:
result = chain.invoke(input)
print(result)
print(chain.get_graph().draw_ascii())
return result
3. 人物抽出のクラス
固有表現抽出で人物をLLMで抽出するためのクラス。
前に以下の記事で試しました。
class Person(BaseModel):
"""Information about a person."""
name: Optional[str] = Field(default=None,
description="The name of the person")
4. Wikiからの情報取得
Wikipediaから情報取得するChainです。後でChainを結合します。
Wikipediaは自然文で検索できないので、その前に人物を抽出してから渡しています。
def get_from_wiki() -> RunnableSequence:
prompt_ner = ChatPromptTemplate.from_messages(
[
SystemMessage(content=
"You are an expert extraction algorithm. "
"Only extract relevant information from the text. "
"If you do not know the value of an attribute asked to extract, "
"return null for the attribute's value.",
),
HumanMessagePromptTemplate.from_template("{text}"),
]
)
model = ChatOpenAI(model="gpt-3.5-turbo")
wikipedia = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper(lang='ja',
top_k_results=1))
chain = {"text": RunnablePassthrough()} \
| prompt_ner \
| model.with_structured_output(schema=Person) \
| RunnableLambda(lambda person: wikipedia.run(person.name))
return chain
wiki_chain = get_from_wiki()
wiki_result = test_chain(wiki_chain, "バイデンについて教えて")
出力
Page: ジョー・バイデン
Summary: ジョー・バイデン（英語: Joe Biden、発音: [dʒoʊ ˈbaɪdən] ( 音声ファイル)）、本名ジョセフ・ロビネット・バイデン・ジュニア（Joseph Robinette Biden Jr.、1942年11月20日 - ）は、アメリカ合衆国の政治家、弁護士。同国第46代大統領（在任: 2021年1月20日 - ）。現在、史上最年長にしてアメリカ合衆国史上最高齢の大統領である。
民主党に所属し、ニューキャッスル郡議会議員、デラウェア州選出連邦上院議員、副大統領を歴任した後、2021年1月20日に78歳で大統領に就任した。ジョン・F・ケネディ以来2人目のカトリックの大統領であり、ジェームズ・ブキャナン以来2人目のペンシルベニア州出身の大統領でもある。
+---------------------+
| Parallel<text>Input |
+---------------------+
*
*
*
+-------------+
| Passthrough |
+-------------+
*
*
*
+--------------------+
| ChatPromptTemplate |
+--------------------+
*
*
*
+------------+
| ChatOpenAI |
+------------+
*
*
*
+---------------------+
| PydanticToolsParser |
+---------------------+
*
*
*
+-------------+
| Lambda(...) |
+-------------+
*
*
*
+--------------+
| LambdaOutput |
+--------------+
5. DuckDuckGo から情報取得
DuckDuckGo から情報取得するRunnable。 DuckDuckGoは自然文でも検索できます(精度はあまり確認したことない)。
ddgs = RunnableLambda(lambda query: DDGS().chat(query, model='claude-3-haiku'))
ddgs_result = ddgs.invoke("バイデンについて教えて")
print(ddgs_result)
出力
はい、バイデン大統領についてお話しましょう。バイデン大統領は2021年1月に就任し、現在アメリカ合衆国の第46代大統領を務めています。主な政策としては、新型コロナウイルス対策、気候変動対策、インフラ投資などが挙げられます。また、外交面では同盟国との関係強化や中国との競争関係への対応などに取り組んでいます。ただし、政治的には共和党との対立も見られ、政策実現には難しい面もあるようです。バイデン大統領の評価については、賛否両論があるのが現状だと言えるでしょう。
6. 要約
Wikipedia と DuckDuckGo から取得したテキストを要約させます。
def summarize() -> RunnableSequence:
prompt_summary = ChatPromptTemplate.from_messages(
[
SystemMessage(content="You are an expert summarizer and analyzer who can help me."),
HumanMessagePromptTemplate.from_template(
"以下のテキストを日本語で簡潔に100文字程度で要約してください。"
"CONTEXT:{wiki}"
"{ddgs}"
"SUMMARY:"),
]
)
chain = prompt_summary | ChatOpenAI(model="gpt-3.5-turbo")
return chain
summary_chain = summarize()
summary_input = {
"wiki": wiki_result,
"ddgs": ddgs_result,
}
test_chain(summary_chain, summary_input)
出力
content='ジョー・バイデンはアメリカ合衆国の第46代大統領で、民主党所属。新型コロナウイルス対策や気候変動対策を重視し、外交面では同盟国関係強化や中国との競争に注力。共和党との対立もあり、評価は賛否両論。' response_metadata={'token_usage': {'completion_tokens': 117, 'prompt_tokens': 657, 'total_tokens': 774}, 'model_name': 'gpt-3.5-turbo-0125', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None} id='run-7c30aec4-d9b4-424e-8228-b2ad13b4ea53-0' usage_metadata={'input_tokens': 657, 'output_tokens': 117, 'total_tokens': 774}
+-------------+
| PromptInput |
+-------------+
*
*
*
+--------------------+
| ChatPromptTemplate |
+--------------------+
*
*
*
+------------+
| ChatOpenAI |
+------------+
*
*
*
+------------------+
| ChatOpenAIOutput |
+------------------+
7. 全体Chain化
全体を一つのChainにします。
chain = {"wiki": wiki_chain, "ddgs": ddgs } \
| summary_chain
test_chain(chain, "バイデンについて教えて")
出力
content='ジョー・バイデンは2021年1月にアメリカ合衆国の第46代大統領に就任。新型コロナウイルス対策や気候変動対策、インフラ投資を推進。同盟国との関係強化や中国との競争にも注力。共和党との対立もあり、評価は賛否両論。' response_metadata={'token_usage': {'completion_tokens': 122, 'prompt_tokens': 657, 'total_tokens': 779}, 'model_name': 'gpt-3.5-turbo-0125', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None} id='run-df8b4324-30e4-4015-8c46-8657290f699e-0' usage_metadata={'input_tokens': 657, 'output_tokens': 122, 'total_tokens': 779}
+--------------------------+
| Parallel<wiki,ddgs>Input |
+--------------------------+
*** ****
**** ***
** ****
+-------------+ **
| Passthrough | *
+-------------+ *
* *
* *
* *
+--------------------+ *
| ChatPromptTemplate | *
+--------------------+ *
* *
* *
* *
+------------+ *
| ChatOpenAI | *
+------------+ *
* *
* *
* *
+---------------------+ *
| PydanticToolsParser | *
+---------------------+ *
* *
* *
* *
+-------------+ +---------------------------------------+
| Lambda(...) | | Lambda(lambda query: DDGS().chat(q... |
+-------------+** +---------------------------------------+
*** ***
**** ****
** **
+---------------------------+
| Parallel<wiki,ddgs>Output |
+---------------------------+
*
*
*
+--------------------+
| ChatPromptTemplate |
+--------------------+
*
*
*
+------------+
| ChatOpenAI |
+------------+
*
*
*
+------------------+
| ChatOpenAIOutput |
+------------------+