OUTPUT PARSER Archives - 2 중 2 번째 페이지

[PYTHON/LANGCHAIN] yield 키워드를 사용해 스트리밍 커스텀 출력 파서 만들기

■ yield 키워드를 사용해 스트리밍 처리되는 커스텀 출력 파서를 만드는 방법을 보여준다. ▶ main.py


import asyncio
import os

from langchain_openai              import ChatOpenAI
from langchain_core.output_parsers import JsonOutputParser

async def getCountryNameGenerator(inputAsyncGenerator):
    """입력 스트림에서 작동하는 함수이다."""
    countryNameSet = set()
    async for inputDictionary in inputAsyncGenerator:
        if not isinstance(inputDictionary, dict):
            continue
        if "countries" not in inputDictionary:
            continue
        countryList = inputDictionary["countries"]
        if not isinstance(countryList, list):
            continue
        for country in countryList:
            countryName = country.get("name")
            if not countryName:
                continue
            if countryName not in countryNameSet:
                yield countryName
                countryNameSet.add(countryName)

async def main():
    os.environ["OPENAI_API_KEY"] = "<OPENAI_API_KEY>"

    chatOpenAI = ChatOpenAI(model = "gpt-3.5-turbo-0125")

    jsonOutputParser = JsonOutputParser()

    runnableSequence = chatOpenAI | jsonOutputParser | getCountryNameGenerator

    requestString = """output a list of the countries france, spain and japan and their populations in JSON format. \
Use a dict with an outer key of "countries" which contains a list of countries. \
Each country should have the key `name` and `population`"""

    async for responseChunkString in runnableSequence.astream(requestString):
        print(responseChunkString, end = "|", flush = True)

asyncio.run(main())

"""
France|Spain|Japan|
"""

import asyncio

import os

from langchain_openai import ChatOpenAI

from langchain_core.output_parsers import JsonOutputParser

async def getCountryNameGenerator(inputAsyncGenerator):

"""입력 스트림에서 작동하는 함수이다."""

countryNameSet = set()

async for inputDictionary in inputAsyncGenerator:

if not isinstance(inputDictionary, dict):

continue

if "countries" not in inputDictionary:

continue

countryList = inputDictionary["countries"]

if not isinstance(countryList, list):

continue

for country in countryList:

countryName = country.get("name")

if not countryName:

continue

if countryName not in countryNameSet:

yield countryName

countryNameSet.add(countryName)

async def main():

os.environ["OPENAI_API_KEY"] = "<OPENAI_API_KEY>"

chatOpenAI = ChatOpenAI(model = "gpt-3.5-turbo-0125")

jsonOutputParser = JsonOutputParser()

runnableSequence = chatOpenAI | jsonOutputParser | getCountryNameGenerator

requestString = """output a list of the countries france, spain and japan and their populations in JSON format. \

Use a dict with an outer key of "countries" which contains a list of countries. \

Each country should have the key `name` and `population`"""

async for responseChunkString in runnableSequence.astream(requestString):

print(responseChunkString, end = "|", flush = True)

asyncio.run(main())

"""

France|Spain|Japan|

"""

▶ requirements.txt


annotated-types==0.7.0
anyio==4.4.0
certifi==2024.6.2
charset-normalizer==3.3.2
distro==1.9.0
exceptiongroup==1.2.1
h11==0.14.0
httpcore==1.0.5
httpx==0.27.0
idna==3.7
jsonpatch==1.33
jsonpointer==3.0.0
langchain-core==0.2.7
langchain-openai==0.1.8
langsmith==0.1.77
openai==1.34.0
orjson==3.10.5
packaging==24.1
pydantic==2.7.4
pydantic_core==2.18.4
PyYAML==6.0.1
regex==2024.5.15
requests==2.32.3
sniffio==1.3.1
tenacity==8.4.1
tiktoken==0.7.0
tqdm==4.66.4
typing_extensions==4.12.2
urllib3==2.2.2

annotated-types==0.7.0

anyio==4.4.0

certifi==2024.6.2

charset-normalizer==3.3.2

distro==1.9.0

exceptiongroup==1.2.1

h11==0.14.0

httpcore==1.0.5

httpx==0.27.0

idna==3.7

jsonpatch==1.33

jsonpointer==3.0.0

langchain-core==0.2.7

langchain-openai==0.1.8

langsmith==0.1.77

openai==1.34.0

orjson==3.10.5

packaging==24.1

pydantic==2.7.4

pydantic_core==2.18.4

PyYAML==6.0.1

regex==2024.5.15

requests==2.32.3

sniffio==1.3.1

tenacity==8.4.1

tiktoken==0.7.0

tqdm==4.66.4

typing_extensions==4.12.2

urllib3==2.2.2

※ pip install langchain-openai

[PYTHON/LANGCHAIN] 커스텀 출력 파서 만들기

■ 커스텀 출력 파서를 만드는 방법을 보여준다. ▶ main.py


import os

from langchain_openai              import ChatOpenAI
from langchain_core.output_parsers import JsonOutputParser

os.environ["OPENAI_API_KEY"] = "<OPENAI_API_KEY>"

chatOpenAI = ChatOpenAI(model = "gpt-3.5-turbo-0125")

jsonOutputParser = JsonOutputParser()

def getCountryNameList(inputDictionary):
    """입력 스트림에서 작동하지 않고 스트리밍을 중단하는 함수이다."""

    if not isinstance(inputDictionary, dict):
        return ""

    if "countries" not in inputDictionary:
        return ""

    countryList = inputDictionary["countries"]

    if not isinstance(countryList, list):
        return ""

    countryNameList = [country.get("name") for country in countryList if isinstance(country, dict)]

    return countryNameList

runnableSequence = chatOpenAI | jsonOutputParser | getCountryNameList

requestString = """output a list of the countries france, spain and japan and their populations in JSON format. \
Use a dict with an outer key of "countries" which contains a list of countries. \
Each country should have the key `name` and `population`"""

for responseChunkString in runnableSequence.stream(requestString):
    print(responseChunkString, flush = True)

"""
['France', 'Spain', 'Japan']
"""

import os

from langchain_openai import ChatOpenAI

from langchain_core.output_parsers import JsonOutputParser

