Supervisor 구현과 그 대가 — 오염·루프·비용

members = ["Researcher", "Stock_Analyzer", "Chart_Generator"]
options = ["FINISH"] + members

system_prompt = (
    "You are a supervisor tasked with managing a conversation between the"
    " following workers:  {members}. Given the following user request,"
    " respond with the worker to act next. Each worker will perform a"
    " task and respond with their results and status. When finished,"
    " respond with FINISH."
)

class routeResponse(BaseModel):
    next: Literal[*options]          # ← 허용값을 타입으로 못 박는다

prompt = ChatPromptTemplate.from_messages([
    ("system", system_prompt),
    MessagesPlaceholder(variable_name="messages"),
    ("system",
     "Given the conversation above, who should act next?"
     " Or should we FINISH? Select one of: {options}"),
]).partial(options=str(options), members=", ".join(members))

def supervisor_agent(state):
    supervisor_chain = prompt | llm.with_structured_output(routeResponse)
    return supervisor_chain.invoke(state)   # → {"next": "Researcher"} 형태로 상태에 병합

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]
    next: str          # ← Supervisor가 채우는 라우팅 필드

def agent_node(state, agent, name):
    result = agent.invoke(state)
    # 핵심: 워커 결과를 AIMessage가 아니라 name 태그가 붙은 HumanMessage로 되돌린다
    return {"messages": [HumanMessage(content=result["messages"][-1].content, name=name)]}

research_agent = create_react_agent(llm, tools=[tavily_tool],
                                    state_modifier=research_system_prompt)
research_node = functools.partial(agent_node, agent=research_agent, name="Researcher")

stock_agent = create_react_agent(llm, tools=[stock_analysis],
                                 state_modifier=stock_system_prompt)
stock_node = functools.partial(agent_node, agent=stock_agent, name="Stock_Analyzer")

chart_agent = create_react_agent(llm, tools=[python_repl_tool],
                                 state_modifier=chart_system_prompt)
chart_node = functools.partial(agent_node, agent=chart_agent, name="Chart_Generator")

workflow = StateGraph(AgentState)
workflow.add_node("Researcher", research_node)
workflow.add_node("Stock_Analyzer", stock_node)
workflow.add_node("Chart_Generator", chart_node)
workflow.add_node("supervisor", supervisor_agent)

# (1) 모든 워커는 반드시 supervisor로 복귀 — Supervisor 패턴의 정의 그 자체
for member in members:
    workflow.add_edge(member, "supervisor")

# (2) supervisor가 채운 state["next"] 값을 그대로 노드 이름으로 사용
conditional_map = {k: k for k in members}
conditional_map["FINISH"] = END

workflow.add_conditional_edges(
    "supervisor",
    lambda x: x["next"],      # ← 라우팅 함수가 단 한 줄
    conditional_map)

workflow.add_edge(START, "supervisor")
graph = workflow.compile()

@tool
def stock_analysis(ticker: str) -> str:
    """
    주어진 주식 티커에 대한 업데이트된 종합적인 재무 분석을 수행합니다.
    최신 주가 정보, 재무 지표, 성장률, 밸류에이션 및 주요 비율을 제공합니다.
    가장 최근 영업일 기준의 데이터를 사용합니다.
    """
    ticker = yf.Ticker(ticker)
    historical_prices = ticker.history(period='5d', interval='1d')
    last_5_days_close = historical_prices['Close'].tail(5)
    annual_financials = ticker.get_financials()
    quarterly_financials = ticker.get_financials(freq="quarterly")
    return str({
        "최근 5일간 종가": {...},
        "연간 재무제표 요약": format_financial_summary(annual_financials),
        "분기별 재무제표 요약": format_financial_summary(quarterly_financials),
    })

for s in super_graph.stream(
    {"messages": [HumanMessage(content="OpenAI가 최근에 출시한 o1 모델과 AI 에이전트")]},
    {"recursion_limit": 150},          # ← 기본값 25로는 3층 구조가 완주하지 못한다
):
    if "__end__" not in s:
        print(s)