pipeline 실행 결과 저장, async 적용 #3

api_server.py

@@ -1,22 +1,94 @@
-from fastapi import FastAPI, Request
+import asyncio
+import json
+from contextlib import suppress
+from uuid import uuid4
+
+from fastapi import FastAPI
 from pydantic import BaseModel
+from fastapi.responses import StreamingResponse
+
 from pipeline import pipeline
-from fastapi.responses import JSONResponse
 
 app = FastAPI()
 
-class QueryRequest(BaseModel):
+
+class StreamAnalyzeRequest(BaseModel):
     query: str
 
-@app.post("/analyze")
-async def analyze(request: QueryRequest):
-    result = pipeline(request.query)
-    # Convert AnalysisResult to dict for JSON serialization
-    return JSONResponse(content={
-        "query": result.query,
-        "ticker": result.ticker,
-        "analysis_type": result.analysis_type,
-        "data_context": result.data_context,
-        "llm_response": result.llm_response,
-        "timestamp": getattr(result, "timestamp", None)
-    })
+
+@app.post("/analyze/")
+async def analyze_stream(request: StreamAnalyzeRequest):
+    request_id = uuid4().hex[:8]
+    loop = asyncio.get_running_loop()
+    queue = asyncio.Queue()
+    state = {"result": None, "error": None}
+
+    def emit_event(payload: dict):
+        payload.setdefault("request_id", request_id)
+        loop.call_soon_threadsafe(queue.put_nowait, payload)
+
+    def on_delta(delta: str):
+        emit_event({"type": "delta", "delta": delta})
+
+    def on_status(message: str):
+        emit_event({"type": "status", "message": message})
+
+    async def run_pipeline_task():
+        try:
+            state["result"] = await pipeline(
+                request.query,
+                stream_output=True,
+                stream_callback=on_delta,
+                status_callback=on_status,
+                request_id=request_id,
+            )
+        except Exception as exc:
+            state["error"] = str(exc)
+        finally:
+            await queue.put(None)
+
+    task = asyncio.create_task(run_pipeline_task())
+
+    async def event_generator():
+        try:
+            while True:
+                event = await queue.get()
+                if event is None:
+                    break
+                payload = json.dumps(event, ensure_ascii=False)
+                yield f"data: {payload}\n\n"
+
+            if state["error"]:
+                payload = json.dumps({"type": "error", "message": state["error"], "request_id": request_id}, ensure_ascii=False)
+                yield f"data: {payload}\n\n"
+            else:
+                result_payload = json.dumps(
+                    {
+                        "type": "result",
+                        "result": {
+                            "request_id": request_id,
+                            "query": state["result"].query,
+                            "ticker": state["result"].ticker,
+                            "analysis_type": state["result"].analysis_type,
+                            "data_context": state["result"].data_context,
+                            "llm_response": state["result"].llm_response,
+                            "timestamp": getattr(state["result"], "timestamp", None),
+                        },
+                        "request_id": request_id,
+                    },
+                    ensure_ascii=False,
+                )
+                yield f"data: {result_payload}\n\n"
+                yield f"data: {{\"type\":\"done\",\"request_id\":\"{request_id}\"}}\n\n"
+        finally:
+            if not task.done():
+                task.cancel()
+                with suppress(asyncio.CancelledError):
+                    await task
+
+    headers = {
+        "Cache-Control": "no-cache",
+        "Connection": "keep-alive",
+        "X-Accel-Buffering": "no",
+    }
+    return StreamingResponse(event_generator(), media_type="text/event-stream", headers=headers)

data/collect_data.py

@@ -1,6 +1,4 @@
-from typing import Any
-from datetime import datetime
-import yfinance as yf
+import asyncio
 
 from data.earnings import fetch_earnings_data
 from data.finance_fetchers import fetch_fundamentals, fetch_news, fetch_price_data
@@ -9,7 +7,7 @@ from llm.web_search import fetch_web_search
 from utils.data_types import MarketData
 
 
-def collect_data(client, intent, tools):
+async def collect_data(client, intent, tools):
     # 인텐트와 선택된 도구 목록을 기반으로 모든 시장 데이터 수집
     ticker = intent.get("ticker", "")
     period = intent.get("time_range", "1y")