os.environ["OPENAI_API_KEY"] = "<OPENAI_API_KEY>"

chatOpenAI = ChatOpenAI(model = "gpt-3.5-turbo-0125")

jsonOutputParser = JsonOutputParser()

def getCountryNameList(inputDictionary):

"""입력 스트림에서 작동하지 않고 스트리밍을 중단하는 함수이다."""

if not isinstance(inputDictionary, dict):

return ""

if "countries" not in inputDictionary:

return ""

countryList = inputDictionary["countries"]

if not isinstance(countryList, list):

return ""

countryNameList = [country.get("name") for country in countryList if isinstance(country, dict)]

return countryNameList

runnableSequence = chatOpenAI | jsonOutputParser | getCountryNameList

requestString = """output a list of the countries france, spain and japan and their populations in JSON format. \

Use a dict with an outer key of "countries" which contains a list of countries. \

Each country should have the key `name` and `population`"""

for responseChunkString in runnableSequence.stream(requestString):

print(responseChunkString, flush = True)

"""

['France', 'Spain', 'Japan']

"""

▶ requirements.txt


annotated-types==0.7.0
anyio==4.4.0
certifi==2024.6.2
charset-normalizer==3.3.2
distro==1.9.0
exceptiongroup==1.2.1
h11==0.14.0
httpcore==1.0.5
httpx==0.27.0
idna==3.7
jsonpatch==1.33
jsonpointer==3.0.0
langchain-core==0.2.7
langchain-openai==0.1.8
langsmith==0.1.77
openai==1.34.0
orjson==3.10.5
packaging==24.1
pydantic==2.7.4
pydantic_core==2.18.4
PyYAML==6.0.1
regex==2024.5.15
requests==2.32.3
sniffio==1.3.1
tenacity==8.4.1
tiktoken==0.7.0
tqdm==4.66.4
typing_extensions==4.12.2
urllib3==2.2.2

annotated-types==0.7.0

anyio==4.4.0

certifi==2024.6.2

charset-normalizer==3.3.2

distro==1.9.0

exceptiongroup==1.2.1

h11==0.14.0

httpcore==1.0.5

httpx==0.27.0

idna==3.7

jsonpatch==1.33

jsonpointer==3.0.0

langchain-core==0.2.7

langchain-openai==0.1.8

langsmith==0.1.77

openai==1.34.0

orjson==3.10.5

packaging==24.1

pydantic==2.7.4

pydantic_core==2.18.4

PyYAML==6.0.1

regex==2024.5.15

requests==2.32.3

sniffio==1.3.1

tenacity==8.4.1

tiktoken==0.7.0

tqdm==4.66.4

typing_extensions==4.12.2

urllib3==2.2.2

※ pip install langchain-openai 명령을 실행했다.

[PYTHON/LANGCHAIN] JsonOutputParser 클래스 : 체인에서 질의 응답시 구조화된 데이터 스트리밍하기

■ JsonOutputParser 클래스를 사용해 체인에서 질의 응답시 구조화된 데이터를 스트리밍하는 방법을 보여준다. ▶ main.py


import os

from langchain_openai              import ChatOpenAI
from langchain_core.output_parsers import JsonOutputParser

os.environ["OPENAI_API_KEY"] = "<OPENAI_API_KEY>"

chatOpenAI = ChatOpenAI(model = "gpt-3.5-turbo-0125")

jsonOutputParser = JsonOutputParser()

runnableSequence = chatOpenAI | jsonOutputParser

requestString = """output a list of the countries france, spain and japan and their populations in JSON format. \
Use a dict with an outer key of "countries" which contains a list of countries. \
Each country should have the key `name` and `population`"""

for responseChunkString in runnableSequence.stream(requestString):
    print(responseChunkString, flush = True)

"""
{}
{'countries': []}
{'countries': [{}]}
{'countries': [{'name': ''}]}
{'countries': [{'name': 'France'}]}
{'countries': [{'name': 'France', 'population': 670}]}
{'countries': [{'name': 'France', 'population': 670760}]}
{'countries': [{'name': 'France', 'population': 67076000}]}
{'countries': [{'name': 'France', 'population': 67076000}, {}]}
{'countries': [{'name': 'France', 'population': 67076000}, {'name': ''}]}
{'countries': [{'name': 'France', 'population': 67076000}, {'name': 'Spain'}]}
{'countries': [{'name': 'France', 'population': 67076000}, {'name': 'Spain', 'population': 473}]}
{'countries': [{'name': 'France', 'population': 67076000}, {'name': 'Spain', 'population': 473299}]}
{'countries': [{'name': 'France', 'population': 67076000}, {'name': 'Spain', 'population': 47329981}]}
{'countries': [{'name': 'France', 'population': 67076000}, {'name': 'Spain', 'population': 47329981}, {}]}
{'countries': [{'name': 'France', 'population': 67076000}, {'name': 'Spain', 'population': 47329981}, {'name': ''}]}
{'countries': [{'name': 'France', 'population': 67076000}, {'name': 'Spain', 'population': 47329981}, {'name': 'Japan'}]}
{'countries': [{'name': 'France', 'population': 67076000}, {'name': 'Spain', 'population': 47329981}, {'name': 'Japan', 'population': 125}]}
{'countries': [{'name': 'France', 'population': 67076000}, {'name': 'Spain', 'population': 47329981}, {'name': 'Japan', 'population': 125960}]}
{'countries': [{'name': 'France', 'population': 67076000}, {'name': 'Spain', 'population': 47329981}, {'name': 'Japan', 'population': 125960000}]}
"""

import os

from langchain_openai import ChatOpenAI

from langchain_core.output_parsers import JsonOutputParser

os.environ["OPENAI_API_KEY"] = "<OPENAI_API_KEY>"

chatOpenAI = ChatOpenAI(model = "gpt-3.5-turbo-0125")

jsonOutputParser = JsonOutputParser()

runnableSequence = chatOpenAI | jsonOutputParser

requestString = """output a list of the countries france, spain and japan and their populations in JSON format. \

Use a dict with an outer key of "countries" which contains a list of countries. \

Each country should have the key `name` and `population`"""

for responseChunkString in runnableSequence.stream(requestString):

