LangChain - How to - Run arbitrary functions

https://python.langchain.com/docs/expression_language/how_to/functions

Run arbitrary functions | 🦜️🔗 Langchain

 

You can use arbitrary functions in the pipeline

 

파이프라인에서 임의의 함수를 사용할 수 있습니다.

 

Note that all inputs to these functions need to be a SINGLE argument. If you have a function that accepts multiple arguments, you should write a wrapper that accepts a single input and unpacks it into multiple argument.

 

이러한 함수에 대한 모든 입력은 SINGLE 인수여야 합니다. 여러 인수를 허용하는 함수가 있는 경우 단일 입력을 허용하고 이를 여러 인수로 압축 해제하는 래퍼를 작성해야 합니다.

 

from langchain.schema.runnable import RunnableLambda
from langchain.prompts import ChatPromptTemplate
from langchain.chat_models import ChatOpenAI
from operator import itemgetter

def length_function(text):
    return len(text)

def _multiple_length_function(text1, text2):
    return len(text1) * len(text2)

def multiple_length_function(_dict):
    return _multiple_length_function(_dict["text1"], _dict["text2"])

prompt = ChatPromptTemplate.from_template("what is {a} + {b}")
model = ChatOpenAI()

chain1 = prompt | model

chain = {
    "a": itemgetter("foo") | RunnableLambda(length_function),
    "b": {"text1": itemgetter("foo"), "text2": itemgetter("bar")} | RunnableLambda(multiple_length_function)
} | prompt | model

chain.invoke({"foo": "bar", "bar": "gah"})

이 코드는 다양한 RunnableLambda 함수와 대화 체인을 사용하여 계산 및 대화 모델을 조합하는 예시를 제공합니다. 

1. RunnableLambda 함수 정의:
   • length_function(text) 함수는 입력 문자열의 길이를 반환합니다.
   • _multiple_length_function(text1, text2) 함수는 두 개의 입력 문자열 길이를 곱한 값을 반환합니다.
   • multiple_length_function(_dict) 함수는 두 개의 입력 문자열 길이를 곱한 값을 반환하기 위해 _multiple_length_function 함수를 호출합니다.
2. 대화 템플릿 및 모델 설정:
   • ChatPromptTemplate.from_template("what is {a} + {b}")를 사용하여 두 변수 "a"와 "b"를 사용하는 대화 템플릿을 설정합니다.
   • ChatOpenAI()를 사용하여 ChatOpenAI 모델을 설정합니다.
   • "chain1"은 "prompt" 템플릿과 "model" 모델을 연결한 대화 체인을 설정합니다.
3. RunnableLambda 함수 및 변수를 사용하여 체인 설정:
   • "chain"은 변수 "a"와 "b"를 설정하는 대화 체인을 생성합니다.
   • "a" 변수는 "foo" 변수에서 문자열을 추출하여 길이를 계산하는 RunnableLambda 함수를 사용하여 설정됩니다.
   • "b" 변수는 "text1"과 "text2" 변수에서 문자열을 추출하여 두 문자열의 길이를 곱한 값을 계산하기 위해 RunnableLambda 함수를 사용하여 설정됩니다.
4. 대화 체인 호출:
   • chain.invoke({"foo": "bar", "bar": "gah"})는 "foo"와 "bar" 변수에 각각 "bar"와 "gah" 문자열을 할당하고, 대화 체인을 호출하여 "a"와 "b" 변수를 설정하고 결과를 반환합니다.

이 코드는 RunnableLambda 함수와 대화 체인을 사용하여 변수 및 함수를 효과적으로 조합하는 방법을 보여줍니다. 설정된 함수 및 변수를 활용하여 대화 체인에서 계산을 수행하고 결과를 얻을 수 있습니다.

 

    AIMessage(content='3 + 9 equals 12.', additional_kwargs={}, example=False)

 

 

Accepting a Runnable Config

 

Runnable lambdas can optionally accept a RunnableConfig, which they can use to pass callbacks, tags, and other configuration information to nested runs.

 

실행 가능한 람다는 콜백, 태그 및 기타 구성 정보를 중첩 실행에 전달하는 데 사용할 수 있는 RunnableConfig를 선택적으로 허용할 수 있습니다.

 

from langchain.schema.runnable import RunnableConfig
from langchain.schema.output_parser import StrOutputParser

import json

def parse_or_fix(text: str, config: RunnableConfig):
    fixing_chain = (
        ChatPromptTemplate.from_template(
            "Fix the following text:\n\n```text\n{input}\n```\nError: {error}"
            " Don't narrate, just respond with the fixed data."
        )
        | ChatOpenAI()
        | StrOutputParser()
    )
    for _ in range(3):
        try:
            return json.loads(text)
        except Exception as e:
            text = fixing_chain.invoke({"input": text, "error": e}, config)
    return "Failed to parse"

from langchain.callbacks import get_openai_callback

with get_openai_callback() as cb:
    RunnableLambda(parse_or_fix).invoke("{foo: bar}", {"tags": ["my-tag"], "callbacks": [cb]})
    print(cb)

이 코드는 RunnableConfig 및 StrOutputParser를 사용하여 문자열을 구문 분석하거나 수정하는 함수를 정의하고 OpenAI 콜백을 사용하여 함수를 호출하는 예시를 제공합니다. 

1. parse_or_fix 함수 정의:
   • parse_or_fix 함수는 두 개의 매개변수 text (문자열)와 config (RunnableConfig)를 가집니다.
   • fixing_chain은 text를 수정하기 위한 대화 체인을 정의합니다. 이 체인은 입력 문자열과 오류 메시지를 받아 수정된 데이터를 반환합니다.
   • "Fix the following text"이라는 시스템 메시지와 입력 문자열 및 오류 메시지를 포함하는 사용자 메시지로 구성된 대화 템플릿을 생성합니다. 그런 다음 ChatOpenAI 모델과 StrOutputParser를 사용하여 오류를 수정하는 체인을 설정합니다.
   • try-except 루프를 사용하여 입력 문자열을 JSON으로 파싱하려고 시도하고 예외가 발생한 경우 수정 체인을 사용하여 3 번 시도합니다. JSON 파싱이 성공하면 파싱된 결과를 반환하고, 세 번의 시도 후에도 성공하지 못하면 "Failed to parse"를 반환합니다.
2. OpenAI 콜백 사용:
   • from langchain.callbacks import get_openai_callback을 사용하여 OpenAI 콜백을 가져옵니다.
   • with get_openai_callback() as cb:를 사용하여 OpenAI 콜백을 설정하고 cb 변수에 할당합니다.
   • RunnableLambda(parse_or_fix).invoke("{foo: bar}", {"tags": ["my-tag"], "callbacks": [cb]})은 "parse_or_fix" 함수를 호출하고 입력 문자열과 구성 옵션을 전달합니다. 또한 "callbacks" 매개변수에 OpenAI 콜백을 추가하여 결과를 OpenAI에 보고합니다.
   • print(cb)는 OpenAI 콜백 객체를 출력하여 OpenAI에서 실행한 작업 및 결과를 확인합니다.

이 코드는 RunnableConfig 및 OpenAI 콜백을 활용하여 문자열을 구문 분석하거나 수정하는 함수를 구현하고, 이를 OpenAI에 보고하는 방법을 보여줍니다.

    Tokens Used: 65
        Prompt Tokens: 56
        Completion Tokens: 9
    Successful Requests: 1
    Total Cost (USD): $0.00010200000000000001