@@ -25,31 +23,32 @@ def collect_data(client, intent, tools):
 
     if ticker:
         if "price" in tools:
-            price_data = fetch_price_data(ticker, period)
+            price_data = await asyncio.to_thread(fetch_price_data, ticker, period)
             print(f"    → 가격 데이터: {len(price_data)}개 지표")
 
         if "fundamentals" in tools:
-            fundamentals = fetch_fundamentals(ticker)
+            fundamentals = await asyncio.to_thread(fetch_fundamentals, ticker)
             print(f"    → 펀더멘털: {len(fundamentals)}개 지표")
 
         if "technicals" in tools:
-            technicals = fetch_technicals(ticker, "6mo")
+            technicals = await asyncio.to_thread(fetch_technicals, ticker, "6mo")
             print(f"    → 기술지표: {len(technicals)}개 지표")
 
         if "news" in tools:
-            news_snippets = fetch_news(ticker)
+            news_snippets = await asyncio.to_thread(fetch_news, ticker)
             print(f"    → 뉴스: {len(news_snippets)}개 헤드라인")
 
         if "earnings" in tools:
-            earnings_data = fetch_earnings_data(
+            earnings_data = await asyncio.to_thread(
+                fetch_earnings_data,
                 ticker,
                 target_year=intent.get("target_year"),
-                target_quarter=intent.get("target_quarter")
+                target_quarter=intent.get("target_quarter"),
             )
 
     if "web_search" in tools:
         company_name = fundamentals.get("company_name", ticker) if fundamentals else ticker
-        web_search_results = fetch_web_search(
+        web_search_results = await fetch_web_search(
             client=client,
             ticker=ticker,
             company_name=company_name,

intent/intent_parser.py

@@ -31,8 +31,7 @@ def route_tools(intent):
     return selected
 
 
-
-def parse_intent(client, user_query):
+async def parse_intent(client, user_query):
     LLM_MODEL_NAME = os.environ.get('LLM_MODEL_NAME')
     INTENT_TOOL = {
         "type": "function",
@@ -54,7 +53,7 @@ def parse_intent(client, user_query):
             }
         }
     }
-    response = client.chat.completions.create(
+    response = await client.chat.completions.create(
         model=LLM_MODEL_NAME,
         messages=[
             {

llm/generator.py

@@ -1,5 +1,6 @@
 import os
 import json
+import time
 from .prompts import SYSTEM_PROMPTS
 from data.news import search_google_news, format_news_list
 
@@ -7,6 +8,7 @@ from data.news import search_google_news, format_news_list
 # ⑤ LLM 분석 생성 (Responses API + web_search 상시 활성)
 # ─────────────────────────────────────────────
 
+
 def extract_response_text(resp):
     # Responses API 출력에서 텍스트 블록만 추출하여 합산
     texts = []
@@ -21,7 +23,7 @@ def extract_response_text(resp):
     return "\n".join(texts)
 
 
-def generate_search_keywords(client, user_query, intent):
+async def generate_search_keywords(client, user_query, intent):
     """LLM을 통해 구글 뉴스 검색어 리스트 생성"""
     language = intent.get("language", "ko")
     prompt = f"""
@@ -30,37 +32,56 @@ def generate_search_keywords(client, user_query, intent):
 [사용자 질의]
 {user_query}
 """
-    LLM_MODEL_NAME = os.environ.get('LLM_MODEL_NAME')
-    resp = client.chat.completions.create(
+    LLM_MODEL_NAME = os.environ.get("LLM_MODEL_NAME")
+    resp = await client.chat.completions.create(
         model=LLM_MODEL_NAME,
-        messages=[{"role": "system", "content": "뉴스 검색 키워드 생성"},
-                  {"role": "user", "content": prompt}],
+        messages=[
+            {"role": "system", "content": "뉴스 검색 키워드 생성"},
+            {"role": "user", "content": prompt},
+        ],
     )
-    # 키워드 리스트 추출 (예: ['삼성전자', '반도체 전망', ...])
     content = resp.choices[0].message.content
     try:
         keywords = json.loads(content)
         if isinstance(keywords, list):
             return keywords
     except Exception:
-        # 리스트 형태가 아니면 줄바꿈/쉼표로 분리
-        return [k.strip() for k in content.replace('\n', ',').split(',') if k.strip()]
+        return [k.strip() for k in content.replace("\n", ",").split(",") if k.strip()]
     return []
 
 
-def generate_news_info(client, user_query, intent):
-    """LLM 키워드 생성 + 구글 뉴스 검색 + 포맷까지 한 번에 처리"""
+async def generate_news_info(client, user_query, intent):
+    """LLM 키워드 생성 + 구글 뉴스 검색 + 포맷까지 한 번에 처리 (Async LLM)"""
     language = intent.get("language", "ko")
-    keywords = generate_search_keywords(client, user_query, intent)
+    keywords = await generate_search_keywords(client, user_query, intent)
     news_list = search_google_news(keywords, language=language)
     news_str = format_news_list(news_list)
     return news_str
 
 
-def generate_analysis(client, user_query, context, intent, news_str):
-    # Responses API + web_search로 최종 투자 분석 리포트 생성
+def _extract_delta_text(event):
+    # OpenAI SDK 이벤트 객체(dict/typed)에서 delta 텍스트를 안전하게 추출
+    delta = getattr(event, "delta", None)
+    if delta:
+        return delta
+    if isinstance(event, dict):
+        return event.get("delta", "")
+    return ""
+
+
+async def generate_analysis(
+    client,
+    user_query,
+    context,
+    intent,
+    news_str,
+    stream_output=False,
+    stream_callback=None,
+    log_callback=None,
+):
+    # Async Responses API로 최종 투자 분석 리포트 생성
     analysis_type = intent.get("analysis_type", "general")
-    language      = intent.get("language", "ko")
+    language = intent.get("language", "ko")
     system_prompt = SYSTEM_PROMPTS.get(analysis_type, SYSTEM_PROMPTS["general"])
     system_prompt += f"\n\n반드시 {language} 언어로 답변하세요. 투자 조언이 아닌 정보 제공임을 명시하세요."
 
@@ -75,20 +96,122 @@ def generate_analysis(client, user_query, context, intent, news_str):
 [최신 구글 뉴스]
 {news_str} """
 
-    LLM_MODEL_NAME = os.environ.get('LLM_MODEL_NAME')
-    print(f"[⑤] LLM 분석 생성 중 (Responses API, 모델: {LLM_MODEL_NAME})...")
+    LLM_MODEL_NAME = os.environ.get("LLM_MODEL_NAME")
+    delta_log_every = int(os.environ.get("DELTA_LOG_EVERY", "20"))
+    delta_log_preview_chars = int(os.environ.get("DELTA_LOG_PREVIEW_CHARS", "80"))
+    start_msg = f"[⑤] LLM 분석 생성 중 (Responses API, 모델: {LLM_MODEL_NAME})..."
+    if log_callback:
+        log_callback(start_msg)
+    else:
+        print(start_msg)
 
-    resp = client.responses.create(
+    request_kwargs = dict(
         model=LLM_MODEL_NAME,
         tools=[
             {
                 "type": "web_search",
-                "user_location": {"type": "approximate", "country": "KR"}
+                "user_location": {"type": "approximate", "country": "KR"},
             }
         ],
         input=full_input,
     )
 
+    if stream_output:
+        try:
+            all_deltas = []
+            first_event_logged = False
+            delta_count = 0
+            delta_total_chars = 0
+            response_created_at = None
+            first_delta_latency_logged = False
+
+            async with client.responses.stream(**request_kwargs) as stream:
+                if log_callback:
+                    log_callback("[⑤][stream] stream context 진입 완료")
+                else:
+                    print("[⑤][stream] stream context 진입 완료")
+
+                async for event in stream:
+                    if not first_event_logged:
+                        event_type = getattr(event, "type", None)
+                        msg = f"[⑤][stream] 첫 이벤트 수신 type={event_type}"
+                        if log_callback:
+                            log_callback(msg)
+                        else:
+                            print(msg)
+                        first_event_logged = True
+
+                    if getattr(event, "type", None) == "response.created" and response_created_at is None:
+                        response_created_at = time.perf_counter()
+
+                    if getattr(event, "type", None) == "response.output_text.delta":
+                        delta = _extract_delta_text(event)
+                        if delta:
+                            all_deltas.append(delta)
+                            delta_count += 1
+                            delta_total_chars += len(delta)
+
+                            if response_created_at is not None and not first_delta_latency_logged:
+                                latency_s = time.perf_counter() - response_created_at
+                                latency_msg = f"[⑤][stream] response.created -> 첫 delta 지연: {latency_s:.3f}초"
+                                if log_callback:
+                                    log_callback(latency_msg)
+                                else:
+                                    print(latency_msg)
+                                first_delta_latency_logged = True
+
+                            should_log_delta = delta_count <= 3 or (delta_log_every > 0 and delta_count % delta_log_every == 0)
+                            if should_log_delta:
+                                preview = delta.replace("\n", "\\n")[:delta_log_preview_chars]
+                                msg = f"[⑤][stream] delta#{delta_count} len={len(delta)} total_chars={delta_total_chars} preview='{preview}'"
+                                if log_callback:
+                                    log_callback(msg)
+                                else:
+                                    print(msg)
+
+                            if stream_callback:
+                                stream_callback(delta)
+                            else:
+                                print(delta, end="", flush=True)
+
+                if not stream_callback:
+                    print()
+
+                if response_created_at is not None and not first_delta_latency_logged:
+                    no_delta_msg = "[⑤][stream] response.created 이후 delta 미수신"
+                    if log_callback:
+                        log_callback(no_delta_msg)
+                    else:
+                        print(no_delta_msg)
+
+                summary_msg = f"[⑤][stream] delta 수집 요약: chunks={delta_count}, total_chars={delta_total_chars}"
+                if log_callback:
+                    log_callback(summary_msg)
+                else:
+                    print(summary_msg)
+
+                if log_callback:
+                    log_callback("[⑤][stream] 최종 응답 수집 시작")
+                else:
+                    print("[⑤][stream] 최종 응답 수집 시작")
+                final_resp = await stream.get_final_response()
+                if log_callback:
+                    log_callback("[⑤][stream] 최종 응답 수집 완료")
+                else:
+                    print("[⑤][stream] 최종 응답 수집 완료")
+
+                result = extract_response_text(final_resp)
+                if result:
+                    return result
+                joined = "".join(all_deltas).strip()
+                return joined or "(분석 결과를 가져오지 못했습니다)"
+        except Exception as e:
+            err_msg = f"[⑤] 스트리밍 출력 실패, 일반 모드로 재시도: {e}"
+            if log_callback:
+                log_callback(err_msg)
+            else:
+                print(err_msg)
+
+    resp = await client.responses.create(**request_kwargs)
     result = extract_response_text(resp)
     return result or "(분석 결과를 가져오지 못했습니다)"
-

llm/web_search.py

@@ -1,6 +1,7 @@
 from datetime import datetime
 import os
 
+
 def build_web_search_query(analysis_type, base, target_year, target_quarter, today):
     if analysis_type == "news_summary":
         return f"{today} 기준 {base} 최신 뉴스 및 주요 이슈"
@@ -14,6 +15,7 @@ def build_web_search_query(analysis_type, base, target_year, target_quarter, tod
         return f"{base} 최신 사업 현황, 경쟁사 동향, 리스크 요인 {today}"
     return f"{base} 최신 투자 정보 및 시장 동향 {today}"
 
+
 def extract_web_search_blocks(resp):
     results = []
     for item in resp.output:
@@ -22,12 +24,12 @@ def extract_web_search_blocks(resp):
         for block in (getattr(item, "content", []) or []):
             if getattr(block, "type", None) != "output_text":
                 continue
-            text        = getattr(block, "text", "") or ""
+            text = getattr(block, "text", "") or ""
             annotations = getattr(block, "annotations", []) or []
-            citations   = [
+            citations = [
                 {
                     "title": getattr(ann, "title", ""),
-                    "url":   getattr(ann, "url", ""),
+                    "url": getattr(ann, "url", ""),
                 }
                 for ann in annotations
                 if getattr(ann, "type", "") == "url_citation"
@@ -36,7 +38,8 @@ def extract_web_search_blocks(resp):
                 results.append({"text": text.strip(), "citations": citations})
     return results
 
-def fetch_web_search(client, ticker, company_name, analysis_type, language="ko", target_year=None, target_quarter=None):
+
+async def fetch_web_search(client, ticker, company_name, analysis_type, language="ko", target_year=None, target_quarter=None):
     today  = datetime.now().strftime("%Y년 %m월 %d일")
     base   = f"{company_name}({ticker})" if ticker else company_name
     query  = build_web_search_query(
@@ -46,7 +49,7 @@ def fetch_web_search(client, ticker, company_name, analysis_type, language="ko",
     results = []
     LLM_MODEL_NAME = os.environ.get('LLM_MODEL_NAME')
     try:
-        resp = client.responses.create(
+        resp = await client.responses.create(
             model=LLM_MODEL_NAME,
             tools=[
                 {

pipeline.py

@@ -1,32 +1,125 @@
 import os
 import json
+import asyncio
 import textwrap
+from contextlib import suppress
 from datetime import datetime
-from typing import Any
-from dataclasses import dataclass, field
-from openai import OpenAI
+from pathlib import Path
+from functools import partial
+from uuid import uuid4
 
+from openai import AsyncOpenAI
+from dotenv import load_dotenv
 
-from intent.intent_parser import parse_intent
+from intent.intent_parser import parse_intent, route_tools
 from llm.generator import generate_analysis, generate_news_info
-
 from data.collect_data import collect_data
-
 from context.context_builder import build_context
-from intent.intent_parser import route_tools
 from utils.data_types import AnalysisResult
-from dotenv import load_dotenv
 
 load_dotenv()
 
 
 api_key = os.environ.get("LLM_MODEL_API_KEY")
 if api_key:
-    client = OpenAI(api_key=api_key)
+    async_client = AsyncOpenAI(api_key=api_key)
 else:
-    client = OpenAI()
+    async_client = AsyncOpenAI()
+
+PIPELINE_LOG_PATH = Path("results/pipeline.jsonl")
+WAITING_STATUS_MESSAGES = [
+    "답변 구조를 정리하고 있어요...",
+    "핵심 포인트를 우선순위로 정리 중이에요...",
+    "시장 데이터와 맥락을 교차 검증하고 있어요...",
+    "근거를 확인하면서 답변을 다듬고 있어요...",
+    "요약과 리스크 포인트를 함께 정리하고 있어요...",
+    "답변 완성도를 높이고 있어요...",
+]
+WAITING_STATUS_INTERVAL_RANGE_SECONDS = 12
+
+
+def append_pipeline_jsonl_log(result, executed_at, request_id, process_timings=None):
+    PIPELINE_LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
+
+    log_record = {
+        "request_id": request_id,
+        "executed_at": executed_at,
+        "query": result.query,
+        "ticker": result.ticker,
+        "analysis_type": result.analysis_type,
+        "timestamp": result.timestamp,
+        "data_context": result.data_context,
+        "llm_response": result.llm_response,
+        "process_timings": process_timings or {},
+    }
+
+    with PIPELINE_LOG_PATH.open("a", encoding="utf-8") as f:
+        f.write(json.dumps(log_record, ensure_ascii=False) + "\n")
+
+
+def _emit_status(message, status_callback=None):
+    if status_callback:
+        status_callback(message)
+
+
+def _now_str():
+    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+
+
+def _req_prefix(request_id):
+    return f"[req:{request_id}]"
+
+
+def _req_log_and_status_callback(request_id, on_latency_marker=None):
+    def _log(message):
+        print(f"{_req_prefix(request_id)} {message}")
+        if on_latency_marker and "response.created -> 첫 delta 지연:" in message:
+            on_latency_marker()
+
+    return _log
+
+
+async def _emit_waiting_statuses_in_order(status_callback, stop_event):
+    if not status_callback:
+        return
+
+    index = 0
+    message_count = len(WAITING_STATUS_MESSAGES)
+
+    while not stop_event.is_set():
+        await asyncio.sleep(WAITING_STATUS_INTERVAL_RANGE_SECONDS)
+        if stop_event.is_set():
+            break
+        _emit_status(WAITING_STATUS_MESSAGES[index], status_callback)
+        index = (index + 1) % message_count
 
-def pipeline(query):
+
+async def _run_stage(label, coro, stream_output, process_timings, request_id):
+    started_at_dt = datetime.now()
+    started_at = started_at_dt.strftime("%Y-%m-%d %H:%M:%S")
+
+    if stream_output:
+        print(f"{_req_prefix(request_id)} [{label}] 시작: {started_at}")
+
+    value = await coro
+
+    ended_at_dt = datetime.now()
+    ended_at = ended_at_dt.strftime("%Y-%m-%d %H:%M:%S")
+    elapsed_seconds = round((ended_at_dt - started_at_dt).total_seconds(), 3)
+
+    process_timings[label] = {
+        "started_at": started_at,
+        "ended_at": ended_at,
+        "elapsed_seconds": elapsed_seconds,
+    }
+
+    if stream_output:
+        print(f"{_req_prefix(request_id)} [{label}] 종료: {ended_at} (소요 {elapsed_seconds}초)")
+
+    return value
+
+
+async def pipeline(query, stream_output=False, stream_callback=None, status_callback=None, request_id=None):
     """
     파이프라인:
         ① 인텐트 파싱  (Chat Completions + Function Calling)
@@ -35,43 +128,112 @@ def pipeline(query):
         ④ 컨텍스트 조립
         ⑤ 분석 생성   (Responses API + web_search 상시 활성)
     """
-    global client
+    global async_client
+    request_id = request_id or uuid4().hex[:8]
+    executed_at = _now_str()
+    process_timings = {}
+
     print(f"\n{'='*60}")
-    print(f"Wallstreet-AI 분석 시작: {query}")
+    print(f"{_req_prefix(request_id)} Wallstreet-AI 분석 시작: {query}")
     print('='*60)
 
-    # 1. intent 파싱, 2. tool 라우팅, 3. 데이터 수집
-    intent      = parse_intent(client, query)
-    tools       = route_tools(intent)
-    market_data = collect_data(client, intent, tools)
+    _emit_status("질문 의도를 파악하고 있어요...", status_callback)
+    intent = await _run_stage(
+        "① 인텐트 파싱",
+        parse_intent(async_client, query),
+        stream_output,
+        process_timings,
+        request_id,
+    )
+
+    _emit_status("분석에 필요한 도구를 고르고 있어요...", status_callback)
+    tools = await _run_stage(
+        "② Tool 라우팅",
+        asyncio.to_thread(route_tools, intent),
+        stream_output,
+        process_timings,
+        request_id,
+    )
+
+    _emit_status("시장 데이터를 수집하고 있어요...", status_callback)
+    market_data = await _run_stage(
+        "③ 데이터 수집",
+        collect_data(async_client, intent, tools),
+        stream_output,
+        process_timings,
+        request_id,
+    )
+
+    _emit_status("수집 데이터를 정리하고 있어요...", status_callback)
+    news_str = await generate_news_info(async_client, query, intent)
+    context = await _run_stage(
+        "④ 컨텍스트 조립",
+        asyncio.to_thread(partial(build_context, market_data, intent, news_str=news_str)),
+        stream_output,
+        process_timings,
+        request_id,
+    )
+    print(f"{_req_prefix(request_id)} [④] 컨텍스트 빌드 완료 ({len(context)} 문자)")
 
-    # 4. 구글 뉴스 정보 생성 및 컨텍스트에 포함
-    news_str = generate_news_info(client, query, intent)
-    context = build_context(market_data, intent, news_str=news_str)
-    print(f"[④] 컨텍스트 빌드 완료 ({len(context)} 문자)")
+    _emit_status("AI가 답변을 작성하고 있어요...", status_callback)
+    waiting_stop_event = asyncio.Event()
+    waiting_task = None
+    if status_callback and stream_output:
+        waiting_task = asyncio.create_task(
+            _emit_waiting_statuses_in_order(status_callback, waiting_stop_event)
+        )
 
-    # 5. 분석 생성 (news_str는 이미 context에 포함되었으므로 전달하지 않음)
-    response = generate_analysis(client, query, context, intent, news_str=None)
+    try:
+        response = await _run_stage(
+            "⑤ 분석 생성",
+            generate_analysis(
+                async_client,
+                query,
+                context,
+                intent,
+                news_str=news_str,
+                stream_output=stream_output,
+                stream_callback=stream_callback,
+                log_callback=_req_log_and_status_callback(
+                    request_id,
+                    on_latency_marker=waiting_stop_event.set,
+                ),
+            ),
+            stream_output,
+            process_timings,
+            request_id,
+        )
+    finally:
+        waiting_stop_event.set()
+        if waiting_task:
+            waiting_task.cancel()
+            with suppress(asyncio.CancelledError):
+                await waiting_task
 
     result = AnalysisResult(
         query=query,
         ticker=intent.get("ticker", ""),
         analysis_type=intent.get("analysis_type", "general"),
         data_context={
-            "price":             market_data.price_data,
-            "fundamentals":      market_data.fundamentals,
-            "technicals":        market_data.technicals,
-            "news_count":        len(market_data.news_snippets),
-            "earnings":          market_data.earnings_data,
+            "price": market_data.price_data,
+            "fundamentals": market_data.fundamentals,
+            "technicals": market_data.technicals,
+            "news_count": len(market_data.news_snippets),
+            "earnings": market_data.earnings_data,
             "web_search_blocks": len(market_data.web_search_results),
-            "google_news":       news_str,  
+            "google_news": news_str,
         },
-        llm_response=response
+        llm_response=response,
     )
 
-    print("\n[완료] 분석 완료 ✓")
-    print('\ncontext info')
+    await asyncio.to_thread(append_pipeline_jsonl_log, result, executed_at, request_id, process_timings)
+    print(f"{_req_prefix(request_id)} [로그] JSONL 저장 완료: {PIPELINE_LOG_PATH}")
+
+    print(f"\n{_req_prefix(request_id)} [완료] 분석 완료 ✓")
+    print(f"{_req_prefix(request_id)} context info")
     print(context)
+
+    _emit_status("답변 준비가 끝났어요.", status_callback)
     return result
 
 
@@ -79,14 +241,16 @@ def pipeline(query):
 # CLI 출력 + 진입점
 # ─────────────────────────────────────────────
 
-def print_result(result):
+def print_result(result, include_response=True):
     print(f"\n{'='*60}")
     print(f"분석 결과 | {result.ticker} | {result.analysis_type.upper()}")
     print(f"생성 시각: {result.timestamp}")
     print('='*60)
-    for line in result.llm_response.split("\n"):
-        print(textwrap.fill(line, width=80) if len(line) > 80 else line)
-    print()
+
+    if include_response:
+        for line in result.llm_response.split("\n"):
+            print(textwrap.fill(line, width=80) if len(line) > 80 else line)
+        print()
 
     ctx = result.data_context
     print("[수집 데이터 요약]")
@@ -99,20 +263,20 @@ def print_result(result):
     print(f"  웹 검색 블록 수:     {ctx.get('web_search_blocks', 0)}")
     print(f"  구글 뉴스 요약 포함:   {'예' if ctx.get('google_news') else '아니오'}")
 
-def main():
+
+async def main():
     while True:
         text = input("\n질문> ").strip()
         if text.lower() in ("exit", "quit", "종료"):
             break
         if not text:
             continue
-        result = pipeline(text)
-        print_result(result)
+        result = await pipeline(text, stream_output=True)
+        print_result(result, include_response=False)
 
 
 if __name__ == "__main__":
-    main()
+    asyncio.run(main())
 
 # FastAPI 실행 진입점
 # (uvicorn으로 실행: uvicorn api_server:app --reload)
-

requirements.txt

@@ -8,4 +8,6 @@ tqdm
 jupyter
 openai
 fastapi
-uvicorn
+uvicorn
+dotenv
+gradio