print(responseChunkString, flush = True)

"""

{}

{'countries': []}

{'countries': [{}]}

{'countries': [{'name': ''}]}

{'countries': [{'name': 'France'}]}

{'countries': [{'name': 'France', 'population': 670}]}

{'countries': [{'name': 'France', 'population': 670760}]}

{'countries': [{'name': 'France', 'population': 67076000}]}

{'countries': [{'name': 'France', 'population': 67076000}, {}]}

{'countries': [{'name': 'France', 'population': 67076000}, {'name': ''}]}

{'countries': [{'name': 'France', 'population': 67076000}, {'name': 'Spain'}]}

{'countries': [{'name': 'France', 'population': 67076000}, {'name': 'Spain', 'population': 473}]}

{'countries': [{'name': 'France', 'population': 67076000}, {'name': 'Spain', 'population': 473299}]}

{'countries': [{'name': 'France', 'population': 67076000}, {'name': 'Spain', 'population': 47329981}]}

{'countries': [{'name': 'France', 'population': 67076000}, {'name': 'Spain', 'population': 47329981}, {}]}

{'countries': [{'name': 'France', 'population': 67076000}, {'name': 'Spain', 'population': 47329981}, {'name': ''}]}

{'countries': [{'name': 'France', 'population': 67076000}, {'name': 'Spain', 'population': 47329981}, {'name': 'Japan'}]}

{'countries': [{'name': 'France', 'population': 67076000}, {'name': 'Spain', 'population': 47329981}, {'name': 'Japan', 'population': 125}]}

{'countries': [{'name': 'France', 'population': 67076000}, {'name': 'Spain', 'population': 47329981}, {'name': 'Japan', 'population': 125960}]}

{'countries': [{'name': 'France', 'population': 67076000}, {'name': 'Spain', 'population': 47329981}, {'name': 'Japan', 'population': 125960000}]}

"""

▶ requirements.txt


annotated-types==0.7.0
anyio==4.4.0
certifi==2024.6.2
charset-normalizer==3.3.2
distro==1.9.0
exceptiongroup==1.2.1
h11==0.14.0
httpcore==1.0.5
httpx==0.27.0
idna==3.7
jsonpatch==1.33
jsonpointer==3.0.0
langchain-core==0.2.7
langchain-openai==0.1.8
langsmith==0.1.77
openai==1.34.0
orjson==3.10.5
packaging==24.1
pydantic==2.7.4
pydantic_core==2.18.4
PyYAML==6.0.1
regex==2024.5.15
requests==2.32.3
sniffio==1.3.1
tenacity==8.4.1
tiktoken==0.7.0
tqdm==4.66.4
typing_extensions==4.12.2
urllib3==2.2.2

annotated-types==0.7.0

anyio==4.4.0

certifi==2024.6.2

charset-normalizer==3.3.2

distro==1.9.0

exceptiongroup==1.2.1

h11==0.14.0

httpcore==1.0.5

httpx==0.27.0

idna==3.7

jsonpatch==1.33

jsonpointer==3.0.0

langchain-core==0.2.7

langchain-openai==0.1.8

langsmith==0.1.77

openai==1.34.0

orjson==3.10.5

packaging==24.1

pydantic==2.7.4

pydantic_core==2.18.4

PyYAML==6.0.1

regex==2024.5.15

requests==2.32.3

sniffio==1.3.1

tenacity==8.4.1

tiktoken==0.7.0

tqdm==4.66.4

typing_extensions==4.12.2

urllib3==2.2.2

※ pip install langchain-openai

[PYTHON/LANGCHAIN] PydanticToolsParser 클래스 : 생성자에서 tools 인자를 사용해 Pydantic 클래스로 변환하기

■ PydanticToolsParser 클래스의 생성자에서 tools 인자를 사용해 Pydantic 클래스로 변환하는 방법을 보여준다. ▶ main.py


import os

from langchain_core.pydantic_v1                 import BaseModel, Field
from langchain_openai                           import ChatOpenAI
from langchain_core.output_parsers.openai_tools import PydanticToolsParser

os.environ["OPENAI_API_KEY"] = "<OPENAI_API_KEY>"

# 여기의 독스트링은 클래스 이름과 함께 모델에 전달되므로 매우 중요하다.
class Add(BaseModel):
    """Add two integers together."""

    a : int = Field(..., description = "First integer" )
    b : int = Field(..., description = "Second integer")

    def execute(self):
        return self.a + self.b

class Multiply(BaseModel):
    """Multiply two integers together."""

    a : int = Field(..., description = "First integer" )
    b : int = Field(..., description = "Second integer")

    def execute(self):
        return self.a * self.b

toolList = [Add, Multiply]

chatOpenAI = ChatOpenAI(model="gpt-3.5-turbo-0125")

toolRunnableSequence = chatOpenAI.bind_tools(toolList)

executeRunnableSequence = toolRunnableSequence | PydanticToolsParser(tools = toolList)

toolList = executeRunnableSequence.invoke("What is 3 * 12? Also, what is 11 + 49?")

for tool in toolList:
    print(tool.execute())

"""
36
60
"""

import os

from langchain_core.pydantic_v1 import BaseModel, Field

from langchain_openai import ChatOpenAI

from langchain_core.output_parsers.openai_tools import PydanticToolsParser

os.environ["OPENAI_API_KEY"] = "<OPENAI_API_KEY>"

# 여기의 독스트링은 클래스 이름과 함께 모델에 전달되므로 매우 중요하다.

class Add(BaseModel):

"""Add two integers together."""

a : int = Field(..., description = "First integer" )

b : int = Field(..., description = "Second integer")

def execute(self):

return self.a + self.b

class Multiply(BaseModel):

"""Multiply two integers together."""

a : int = Field(..., description = "First integer" )

b : int = Field(..., description = "Second integer")

def execute(self):

return self.a * self.b

toolList = [Add, Multiply]

chatOpenAI = ChatOpenAI(model="gpt-3.5-turbo-0125")

toolRunnableSequence = chatOpenAI.bind_tools(toolList)

executeRunnableSequence = toolRunnableSequence | PydanticToolsParser(tools = toolList)

toolList = executeRunnableSequence.invoke("What is 3 * 12? Also, what is 11 + 49?")

