Hierarchical 계층화 — 팀 서브그래프와 상태 격리

def create_team_supervisor(llm: ChatOpenAI, system_prompt, members) -> str:
    """An LLM-based router."""
    options = ["FINISH"] + members
    function_def = {
        "name": "route",
        "description": "Select the next role.",
        "parameters": {
            "title": "routeSchema",
            "type": "object",
            "properties": {
                "next": {"title": "Next", "anyOf": [{"enum": options}]},
            },
            "required": ["next"],
        },
    }
    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), team_members=", ".join(members))
    return (
        prompt
        | llm.bind_functions(functions=[function_def], function_call="route")
        | JsonOutputFunctionsParser()
    )

class ResearchTeamState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]
    team_members: List[str]   # 서로의 역량을 알기 위해 팀원 목록을 상태에 보유
    next: str

research_graph = StateGraph(ResearchTeamState)
research_graph.add_node("Search", search_node)
research_graph.add_node("WebScraper", research_node)
research_graph.add_node("supervisor", supervisor_agent)

research_graph.add_edge("Search", "supervisor")
research_graph.add_edge("WebScraper", "supervisor")
research_graph.add_conditional_edges(
    "supervisor",
    lambda x: x["next"],
    {"Search": "Search", "WebScraper": "WebScraper", "FINISH": END},
)
research_graph.add_edge(START, "supervisor")
chain = research_graph.compile()

class PodcastTeamState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]
    team_members: List[str]
    next: str
    script: List[dict]           # PodcastPlanner가 채움
    audio_segments: List[str]    # AudioProducer가 채움

# (1) 진입 어댑터 — 상위의 문자열 하나를 팀 State 전체로 부풀린다
def enter_chain(message: str):
    return {
        "messages": [HumanMessage(content=message)],
        "team_members": ["PodcastPlanner", "AudioProducer"],
        "next": "",
        "script": [],
        "audio_segments": []
    }

# (2) 추출 어댑터 — 상위 State에서 마지막 메시지 본문만 뽑는다
def get_last_message(state: State) -> str:
    return state["messages"][-1].content

# (3) 복귀 어댑터 — 팀 결과 중 마지막 메시지 1개만 상위로 올린다
def join_graph(response: dict):
    return {"messages": [response["messages"][-1]]}

podcast_chain = enter_chain | chain     # 진입부 결합

super_graph.add_node("ResearchTeam", get_last_message | research_chain | join_graph)
super_graph.add_node("PodcastTeam",  get_last_message | podcast_chain  | join_graph)

class State(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]
    next: str

super_graph = StateGraph(State)
super_graph.add_node("ResearchTeam", get_last_message | research_chain | join_graph)
super_graph.add_node("PodcastTeam", get_last_message | podcast_chain | join_graph)
super_graph.add_node("supervisor", supervisor_node)

super_graph.add_edge("ResearchTeam", "supervisor")
super_graph.add_edge("PodcastTeam", "supervisor")
super_graph.add_conditional_edges(
    "supervisor",
    lambda x: x["next"],
    {"PodcastTeam": "PodcastTeam", "ResearchTeam": "ResearchTeam", "FINISH": END},
)
super_graph.add_edge(START, "supervisor")
super_graph = super_graph.compile()