아래 예시 코드는 n8n에서 보여지는 플로우(트리거 → 유튜브 자막 추출 → 요약 → 정보 추출 → AI Agent → 데이터 머지 → 디스코드 발송/노션 저장 → …)를 파이썬에서 유사한 기능으로 구성했을 때의 예시입니다. 실제 서비스에 사용하기 위해서는 각종 API 키, 보안, 에러 처리, 예외 상황 대응 등이 추가로 필요하며, 경우에 따라 더 세분화된 로직이 요구될 수 있습니다. 또한 설명을 돕기 위한 간소화된 예시이므로 필요에 맞게 수정하여 사용하세요.
1. Trigger (예시: 스크립트 시작 또는 Flask 엔드포인트)
n8n의 “When clicking ‘Test workflow’”처럼, 파이썬 스크립트가 시작되는 시점을 트리거로 봅니다. 간단히 “main.py”로 시작하는 경우 예시는 아래와 같습니다.
# main.py
def main():
print("Workflow started!")
# 1) YouTube Transcript 가져오기
transcript_text = get_youtube_transcript("동영상_ID")
# 2) 요약
summary_text = summarize_text(transcript_text)
# 3) 정보 추출
extracted_info = extract_information(transcript_text)
# 4) AI Agent 호출
ai_result = run_ai_agent(summary_text, extracted_info)
# 5) 두 개 결과 머지
merged_data = merge_data(summary_text, ai_result)
# 6) 디스코드로 전송
send_to_discord(WEBHOOK_URL, merged_data)
# 7) 노션에 저장
append_block_to_notion(NOTION_PAGE_ID, merged_data)
print("Workflow finished!")
if __name__ == "__main__":
main()
위 main() 함수는 n8n에서 클릭해서 트리거되는 부분을 간단히 대체합니다.
2. YouTube Transcript (자막 추출)
n8n의 “YouTube Transcript” 노드에 해당하는 기능을 파이썬에서 구현할 수 있습니다.
youtube_transcript_api 라이브러리를 사용하면 됩니다.
# youtube_transcript.py
from youtube_transcript_api import YouTubeTranscriptApi
def get_youtube_transcript(video_id: str, lang_code: str = 'ko') -> str:
"""
주어진 YouTube 동영상 ID로부터 자막을 가져와서 이어붙인 텍스트를 반환합니다.
"""
try:
transcript_list = YouTubeTranscriptApi.get_transcript(video_id, languages=[lang_code, 'en'])
full_text = " ".join([item['text'] for item in transcript_list])
return full_text
except Exception as e:
print(f"Error fetching transcript: {e}")
return ""
3. Summarization Chain (요약)
n8n의 “Summarization Chain” 노드를 파이썬에서 OpenAI API 등을 사용해 구현하는 예시입니다.
# summarization.py
import os
import openai
openai.api_key = os.getenv("OPENAI_API_KEY") # 환경변수 사용 권장
def summarize_text(text: str) -> str:
"""
OpenAI의 GPT 모델을 활용해 긴 텍스트를 요약해주는 함수 예시
"""
if not text:
return "No text provided."
prompt = f"다음 텍스트를 간단히 요약해 주세요:\n{text}\n요약:"
try:
response = openai.Completion.create(
model="text-davinci-003",
prompt=prompt,
max_tokens=200,
temperature=0.7
)
summary = response.choices[0].text.strip()
return summary
except Exception as e:
print(f"Error during summarization: {e}")
return ""
4. Information Extractor (정보 추출)
n8n에서 “Information Extractor” 노드는 텍스트에서 특정 정보를 뽑아내는 용도로 보입니다. 예를 들어 장소, 시간, 인물, 키워드 등을 추출하는 식으로 구현할 수 있습니다.
# info_extractor.py
import openai
def extract_information(text: str) -> dict:
"""
GPT를 활용해서 텍스트에서 특정 정보를 추출하는 예시
(예: 날짜, 장소, 이름 등)
"""
if not text:
return {}
prompt = f"""
다음 텍스트에서 중요한 날짜, 장소, 인물, 핵심 키워드를 JSON 형태로 뽑아주세요:
{text}
출력 예시: {{
"dates": [...],
"locations": [...],
"people": [...],
"keywords": [...]
}}
"""
try:
response = openai.Completion.create(
model="text-davinci-003",
prompt=prompt,
max_tokens=300,
temperature=0
)
# GPT가 JSON 형태로 뱉을 것을 가정
extracted_str = response.choices[0].text.strip()
# 문자열 -> 파이썬 dict로 파싱
import json
info_dict = json.loads(extracted_str)
return info_dict
except Exception as e:
print(f"Error extracting information: {e}")
return {}
5. AI Agent (도구 사용 Agent)
n8n에서 “AI Agent” 노드는 ChatGPT나 다른 모델을 사용하면서, 외부 API를 검색하거나 정보를 받아오는 구조로 보입니다. 예시는 LangChain의 Agent 등을 직접 파이썬으로 대체한 간단 버전입니다.
# ai_agent.py
import requests
def run_ai_agent(summary: str, extracted_info: dict) -> str:
"""
AI 모델을 활용해 추가 질의/대답/도구 사용을 가정한 간단 예시
"""
# 일단 summary와 extracted_info를 가공해 한 문장 결과를 만든다고 가정
result = f"AI Agent Result:\nSummary: {summary}\nExtracted Info: {extracted_info}"
return result
실제로는 LangChain이나 ChatGPT의 Tools/Functions 기능 등을 사용하여
HTTP 요청
계산
검색
등을 수행하도록 만들 수 있습니다.
6. Merge (데이터 병합)
n8n에서 “Merge” 노드는 여러 데이터 스트림을 합치는 역할입니다. 파이썬에서는 단순히 딕셔너리를 합친다든지, 문자열을 합치는 식으로 구현할 수 있습니다.
# merge.py
def merge_data(summary_text: str, ai_result: str) -> str:
"""
두 가지 결과를 간단히 합쳐 문자열 형태로 반환
"""
return f"--- MERGED DATA ---\nSummary:\n{summary_text}\n\nAI Result:\n{ai_result}"
7. Discord (메시지 전송)
n8n의 “Discord” 노드는 디스코드로 메시지를 보내는 기능입니다. 파이썬에서는 Webhook을 사용하거나 discord.py 라이브러리를 사용해 구현할 수 있습니다. 여기서는 웹훅(Webhook) 예시를 듭니다.
# discord_sender.py
import requests
def send_to_discord(webhook_url: str, content: str):
"""
디스코드 Webhook URL로 간단히 메시지를 보내는 예시
"""
data = {
"content": content
}
try:
response = requests.post(webhook_url, json=data)
if response.status_code != 204:
print(f"Error sending message to Discord: {response.status_code}, {response.text}")
else:
print("Message sent to Discord successfully!")
except Exception as e:
print(f"Exception when sending to Discord: {e}")
8. Notion (블록 추가)
n8n의 “Notion” 노드는 노션에 블록을 추가하거나 페이지를 생성합니다. 파이썬에서는 Notion 공식 API를 사용합니다.
# notion_appender.py
import requests
import json
NOTION_SECRET = "YOUR_NOTION_SECRET"
NOTION_API_BASE_URL = "https://api.notion.com/v1"
def append_block_to_notion(page_id: str, content: str):
"""
Notion 페이지(블록)에 텍스트 블록을 append하는 예시
"""
url = f"{NOTION_API_BASE_URL}/blocks/{page_id}/children"
headers = {
"Authorization": f"Bearer {NOTION_SECRET}",
"Notion-Version": "2022-06-28",
"Content-Type": "application/json"
}
new_block = {
"children": [
{
"object": "block",
"type": "paragraph",
"paragraph": {
"rich_text": [
{
"type": "text",
"text": {
"content": content
}
}
]
}
}
]
}
try:
response = requests.patch(url, headers=headers, data=json.dumps(new_block))
if response.status_code != 200:
print(f"Error appending block to Notion: {response.status_code}, {response.text}")
else:
print("Block appended to Notion successfully!")
except Exception as e:
print(f"Exception when appending to Notion: {e}")
9. OpenAI Chat Model (예: ChatCompletion)
n8n에 “OpenAI Chat Model” 노드가 있다면, 파이썬에서는 다음처럼 ChatCompletion을 이용할 수 있습니다.
# openai_chat_model.py
import openai
def chat_with_openai(messages):
"""
GPT-3.5/4 ChatCompletion 예시
messages: [{"role": "system", "content": "..."},
{"role": "user", "content": "..."}]
"""
try:
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=messages,
temperature=0.7
)
return response.choices[0].message["content"]
except Exception as e:
print(f"Error in ChatCompletion: {e}")
return ""
10. HTTP Request (예: Google Search)
n8n의 “HTTP Request” 노드는 외부 API를 호출하는 기능입니다. 파이썬에서는 requests 라이브러리를 사용합니다. 예시로 Google Custom Search API를 호출하는 코드를 보여드립니다.
# http_request.py
import requests
import os
GOOGLE_API_KEY = os.getenv("GOOGLE_API_KEY")
SEARCH_ENGINE_ID = os.getenv("SEARCH_ENGINE_ID")
def google_search(query: str) -> dict:
"""
구글 커스텀 검색 API를 통해 검색 결과를 가져오는 간단한 예시
"""
url = "https://www.googleapis.com/customsearch/v1"
params = {
"key": GOOGLE_API_KEY,
"cx": SEARCH_ENGINE_ID,
"q": query
}
try:
response = requests.get(url, params=params)
data = response.json()
return data
except Exception as e:
print(f"Error making Google search: {e}")
return {}
11. Structured Output Parser (구조화된 출력 파싱)
n8n의 “Structured Output Parser” 노드는 AI 모델이 JSON 형식 등으로 출력한 결과물을 파싱하는 기능이 될 수 있습니다. 파이썬에서는 단순히 json.loads를 쓰거나, 필요한 로직을 추가하면 됩니다.
# structured_output_parser.py
import json
def parse_structured_output(text: str) -> dict:
"""
텍스트(특히 JSON 형태)를 받아 파이썬 딕셔너리로 변환
"""
try:
return json.loads(text)
except json.JSONDecodeError:
print("Invalid JSON format.")
return {}
정리
Trigger: 스크립트 시작(또는 웹훅 엔드포인트 등)
YouTube Transcript: youtube_transcript_api 로 자막 불러오기
Summarization Chain: OpenAI API로 요약
Information Extractor: OpenAI API로 특정 정보 추출
AI Agent: 대화형 AI 에이전트(외부 API, 검색 등 연동 가능)
Merge: 여러 결과물(요약, 추출 정보 등)을 합침
Discord: 디스코드로 메시지 전송 (웹훅 사용)
Notion: 노션 API로 페이지(블록)에 내용 추가
OpenAI Chat Model: ChatCompletion 등 활용
HTTP Request: 외부 API(예: 구글) 호출
Structured Output Parser: AI가 생성한 JSON 텍스트 파싱
위의 각 단계 함수들을 main.py 같은 하나의 스크립트에서 순서대로 호출하면, n8n에서 시각적으로 구성한 워크플로우와 유사한 데이터 흐름을 파이썬으로 재현할 수 있습니다. 실제 서비스를 만들 때에는 API 키/토큰, 오류 처리, 비동기 작업, 대량 데이터 처리, 로그 관리 등을 추가로 고려해 주세요.