for tool in toolList:

print(tool.execute())

"""

▶ requirements.txt


aiohttp==3.9.5
aiosignal==1.3.1
annotated-types==0.7.0
anyio==4.4.0
async-timeout==4.0.3
attrs==23.2.0
certifi==2024.6.2
charset-normalizer==3.3.2
distro==1.9.0
exceptiongroup==1.2.1
frozenlist==1.4.1
greenlet==3.0.3
h11==0.14.0
httpcore==1.0.5
httpx==0.27.0
idna==3.7
jsonpatch==1.33
jsonpointer==3.0.0
langchain==0.2.5
langchain-core==0.2.9
langchain-openai==0.1.9
langchain-text-splitters==0.2.1
langsmith==0.1.81
multidict==6.0.5
numpy==1.26.4
openai==1.35.3
orjson==3.10.5
packaging==24.1
pydantic==2.7.4
pydantic_core==2.18.4
PyYAML==6.0.1
regex==2024.5.15
requests==2.32.3
sniffio==1.3.1
SQLAlchemy==2.0.31
tenacity==8.4.2
tiktoken==0.7.0
tqdm==4.66.4
typing_extensions==4.12.2
urllib3==2.2.2
yarl==1.9.4

aiohttp==3.9.5

aiosignal==1.3.1

annotated-types==0.7.0

anyio==4.4.0

async-timeout==4.0.3

attrs==23.2.0

certifi==2024.6.2

charset-normalizer==3.3.2

distro==1.9.0

exceptiongroup==1.2.1

frozenlist==1.4.1

greenlet==3.0.3

h11==0.14.0

httpcore==1.0.5

httpx==0.27.0

idna==3.7

jsonpatch==1.33

jsonpointer==3.0.0

langchain==0.2.5

langchain-core==0.2.9

langchain-openai==0.1.9

langchain-text-splitters==0.2.1

langsmith==0.1.81

multidict==6.0.5

numpy==1.26.4

openai==1.35.3

orjson==3.10.5

packaging==24.1

pydantic==2.7.4

pydantic_core==2.18.4

PyYAML==6.0.1

regex==2024.5.15

requests==2.32.3

sniffio==1.3.1

SQLAlchemy==2.0.31

tenacity==8.4.2

tiktoken==0.7.0

tqdm==4.66.4

typing_extensions==4.12.2

urllib3==2.2.2

yarl==1.9.4

※ pip install langchain

[PYTHON/LANGCHAIN] 모델 출력 구문 분석 일반 함수를 사용해 LCEL(LangChain Expression Language)로 커스텀 파서 만들기

■ 모델 출력을 구문 분석하는 일반 함수를 사용해 LCEL(LangChain Expression Language)로 커스텀 파서를 만드는 방법을 보여준다. ▶ main.py


import json
import os
import re

from typing import List

from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_core.messages    import AIMessage
from langchain_core.prompts     import ChatPromptTemplate
from langchain_openai           import ChatOpenAI

os.environ["OPENAI_API_KEY"] = "<OPENAI_API_KEY>"

class Person(BaseModel):
    """Information about a person."""

    name              : str   = Field(..., description = "The name of the person"                       )
    height_in_meters  : float = Field(..., description = "The height of the person expressed in meters.")

class People(BaseModel):
    """Identifying information about all people in a text."""

    people : List[Person]

chatPromptTemplate = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "Answer the user query. Output your answer as JSON that  "
            "matches the given schema: ```json\n{schema}\n```. "
            "Make sure to wrap the answer in ```json and ``` tags"
        ),
        ("human", "{query}")
    ]
).partial(schema = People.schema())

def extractJSON(message : AIMessage) -> List[dict]:
    """Extracts JSON content from a string where JSON is embedded between ```json and ``` tags.

    Parameters:
        text (str): The text containing the JSON content.

    Returns:
        list: A list of extracted JSON strings.
    """
    text = message.content

    # JSON 블록과 일치하도록 정규식 패턴 정의한다.
    pattern = r"```json(.*?)```"

    # 문자열에서 패턴과 겹치지 않는 일치 항목을 모두 찾는다.
    matcheList = re.findall(pattern, text, re.DOTALL)

    print(type(matcheList))

    # 선행 또는 후행 공백을 제거하여 일치하는 JSON 문자열 목록을 반환한다.
    try:
        return [json.loads(match.strip()) for match in matcheList]
    except Exception:
        raise ValueError(f"Failed to parse : {message}")

chatOpenAI = ChatOpenAI(model = "gpt-3.5-turbo-0125")

runnableSequence = chatPromptTemplate | chatOpenAI | extractJSON

resultList = runnableSequence.invoke({"query" : "Anna is 23 years old and she is 6 feet tall"})

print(resultList)

"""
[{'title': 'People', 'description': 'Identifying information about all people in a text.', 'type': 'object', 'properties': {'people': {'title': 'People', 'type': 'array', 'items': {'$ref': '#/definitions/Person'}}}, 'required': ['people'], 'definitions': {'Person': {'title': 'Person', 'description': 'Information about a person.', 'type': 'object', 'properties': {'name': {'title': 'Name', 'description': 'The name of the person', 'type': 'string'}, 'height_in_meters': {'title': 'Height In Meters', 'description': 'The height of the person expressed in meters.', 'type': 'number'}}, 'required': ['name', 'height_in_meters']}}}]
"""

import json

import os

import re

from typing import List

from langchain_core.pydantic_v1 import BaseModel, Field

from langchain_core.messages import AIMessage

from langchain_core.prompts import ChatPromptTemplate

from langchain_openai import ChatOpenAI

os.environ["OPENAI_API_KEY"] = "<OPENAI_API_KEY>"

class Person(BaseModel):

"""Information about a person."""

name : str = Field(..., description = "The name of the person" )

height_in_meters : float = Field(..., description = "The height of the person expressed in meters.")

class People(BaseModel):

"""Identifying information about all people in a text."""

people : List[Person]

chatPromptTemplate = ChatPromptTemplate.from_messages(

[

(

"system",

"Answer the user query. Output your answer as JSON that "

"matches the given schema: ```json\n{schema}\n```. "

"Make sure to wrap the answer in ```json and ``` tags"

("human", "{query}")

]

).partial(schema = People.schema())

def extractJSON(message : AIMessage) -> List[dict]:

"""Extracts JSON content from a string where JSON is embedded between ```json and ``` tags.

