체인을 실행하는 방법
LangChain을 사용하여 커스텀 체인을 생성하는 과정은 다음과 같다.
1. 필요한 컴포넌트를 정의하고, 각각 "Runnable" 인터페이스를 구현합니다.
2. 컴포넌트들을 조합하여 사용자 정의 체인을 생성합니다.
3. 생성된 체인을 사용하여 데이터 처리 작업을 수행합니다. 이때, invoke, batch, stream 메소드를 사용하여 원하는 방식으로 데이터를 처리할 수 있습니다.
.invoke() - 주어진 입력에 대해 체인을 호출하고, 결과를 반환합니다.
prompt = ChatPromptTemplate.from_template('인생에서 {topic}이란 뭔지 알려줘') model = ChatOpenAI(model = 'gpt-4o') output_parser = StrOutputParser() chain = prompt | model | output_parser result = chain.invoke({'topic' : '인간'}) result.batch() - 입력 리스트에 대해 체인을 호출하고, 각 입력에 대한 결과를 리스트로 반환합니다.
topic = ['삶', '죽음', '생명'] model = ChatOpenAI(model = 'gpt-4o') output_parser = StrOutputParser() result = chain.batch([{'topic' : t} for t in topic]) for topic, result in zip(topic, result): print(f'{topic} 설명: , {result[:50]}..........')
stream - 입력에 대해 체인을 호출하고, 결과의 조각들을 스트리밍합니다.(실시간)
추가 설명:stream = chain.stream({'topic' : '종교'}) print('stream 결과') for chunk in stream: print(chunk, end='', flush = True) print()
end=''는 print() 함수의 옵션 매개변수로, 출력 후 기본적으로 추가되는 줄바꿈 문자(\n) 대신 다른 문자열을 지정할 수 있게 합니다.
기본 개념
print() 함수는 출력 버퍼를 사용합니다.
출력된 내용은 즉시 화면에 보이는 것이 아니라, 출력 버퍼에 쌓였다가 적절한 시점에 플러시(flush, 즉 출력)를 통해 화면에 나타납니다.
기본적으로, 버퍼는 다음 상황에서 자동으로 플러시됩니다:
줄바꿈 문자 (\n)가 출력될 때.
프로그램이 종료될 때.
다른 시스템 조건에 따라.
하지만 여기서
flush=True를 사용하면 출력 버퍼를 강제로 비워서 (즉시 플러시) 화면에 출력하게 만듭니다.
주로 실시간으로 데이터가 출력되어야 하는 경우에 사용합니다. 예:
긴 작업의 진행 상황을 표시할 때.
실시간 스트리밍 데이터를 출력할 때.
로그 출력에서 지연 없이 결과를 바로 보고 싶을 때.
ainvoke()
import nest_asyncio import asyncio # nest_asyncio 적용 (구글 코랩 등 주피터 노트북에서 실행 필요) nest_asyncio.apply() # 비동기 메소드 사용 (async/await 구문 필요) async def run_async(): result = await chain.ainvoke({"topic": "해류"}) print("ainvoke 결과:", result[:50], "...") asyncio.run(run_async())추가 설명:
먼저 nest_asyncio.
nest_asyncio를 알기 전, 먼저 이벤트 루프란 것이 뭔지 알아야 합니다.(밑에 따로 설명)
밑에서 읽었다는 전제하에 설명을 이어가자면
next_asynico와 그냥 asynico의 차이는 주로 동일한 이벤트 루프를 다시 진입(re-enter) 가능하게 만드느냐에 있습니다. 기본적으로 Python의 asyncio는 이벤트 루프를 한 번만 실행할 수 있게 설계되어 있습니다. 하지만 특정 상황에서는 이벤트 루프를 중첩해서 실행해야 할 때가 있습니다. 이때 nest_asyncio를 사용합니다.
참고로 nest_asyncio 라이브러리는 apply() 메서드를 통해 간편하게 중첩된 이벤트 루프 처리를 구현할 수 있습니다. 그렇기에 비동기에서만 쓰는 .ainvoke()를 쓰는것입니다.
그 전에, 중첩된 이벤트 루프란, 하나의 이벤트 루프가 이미 실행 중인 상태에서 새로운 이벤트 루프를 시작하려는 상황을 말합니다. 이는 이벤트 루프가 중복되어 실행되는 상태를 뜻합니다.
이벤트 루프는 한 번에 하나만 실행될 수 있어요. 만약 이미 이벤트 루프가 실행 중인데 또 다른 루프를 시작하려고 하면, 충돌이 발생할 수 있습니다. 이때 발생하는 것이 바로 중첩된 이벤트 루프입니다.
자, nest_asyncio은 솔직히 말하면 별거 없습니다. 주피터 노트북 쓰시는 분들은 그냥 위에다 nest_asyncio.apply()을 쓰시면 됩니다. 그냥 주피터 노트북에는 오류가 있는데 그걸 해결해주는, 즉 보호막이라 생각하시면 됩니다, 그래서 위에만 있지 코드에서는 따로 안 쓰입니다.
이제 이해를 하고 다시 코드를 보면 쉽게 알 수 있습니다.
1. 그냥 주피터를 쓰면 nest_asyncio.apply()을 쓰고,
2. async def run_async():에서 async def를 써서 비동기 함수를 만든 다음에,
3. result = await chain.ainvoke({"topic": "해류"})여기서 그냥 await을 통해 비동기 함수(또는 코루틴)의 실행을 "실행하고 기다린다"는 역할을 하고, 코루틴을 실행하고 그 결과를 기다리며, CPU가 다른 작업을 할 수 있도록 비동기적으로 대기하는 것입니다. (await 없이 실행하면 코드가 실행되지 않고 "계획"만 세워집니다.)
여기서 잠깐! chain.ainvoke({"topic": "해류"})은 비동기가 아니지 않나? nono 바로 ainvoke
정확히 말하자면 ainvoke는 비동기 함수(또는 비동기 메서드)로, 이를 호출하면 비동기 작업이 실행됩니다. 이를 await로 기다려야 하는 이유는 그 내부에서 비동기 작업이 처리되기 때문입니다. 즉, 비동기 메서드는 코루틴을 반환할 수 있으며, 이를 await으로 기다려야 정상적으로 작업을 완료할 수 있습니다.
4. 그렇게 print("ainvoke 결과:", result[:50], "...")은 결과값을 출력하는 기본 코드이며
5. asyncio.run(run_async())을 통해 값을 출력합니다.
*참고로 asyncio.run(run_async())뿐 아니라 그냥 await run_async()를 해도 실행이 됩니다.
이벤트 루프
이벤트 루프(Event Loop)는 비동기 프로그래밍의 핵심 구성 요소로, 발생하는 이벤트나 작업을 처리하기 위해 끊임없이 실행되는 반복 구조를 말합니다. Python에서는 asyncio 모듈을 통해 이벤트 루프를 관리하고 실행할 수 있습니다.
1. 이벤트 루프의 기본 개념
이벤트 루프는 비동기 작업(async tasks)을 관리하고 실행합니다.
작업(task)이 준비될 때마다 이벤트 루프가 해당 작업을 실행하고, 대기 중인 작업으로 넘어갑니다.
Blocking 작업 없이 여러 작업을 동시에 처리할 수 있게 합니다.
흐름 이해
이벤트 발생 대기: 이벤트 루프는 작업이나 이벤트가 발생하기를 기다립니다.
이벤트 처리: 발생한 이벤트(예: 파일 읽기, 네트워크 응답)를 비동기 함수로 전달해 처리합니다.
작업 전환: 현재 작업이 완료되지 않으면 다른 대기 중인 작업을 처리합니다.
2. 왜 필요할까?
이벤트 루프는 비동기 작업을 효율적으로 관리하기 위해 사용됩니다.
동기 방식에서는 하나의 작업이 완료될 때까지 프로그램이 멈춤(blocking) 상태가 되지만,
비동기 방식에서는 작업이 기다리는 동안 다른 작업을 진행할 수 있습니다.
이해가 잘 안 간다면, 예시를 보겠습니다.
동기 방식 (Blocking):
먼저,import time def task1(): print("Task 1 시작") time.sleep(3) # 3초 동안 멈춤 print("Task 1 완료") def task2(): print("Task 2 시작") time.sleep(2) # 2초 동안 멈춤 print("Task 2 완료") task1() task2()
이렇게 하면 맨 위에 있는 task1이 실행이 되면서, Task 1 시작이라고 뜬 후 3초 후 Task 1 완료라고 뜸과 동시에 Task 2 시작이라고 뜬다. 그리고 2초 후 마지막으로 Task 2완료라고 뜬다. 즉, 하나 하나씩 실행되고 있다는 뜻이다.
Task 1 시작 #3초 대기 Task 1 완료 Task 2 시작 #2초 대기 Task 2 완료
하지만 비동기는 그와 다르게,
import asyncio async def task1(): print("Task 1 시작") await asyncio.sleep(3) # 3초 대기, 다른 작업 실행 가능 print("Task 1 완료") async def task2(): print("Task 2 시작") await asyncio.sleep(2) # 2초 대기, 다른 작업 실행 가능 print("Task 2 완료") async def main(): await asyncio.gather(task1(), task2()) # 동시에 실행 asyncio.run(main())이렇게 실행을 시키면, Task 1과 Task 2가 한꺼번에 실행이 됩니다. 즉, 말 그대로 작업이 기다리는 동안 다른 작업을 진행한다는 것입니다.
이렇게 하면
Task 1 시작 Task 2 시작 #2초 대기 Task 2 완료 #1초 대기(2초가 지났으므로) Task 1 완료이렇게 한꺼번에 실행이 된다는 뜻입니다.
그런데 코드가 이해가 안 갈 수 있습니다.import asyncio1. Python의 내장 라이브러리인 asyncio를 가져옵니다.
2. asyncio는 비동기 프로그래밍을 쉽게 구현할 수 있는 기능(이벤트 루프, 태스크 관리, 비동기 I/O 등)을 제공합니다.async def task1(): print("Task 1 시작") await asyncio.sleep(3) # 3초 대기, 다른 작업 실행 가능 print("Task 1 완료") async def task2(): print("Task 2 시작") await asyncio.sleep(2) # 2초 대기, 다른 작업 실행 가능 print("Task 2 완료")def에서 async를 붙이면, async def 비동기 함수로 정의할 수 있습니다.
참고로 await는 Python 비동기 프로그래밍에서 사용하는 키워드로 비동기 함수(async def) 안에서만 사용할 수 있습니다. 게다가 실행 중인 작업이 완료될 때까지 기다리지만, 그동안 이벤트 루프는 다른 작업을 처리할 수 있도록 실행을 넘깁니다.
즉, await는 실행을 멈추지 않고 다른 작업을 처리할 수 있게 합니다. 즉, await이 호출되면 이벤트 루프는 해당 작업이 끝날 때까지 기다리지 않고 다른 대기 중인 작업으로 전환합니다.
그런데 asyncio.sleep은 time.sleep와 다릅니다. 왜 그럴까요? 그것은 함수가 async 함수이기 때문입니다.
-time.sleep은 동기 방식으로 실행되므로 비동기 함수(async def) 내부에서 사용할 수 없습니다.
-asyncio.sleep은 비동기 방식으로 작동하므로 await와 함께 사용하여 다른 작업을 처리할 수 있습니다.
그래서 await과 asyncio.time의 결합은:
await - 해당 작업이 완료될 때까지 기다리지만, 대기 시간 동안 이벤트 루프는 다른 비동기 작업을 실행할 수 있도록 합니다.
asyncio.sleep(3) - 3초 동안 대기하는 비동기 작업을 생성합니다.
그렇기 때문에 asyncio.sleep은 비동기 함수이기 때문에, 반드시 await를 사용해야 실행됩니다. await 없이는 비동기 작업이 제대로 수행되지 않습니다(그냥 바로 실행이 되어버립니다).
그런데 갑자기 궁금하지 않나요? 그냥 실행만 시키면 되는건데 왜 굳이 await가 붙어야 하는지?
여기서 코루틴(coroutine)이라는 개념이 나옵니다. 여기서 코루틴(coroutine)은 Python의 비동기 함수입니다. 비동기 함수는 async def로 정의되며, 실행하면 코루틴 객체를 반환합니다.
코루틴 객체는 "실행 계획"만 반환할 뿐, 실제로 실행되지는 않으며, 이 코루틴을 실행하려면 await를 사용하거나, 이벤트 루프에 등록해야 합니다.
즉, 위에서 쓰인 asyncio.sleep은 코루틴 객체(coroutine object)를 반환합니다. 하지만 코루틴 객체는 실행 가능한 작업이 아니라, 비동기 실행의 계획만 담고 있는 객체입니다. 그렇기 때문에 await을 써야, 비동기 실행의 계획을 사용한다는 뜻입니다.async def main(): await asyncio.gather(task1(), task2()) # 동시에 실행여기서 asyncio.gather은 여러 비동기 작업을 병렬로 실행할 수 있게 해주는 함수이며, 이 코드에서는 task1()과 task2()를 동시에 실행합니다.
그리고 두 작업이 완료될 때까지 await로 기다립니다.
참고로 asyncio.gather에 더 많은 비동기 함수를 전달하여 병렬 처리를 확장할 수 있습니다
게다가 asynico.gather은 저렇게 안 하고 따로
asyncio.run(asyncio.gather(task1(), task2()))로 해도 작동이 됩니다.
즉, asyncio.gather은 따로 쓰일 수 있습니다.asyncio.run(main())asyncio.run()은 이벤트 루프를 생성하고 코루틴을 실행합니다.
여기서 비동기 작업의 실행을 관리하는 핵심 컴포넌트가 이벤트 루프 (Event Loop)인데, 이 이벤트 루프를 생성하고 실행시키기 위해 존재하는 키워드가 바로 asyncio.run() 또는 asyncio.get_event_loop()입니다.
참고로 굳이 asyncio.run말고 그냥 await main()을 쓰면 asyncio.run과 똑같이 값이 출력이 됩니다.
하지만 asyncio.run()이랑 asyncio.get_event_loop()은 서로 다른 상황에 쓰입니다.
asyncio.get_event_loop()는 현재 이벤트 루프를 반환하는 메서드입니다. 이 메서드는 이벤트 루프를 수동으로 관리하고 싶을 때 사용됩니다.
get_event_loop()를 사용하면 이미 실행 중인 이벤트 루프를 재사용하거나 새로 생성된 루프를 반환받을 수 있습니다.
이 후에 run_until_complete()와 같은 메서드를 사용하여 비동기 함수를 실행하고, 해당 작업이 완료될 때까지 기다립니다.
예:import asyncio async def main(): print("Hello, World!") # 이벤트 루프 객체를 가져옴 loop = asyncio.get_event_loop() # 비동기 작업을 실행하고, 완료될 때까지 기다림 loop.run_until_complete(main()) # "Hello, World!"가 출력됨get_event_loop()는 수동으로 이벤트 루프를 관리해야 할 때 사용됩니다. 예를 들어, 여러 작업을 제어하거나, 이미 실행 중인 루프를 재사용하고자 할 때 유용합니다.
run_until_complete()을 사용하여 이벤트 루프에서 비동기 함수를 실행하도록 지시합니다.
즉, 이벤트 루프를 생성하는 데에는 비슷한 역할을 하지만, 이것을 바로 시작시키느냐, 혹은 변수에 넣어서 다르게 쓰고 싶으냐의 차이입니다.
결론은,
asyncio.run(): 간단한 비동기 작업을 실행할 때 유용합니다. 새로운 이벤트 루프를 생성하고 실행하며, 비동기 작업이 끝나면 자동으로 종료됩니다.
asyncio.get_event_loop(): 이벤트 루프를 수동으로 관리하고 여러 작업을 제어해야 할 때 사용합니다. 이 방법은 더 많은 유연성을 제공하지만, 이벤트 루프의 생명 주기를 직접 관리해야 합니다.
따라서, 간단한 비동기 실행에서는 asyncio.run()을 사용하고, 복잡한 루프 관리나 여러 작업을 동시에 처리하려면 asyncio.get_event_loop()와 run_until_complete()를 사용하는 방식이 적합합니다.
여기서 run_untill_complete()는 asyncio의 이벤트 루프에서 주어진 비동기 작업(코루틴)을 실행하고, 해당 작업이 완료될 때까지 기다리는 메서드입니다. 주로 asyncio.get_event_loop()와 함께 사용되어 비동기 함수나 작업을 실행하고 완료될 때까지 기다리도록 합니다.
동작 원리
이벤트 루프에 비동기 작업 전달: run_until_complete()는 비동기 함수나 작업을 이벤트 루프에 전달하여 실행을 시작합니다.
비동기 작업의 완료를 기다림: 해당 작업이 완료될 때까지 이벤트 루프는 다른 작업을 병렬로 처리하거나 기다리는 동안 다른 I/O 작업을 처리합니다.
작업 완료 후 이벤트 루프 종료: 비동기 작업이 완료되면 run_until_complete()는 종료되고, 이벤트 루프도 종료됩니다.
여기까지...