LangChain에서 Airbyte 소스를 활용하는 방법 소개

에디터 노트: 이 게시물은 Airbyte 팀과 협력하여 작성되었습니다. 그들은 문서 로더(document loaders)로 더 많은 데이터 소스를 LangChain에 연결하는 것을 매우 쉽게 만들었습니다.

이제 Airbyte 소스를 활용하여 Gong, Hubspot, Salesforce, Shopify, Stripe, Typeform 및 Zendesk Support와 같은 다양한 서비스를 직접 LangChain 기반 애플리케이션 내에서 문서 로더 형태로 사용할 수 있게 되었습니다.

예를 들어, 사용자의 Stripe 인보이스를 불러오려면 AirbyteStripeLoader를 사용할 수 있습니다. 설치는 매우 간단합니다. LangChain이 로컬에 설치되어 있다면, 관심 있는 소스만 설치하면 바로 사용할 수 있습니다:

pip install airbyte-source-stripe

그 후에는 로더를 가져와(import) 설정과 불러오고 싶은 스트림을 전달하기만 하면 됩니다:

from langchain.document_loaders.airbyte import AirbyteStripeLoader

config = {

이번 릴리스를 통해 LangChain에서 Python 기반 소스를 사용자의 Python 런타임 내에서 실행하는 것이 그 어느 때보다 쉬워졌습니다. 더 이상 Airbyte 인스턴스를 구동하거나 Airbyte Cloud에 API 호출을 할 필요가 없습니다.

호스팅형과 임베디드 Airbyte 간 이동

내부적으로 동일한 코드가 실행되기 때문에, 모든 Airbyte 빌드 로더는 Airbyte 서비스의 해당 소스와 호환됩니다. 이는 임베디드 로딩 파이프라인을 자체 호스팅(self-hosted)된 Airbyte 설치 또는 Airbyte Cloud 인스턴스로 옮기는 것이 매우 쉽다는 것을 의미합니다. 설정 객체와 레코드의 형태가 100% 호환됩니다.

호스팅형 Airbyte에서 동기화(syncs)를 실행하는 것은 다음을 의미합니다:

• 실행 중인 파이프라인 추적을 위한 UI
• 실패한 동기화에 대한 알림 기능
• 스케줄에 따라 파이프라인을 쉽게 실행할 수 있음

LangChain 로더로 동기화를 실행하는 것은 다음을 의미합니다:

• 또 다른 서비스를 구동할 필요가 없어 오버헤드가 없음
• 타이밍과 파이프라인 실행에 대한 완전한 제어권

Airbyte 레코드를 LangChain 문서로 매핑하기

기본적으로, 각 레코드는 로더의 일부로서 Document로 매핑되며, 레코드 내의 모든 다양한 필드가 해당 레코드의 메타데이터가 됩니다. 문서의 텍스트 부분은 빈 문자열로 남겨집니다. 데이터에 따라 레코드의 텍스트 부분을 구성하기 위해 record_handler를 전달하여 이 동작을 사용자 정의할 수 있습니다:

def handle_record(record, id):

return Document(page_content=record.data["title"], metadata=record.data)

loader = AirbyteGongLoader(config=config, record_handler=handle_record, stream_name="calls")

증분 로드 (Incremental loads)

사용자의 Python 애플리케이션이 기본적으로 Airbyte 플랫폼 역할을 하므로, "동기화"가 어떻게 실행될지에 대한 완전한 제어권을 가집니다. 예를 들어, 스트림이 이를 지원하는 경우 loader의 `

with open('stripe_sync_checkpoint.json', 'wb') as file:

file.write(loader.last_state.json())

// ... later

with open('stripe_sync_checkpoint.json', 'r') as file:

current_state = AirbyteStateMessage.parse_raw(file.read())

incremental_loader = AirbyteStripeLoader(config=config, stream_name="invoices", state=current_state)

new_docs = incremental_loader.load()

사용자 지정 소스 (Custom sources)

현재까지 다음 Airbyte 소스는 pip 패키지로 제공됩니다 (추가 예정):

• Gong: pip install airbyte-source-gong
• Hubspot: pip install airbyte-source-hubspot
• Salesforce: pip install airbyte-source-salesforce
• Shopify: pip install airbyte-source-shopify
• Stripe: pip install airbyte-source-stripe
• Typeform: pip install airbyte-source-typeform
• Zendesk Support: pip install airbyte-source-zendesk-support

하지만, 사용자가 자체적인 사용자 지정 Airbyte 소스를 구현했다면, Airbyte CDK의 Source 인터페이스와 연동되는 AirbyteCDKLoader 기본 클래스를 사용하여 통합하는 것도 가능합니다:

from langchain.document_loaders.airbyte import AirbyteCDKLoader

from my_source.source import MyCustomSource # 여기에 자체 소스 연결

config = {

# 사용자 지정 설정

}

loader = AirbyteCDKLoader(source_class=MyCustomSource, config=config, stream_name="my-stream")

또한, 메인 Airbyte 저장소에서 소스를 설치할 수도 있습니다. 예를 들어, Github 소스를 가져오려면 다음을 실행하기만 하면 됩니다:

pip install "source_github@git+https://github.com/airbytehq/airbyte.git@master#subdirectory=airbyte-integrations/connectors/source-github"

그 후, 해당 소스는 AirbyteCDKLoader에 사용 가능합니다:

from source_github.source import SourceGithub

issues_loader = AirbyteCDKLoader(source_class=SourceGithub, config=config, stream_name="issues")

자체 소스를 작성하는 방법에 대한 커넥터 개발 문서를 확인해 보세요. 시작하기가 매우 쉽고, 필요에 따라 로컬 임베디드 로더(local embedded loaders)부터 호스팅된 Airbyte 인스턴스 사용까지 원활하게 전환할 수 있도록 해줍니다.

질문이 있으신가요? 여러분의 의견을 듣고 싶습니다

Airbyte를 활용하여 데이터를 LLM 기반 애플리케이션으로 전송하는 데 관심이 있다면, 가장 중요한 기능을 우선순위로 지정할 수 있도록 설문조사에 참여해 주시기 바랍니다.

질문이 있거나 이 방식으로 로더로 노출되는 다른 기존 소스에 관심이 있다면, 커뮤니티 slack 채널이나 LangChain discord 서버의 Airbyte 채널을 통해 주저하지 말고 문의해 주세요.