Parameters:

text (str): The text containing the JSON content.

Returns:

list: A list of extracted JSON strings.

"""

text = message.content

# JSON 블록과 일치하도록 정규식 패턴 정의한다.

pattern = r"```json(.*?)```"

# 문자열에서 패턴과 겹치지 않는 일치 항목을 모두 찾는다.

matcheList = re.findall(pattern, text, re.DOTALL)

print(type(matcheList))

# 선행 또는 후행 공백을 제거하여 일치하는 JSON 문자열 목록을 반환한다.

try:

return [json.loads(match.strip()) for match in matcheList]

except Exception:

raise ValueError(f"Failed to parse : {message}")

chatOpenAI = ChatOpenAI(model = "gpt-3.5-turbo-0125")

runnableSequence = chatPromptTemplate | chatOpenAI | extractJSON

resultList = runnableSequence.invoke({"query" : "Anna is 23 years old and she is 6 feet tall"})

print(resultList)

"""

[{'title': 'People', 'description': 'Identifying information about all people in a text.', 'type': 'object', 'properties': {'people': {'title': 'People', 'type': 'array', 'items': {'$ref': '#/definitions/Person'}}}, 'required': ['people'], 'definitions': {'Person': {'title': 'Person', 'description': 'Information about a person.', 'type': 'object', 'properties': {'name': {'title': 'Name', 'description': 'The name of the person', 'type': 'string'}, 'height_in_meters': {'title': 'Height In Meters', 'description': 'The height of the person expressed in meters.', 'type': 'number'}}, 'required': ['name', 'height_in_meters']}}}]

"""

▶ requirements.txt


annotated-types==0.7.0
anyio==4.4.0
certifi==2024.6.2
charset-normalizer==3.3.2
distro==1.9.0
exceptiongroup==1.2.1
h11==0.14.0
httpcore==1.0.5
httpx==0.27.0
idna==3.7
jsonpatch==1.33
jsonpointer==3.0.0
langchain-core==0.2.7
langchain-openai==0.1.8
langsmith==0.1.77
openai==1.34.0
orjson==3.10.5
packaging==24.1
pydantic==2.7.4
pydantic_core==2.18.4
PyYAML==6.0.1
regex==2024.5.15
requests==2.32.3
sniffio==1.3.1
tenacity==8.3.0
tiktoken==0.7.0
tqdm==4.66.4
typing_extensions==4.12.2
urllib3==2.2.1

annotated-types==0.7.0

anyio==4.4.0

certifi==2024.6.2

charset-normalizer==3.3.2

distro==1.9.0

exceptiongroup==1.2.1

h11==0.14.0

httpcore==1.0.5

httpx==0.27.0

idna==3.7

jsonpatch==1.33

jsonpointer==3.0.0

langchain-core==0.2.7

langchain-openai==0.1.8

langsmith==0.1.77

openai==1.34.0

orjson==3.10.5

packaging==24.1

pydantic==2.7.4

pydantic_core==2.18.4

PyYAML==6.0.1

regex==2024.5.15

requests==2.32.3

sniffio==1.3.1

tenacity==8.3.0

tiktoken==0.7.0

tqdm==4.66.4

typing_extensions==4.12.2

urllib3==2.2.1

[PYTHON/LANGCHAIN] PydanticOutputParser 클래스 : 구조화된 데이터 받기

■ PydanticOutputParser 클래스를 사용해 구조화된 데이터를 받는 방법을 보여준다. ▶ main.py


import os

from typing import List

from langchain_core.pydantic_v1    import BaseModel, Field
from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.prompts        import ChatPromptTemplate
from langchain_openai              import ChatOpenAI

os.environ["OPENAI_API_KEY"] = "<OPENAI_API_KEY>"

class Person(BaseModel):
    """Information about a person."""

    name             : str   = Field(..., description = "The name of the person"                       )
    height_in_meters : float = Field(..., description = "The height of the person expressed in meters.")

class People(BaseModel):
    """Identifying information about all people in a text."""

    people : List[Person]

pydanticOutputParser = PydanticOutputParser(pydantic_object = People)

instructionFormatString = pydanticOutputParser.get_format_instructions()

chatPromptTemplate = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "Answer the user query. Wrap the output in `json` tags\n{format_instructions}",
        ),
        ("human", "{query}"),
    ]
).partial(format_instructions = instructionFormatString)

chatOpenAI = ChatOpenAI(model = "gpt-3.5-turbo-0125")

query = "Anna is 23 years old and she is 6 feet tall"

runnableSequence = chatPromptTemplate | chatOpenAI | pydanticOutputParser

people = runnableSequence.invoke({"query" : query})

print(people)

"""
people=[Person(name='Anna', height_in_meters=1.83)]
"""

import os

from typing import List

from langchain_core.pydantic_v1 import BaseModel, Field

from langchain_core.output_parsers import PydanticOutputParser

from langchain_core.prompts import ChatPromptTemplate

from langchain_openai import ChatOpenAI

os.environ["OPENAI_API_KEY"] = "<OPENAI_API_KEY>"

class Person(BaseModel):

"""Information about a person."""

name : str = Field(..., description = "The name of the person" )

height_in_meters : float = Field(..., description = "The height of the person expressed in meters.")

class People(BaseModel):

"""Identifying information about all people in a text."""

people : List[Person]

pydanticOutputParser = PydanticOutputParser(pydantic_object = People)

instructionFormatString = pydanticOutputParser.get_format_instructions()

chatPromptTemplate = ChatPromptTemplate.from_messages(

[

(

"system",

"Answer the user query. Wrap the output in `json` tags\n{format_instructions}",

("human", "{query}"),

]

).partial(format_instructions = instructionFormatString)

chatOpenAI = ChatOpenAI(model = "gpt-3.5-turbo-0125")

query = "Anna is 23 years old and she is 6 feet tall"

runnableSequence = chatPromptTemplate | chatOpenAI | pydanticOutputParser

people = runnableSequence.invoke({"query" : query})

print(people)

"""

people=[Person(name='Anna', height_in_meters=1.83)]

"""

▶ requirements.txt


annotated-types==0.7.0
anyio==4.4.0
certifi==2024.6.2
charset-normalizer==3.3.2
distro==1.9.0
exceptiongroup==1.2.1
h11==0.14.0
httpcore==1.0.5
httpx==0.27.0
idna==3.7
jsonpatch==1.33
jsonpointer==3.0.0
langchain-core==0.2.7
langchain-openai==0.1.8
langsmith==0.1.77
openai==1.34.0
orjson==3.10.5
packaging==24.1
pydantic==2.7.4
pydantic_core==2.18.4
PyYAML==6.0.1
regex==2024.5.15
requests==2.32.3
sniffio==1.3.1
tenacity==8.3.0
tiktoken==0.7.0
tqdm==4.66.4
typing_extensions==4.12.2
urllib3==2.2.1

annotated-types==0.7.0

anyio==4.4.0

certifi==2024.6.2

charset-normalizer==3.3.2

distro==1.9.0

exceptiongroup==1.2.1

h11==0.14.0

httpcore==1.0.5

httpx==0.27.0

idna==3.7

jsonpatch==1.33

jsonpointer==3.0.0

langchain-core==0.2.7

langchain-openai==0.1.8

langsmith==0.1.77

openai==1.34.0

orjson==3.10.5

packaging==24.1

pydantic==2.7.4

pydantic_core==2.18.4

PyYAML==6.0.1

regex==2024.5.15

requests==2.32.3

sniffio==1.3.1

tenacity==8.3.0

tiktoken==0.7.0

tqdm==4.66.4

typing_extensions==4.12.2

urllib3==2.2.1

※ pip install langchain-openai 명령을 실행했다.

[PYTHON/LANGCHAIN] PydanticOutputParser 클래스 : get_format_instructions 메소드를 사용해 명령 포맷 문자열 구하기 2

■ PydanticOutputParser 클래스의 get_format_instructions 메소드를 사용해 명령 포맷 문자열을 구하는 방법을 보여준다. ▶ main.py


from typing import List

from langchain_core.pydantic_v1    import BaseModel, Field
from langchain_core.output_parsers import PydanticOutputParser

class Person(BaseModel):
    """Information about a person."""

    name             : str   = Field(..., description = "The name of the person"                       )
    height_in_meters : float = Field(..., description = "The height of the person expressed in meters.")

class People(BaseModel):
    """Identifying information about all people in a text."""

    people : List[Person]

pydanticOutputParser = PydanticOutputParser(pydantic_object = People)

instructionFormatString = pydanticOutputParser.get_format_instructions()

"""
The output should be formatted as a JSON instance that conforms to the JSON schema below.

As an example, for the schema {"properties": {"foo": {"title": "Foo", "description": "a list of strings", "type": "array", "items": {"type": "string"}}}, "required": ["foo"]}
the object {"foo": ["bar", "baz"]} is a well-formatted instance of the schema. The object {"properties": {"foo": ["bar", "baz"]}} is not well-formatted.

Here is the output schema:
```
{"description": "Identifying information about all people in a text.", "properties": {"people": {"title": "People", "type": "array", "items": {"$ref": "#/definitions/Person"}}}, "required": ["people"], "definitions": {"Person": {"title": "Person", "description": "Information about a person.", "type": "object", "properties": {"name": {"title": "Name", "description": "The name of the person", "type": "string"}, "height_in_meters": {"title": "Height In Meters", "description": "The height of the person expressed in meters.", "type": "number"}}, "required": ["name", "height_in_meters"]}}}
```
"""

from typing import List

from langchain_core.pydantic_v1 import BaseModel, Field

from langchain_core.output_parsers import PydanticOutputParser

class Person(BaseModel):

"""Information about a person."""

name : str = Field(..., description = "The name of the person" )

height_in_meters : float = Field(..., description = "The height of the person expressed in meters.")

class People(BaseModel):

"""Identifying information about all people in a text."""

people : List[Person]

pydanticOutputParser = PydanticOutputParser(pydantic_object = People)

instructionFormatString = pydanticOutputParser.get_format_instructions()

"""

The output should be formatted as a JSON instance that conforms to the JSON schema below.

As an example, for the schema {"properties": {"foo": {"title": "Foo", "description": "a list of strings", "type": "array", "items": {"type": "string"}}}, "required": ["foo"]}

the object {"foo": ["bar", "baz"]} is a well-formatted instance of the schema. The object {"properties": {"foo": ["bar", "baz"]}} is not well-formatted.

Here is the output schema:

```

{"description": "Identifying information about all people in a text.", "properties": {"people": {"title": "People", "type": "array", "items": {"$ref": "#/definitions/Person"}}}, "required": ["people"], "definitions": {"Person": {"title": "Person", "description": "Information about a person.", "type": "object", "properties": {"name": {"title": "Name", "description": "The name of the person", "type": "string"}, "height_in_meters": {"title": "Height In Meters", "description": "The height of the person expressed in meters.", "type": "number"}}, "required": ["name", "height_in_meters"]}}}

```

"""

▶ requirements.txt


aiohttp==3.9.5
aiosignal==1.3.1
annotated-types==0.7.0
async-timeout==4.0.3
attrs==23.2.0
certifi==2024.6.2
charset-normalizer==3.3.2
frozenlist==1.4.1
greenlet==3.0.3
idna==3.7
jsonpatch==1.33
jsonpointer==3.0.0
langchain==0.2.5
langchain-core==0.2.7
langchain-text-splitters==0.2.1
langsmith==0.1.77
multidict==6.0.5
numpy==1.26.4
orjson==3.10.5
packaging==24.1
pydantic==2.7.4
pydantic_core==2.18.4
PyYAML==6.0.1
requests==2.32.3
SQLAlchemy==2.0.30
tenacity==8.3.0
typing_extensions==4.12.2
urllib3==2.2.1
yarl==1.9.4

aiohttp==3.9.5

aiosignal==1.3.1

annotated-types==0.7.0

async-timeout==4.0.3

attrs==23.2.0

certifi==2024.6.2

charset-normalizer==3.3.2

frozenlist==1.4.1

greenlet==3.0.3

idna==3.7

jsonpatch==1.33

jsonpointer==3.0.0

langchain==0.2.5

langchain-core==0.2.7

langchain-text-splitters==0.2.1

langsmith==0.1.77

multidict==6.0.5

numpy==1.26.4

orjson==3.10.5

packaging==24.1

pydantic==2.7.4

pydantic_core==2.18.4

PyYAML==6.0.1

requests==2.32.3

SQLAlchemy==2.0.30

tenacity==8.3.0

typing_extensions==4.12.2

urllib3==2.2.1

yarl==1.9.4

※ pip insall langchain

[PYTHON/LANGCHAIN] PydanticOutputParser 클래스 : parse 메소드를 사용해 명령 문자열 파싱 객체 구하기

■ PydanticOutputParser 클래스의 parse 메소드를 사용해 명령 문자열을 파싱하고 객체를 구하는 방법을 보여준다. ▶ main.py


import os

from typing import List

from langchain_experimental.synthetic_data import DatasetGenerator
from langchain_openai                      import ChatOpenAI
from langchain_core.output_parsers         import PydanticOutputParser
from pydantic                              import BaseModel, Field
from langchain_core.prompts                import PromptTemplate
from langchain_openai                      import OpenAI

os.environ["OPENAI_API_KEY"] = "<OPENAI_API_KEY>"

sourceList = [
    {
        "Actor" : "Tom Hanks",
        "Film"  : ["Forrest Gump", "Saving Private Ryan", "The Green Mile", "Toy Story", "Catch Me If You Can"]
    },
    {
        "Actor" : "Tom Hardy",
        "Film"  : ["Inception", "The Dark Knight Rises", "Mad Max: Fury Road", "The Revenant", "Dunkirk"]
    }
]

chatOpenAI = ChatOpenAI(model = "gpt-3.5-turbo", temperature = 0.7)

datasetGenerator = DatasetGenerator(chatOpenAI, {"style" : "informal", "minimal length" : 500})

targetList = datasetGenerator(sourceList)

class Actor(BaseModel):
    Actor : str       = Field(description = "name of an actor"                      )
    Film  : List[str] = Field(description = "list of names of films they starred in")

pydanticOutputParser = PydanticOutputParser(pydantic_object = Actor)

instructionFormatString = pydanticOutputParser.get_format_instructions()

promptTemplate = PromptTemplate(
    template          = "Extract fields from a given text.\n{format_instructions}\n{text}\n",
    input_variables   = ["text"],
    partial_variables = {"format_instructions" : instructionFormatString},
)

stringPromptValue = promptTemplate.format_prompt(text = targetList[0]["text"])

instructionString = stringPromptValue.to_string()

openAI = OpenAI()

resultString = openAI(instructionString)

actor = pydanticOutputParser.parse(resultString)

print(actor)

"""
Actor='Tom Hanks' Film=['Forrest Gump', 'Saving Private Ryan', 'The Green Mile', 'Toy Story', 'Catch Me If You Can']
"""

import os

from typing import List

from langchain_experimental.synthetic_data import DatasetGenerator

from langchain_openai import ChatOpenAI

from langchain_core.output_parsers import PydanticOutputParser

from pydantic import BaseModel, Field

from langchain_core.prompts import PromptTemplate

from langchain_openai import OpenAI

os.environ["OPENAI_API_KEY"] = "<OPENAI_API_KEY>"

sourceList = [

{

"Actor" : "Tom Hanks",

"Film" : ["Forrest Gump", "Saving Private Ryan", "The Green Mile", "Toy Story", "Catch Me If You Can"]

{

"Actor" : "Tom Hardy",

"Film" : ["Inception", "The Dark Knight Rises", "Mad Max: Fury Road", "The Revenant", "Dunkirk"]

}

]

chatOpenAI = ChatOpenAI(model = "gpt-3.5-turbo", temperature = 0.7)

datasetGenerator = DatasetGenerator(chatOpenAI, {"style" : "informal", "minimal length" : 500})

targetList = datasetGenerator(sourceList)

class Actor(BaseModel):

Actor : str = Field(description = "name of an actor" )

Film : List[str] = Field(description = "list of names of films they starred in")

pydanticOutputParser = PydanticOutputParser(pydantic_object = Actor)

instructionFormatString = pydanticOutputParser.get_format_instructions()

promptTemplate = PromptTemplate(

template = "Extract fields from a given text.\n{format_instructions}\n{text}\n",

input_variables = ["text"],

partial_variables = {"format_instructions" : instructionFormatString},

)

stringPromptValue = promptTemplate.format_prompt(text = targetList[0]["text"])

instructionString = stringPromptValue.to_string()

openAI = OpenAI()

resultString = openAI(instructionString)

actor = pydanticOutputParser.parse(resultString)

print(actor)

"""

Actor='Tom Hanks' Film=['Forrest Gump', 'Saving Private Ryan', 'The Green Mile', 'Toy Story', 'Catch Me If You Can']

"""

▶ requirements.txt


aiohttp==3.9.5
aiosignal==1.3.1
annotated-types==0.7.0
anyio==4.4.0
async-timeout==4.0.3
attrs==23.2.0
certifi==2024.6.2
charset-normalizer==3.3.2
dataclasses-json==0.6.7
distro==1.9.0
exceptiongroup==1.2.1
frozenlist==1.4.1
greenlet==3.0.3
h11==0.14.0
httpcore==1.0.5
httpx==0.27.0
idna==3.7
jsonpatch==1.33
jsonpointer==3.0.0
langchain==0.2.5
langchain-community==0.2.5
langchain-core==0.2.7
langchain-experimental==0.0.61
langchain-openai==0.1.8
langchain-text-splitters==0.2.1
langsmith==0.1.77
marshmallow==3.21.3
multidict==6.0.5
mypy-extensions==1.0.0
numpy==1.26.4
openai==1.34.0
orjson==3.10.5
packaging==24.1
pydantic==2.7.4
pydantic_core==2.18.4
PyYAML==6.0.1
regex==2024.5.15
requests==2.32.3
sniffio==1.3.1
SQLAlchemy==2.0.30
tenacity==8.3.0
tiktoken==0.7.0
tqdm==4.66.4
typing-inspect==0.9.0
typing_extensions==4.12.2
urllib3==2.2.1
yarl==1.9.4

aiohttp==3.9.5

aiosignal==1.3.1

annotated-types==0.7.0

anyio==4.4.0

async-timeout==4.0.3

attrs==23.2.0

certifi==2024.6.2

charset-normalizer==3.3.2

dataclasses-json==0.6.7

distro==1.9.0

exceptiongroup==1.2.1

frozenlist==1.4.1

greenlet==3.0.3

h11==0.14.0

httpcore==1.0.5

httpx==0.27.0

idna==3.7

jsonpatch==1.33

jsonpointer==3.0.0

langchain==0.2.5

langchain-community==0.2.5

langchain-core==0.2.7

langchain-experimental==0.0.61

langchain-openai==0.1.8

langchain-text-splitters==0.2.1

langsmith==0.1.77

marshmallow==3.21.3

multidict==6.0.5

mypy-extensions==1.0.0

numpy==1.26.4

openai==1.34.0

orjson==3.10.5

packaging==24.1

pydantic==2.7.4

pydantic_core==2.18.4

PyYAML==6.0.1

regex==2024.5.15

requests==2.32.3

sniffio==1.3.1

SQLAlchemy==2.0.30

tenacity==8.3.0

tiktoken==0.7.0

tqdm==4.66.4

typing-inspect==0.9.0

typing_extensions==4.12.2

urllib3==2.2.1

yarl==1.9.4

※ pip install

[PYTHON/LANGCHAIN] PydanticOutputParser 클래스 : get_format_instructions 메소드를 사용해 명령 포맷 문자열 구하기 1

■ PydanticOutputParser 클래스의 get_format_instructions 메소드를 사용해 명령 포맷 문자열을 구하는 방법을 보여준다. ▶ main.py


from typing import List

from langchain_core.output_parsers import PydanticOutputParser
from pydantic                      import BaseModel, Field

class Actor(BaseModel):
    Actor : str       = Field(description = "name of an actor"                      )
    Film  : List[str] = Field(description = "list of names of films they starred in")

pydanticOutputParser = PydanticOutputParser(pydantic_object = Actor)

instructionFormatString = pydanticOutputParser.get_format_instructions()

print(instructionFormatString)

"""
The output should be formatted as a JSON instance that conforms to the JSON schema below.

As an example, for the schema {"properties": {"foo": {"title": "Foo", "description": "a list of strings", "type": "array", "items": {"type": "string"}}}, "required": ["foo"]}
the object {"foo": ["bar", "baz"]} is a well-formatted instance of the schema. The object {"properties": {"foo": ["bar", "baz"]}} is not well-formatted.

Here is the output schema:
```
{"properties": {"Actor": {"description": "name of an actor", "title": "Actor", "type": "string"}, "Film": {"description": "list of names of films they starred in", "items": {"type": "string"}, "title": "Film", "type": "array"}}, "requ
"""

from typing import List

from langchain_core.output_parsers import PydanticOutputParser

from pydantic import BaseModel, Field

class Actor(BaseModel):

Actor : str = Field(description = "name of an actor" )

Film : List[str] = Field(description = "list of names of films they starred in")

pydanticOutputParser = PydanticOutputParser(pydantic_object = Actor)

instructionFormatString = pydanticOutputParser.get_format_instructions()

print(instructionFormatString)

"""

The output should be formatted as a JSON instance that conforms to the JSON schema below.

As an example, for the schema {"properties": {"foo": {"title": "Foo", "description": "a list of strings", "type": "array", "items": {"type": "string"}}}, "required": ["foo"]}

the object {"foo": ["bar", "baz"]} is a well-formatted instance of the schema. The object {"properties": {"foo": ["bar", "baz"]}} is not well-formatted.

Here is the output schema:

```

{"properties": {"Actor": {"description": "name of an actor", "title": "Actor", "type": "string"}, "Film": {"description": "list of names of films they starred in", "items": {"type": "string"}, "title": "Film", "type": "array"}}, "requ

"""

▶ requirements.txt


aiohttp==3.9.5
aiosignal==1.3.1
annotated-types==0.7.0
async-timeout==4.0.3
attrs==23.2.0
certifi==2024.6.2
charset-normalizer==3.3.2
frozenlist==1.4.1
greenlet==3.0.3
idna==3.7
jsonpatch==1.33
jsonpointer==3.0.0
langchain==0.2.5
langchain-core==0.2.7
langchain-text-splitters==0.2.1
langsmith==0.1.77
multidict==6.0.5
numpy==1.26.4
orjson==3.10.5
packaging==24.1
pydantic==2.7.4
pydantic_core==2.18.4
PyYAML==6.0.1
requests==2.32.3
SQLAlchemy==2.0.30
tenacity==8.3.0
typing_extensions==4.12.2
urllib3==2.2.1
yarl==1.9.4

aiohttp==3.9.5

aiosignal==1.3.1

annotated-types==0.7.0

async-timeout==4.0.3

attrs==23.2.0

certifi==2024.6.2

charset-normalizer==3.3.2

frozenlist==1.4.1

greenlet==3.0.3

idna==3.7

jsonpatch==1.33

jsonpointer==3.0.0

langchain==0.2.5

langchain-core==0.2.7

langchain-text-splitters==0.2.1

langsmith==0.1.77

multidict==6.0.5

numpy==1.26.4

orjson==3.10.5

packaging==24.1

pydantic==2.7.4

pydantic_core==2.18.4

PyYAML==6.0.1

requests==2.32.3

SQLAlchemy==2.0.30

tenacity==8.3.0

typing_extensions==4.12.2

urllib3==2.2.1

yarl==1.9.4

※ pip install langchain

[PYTHON/LANGCHAIN] StrOutputParser 클래스 : invoke 메소드를 사용해 출력 결과 파싱하기

■ StrOutputParser 클래스의 invoke 메소드를 사용해 출력 결과를 파싱하는 방법을 보여준다. ※ StrOutputParser 클래스는 AIMessageChunk에서 content 필드를 추출하여 모델이 반환한 토큰을 제공하는