ailine/backend/app/agent_subgraphs/contact/graph.py

"""
通讯录子图构建器
Contact Subgraph Builder
支持 API 注入的工厂模式
"""

from langgraph.graph import StateGraph, START, END

from .state import ContactState
from .nodes import create_contact_nodes


def build_contact_subgraph(contact_api=None):
    """
    构建通讯录子图（工厂模式）
    
    Args:
        contact_api: 可选的 ContactAPIClient 实例（支持真实数据库或模拟模式）
                     不传入则使用默认模拟 API（向后兼容）
    
    Returns:
        配置好的 StateGraph
    """
    # 创建节点（传入 API）
    nodes = create_contact_nodes(contact_api) if contact_api else None
    
    # 如果没有传入 API，使用向后兼容的导入
    if nodes is None:
        from .nodes import (
            parse_intent,
            list_contacts,
            add_contact,
            list_emails,
            generate_email_draft,
            human_review,
            send_email,
            sniff_contacts,
            format_result,
            should_continue
        )
    else:
        parse_intent = nodes["parse_intent"]
        list_contacts = nodes["list_contacts"]
        add_contact = nodes["add_contact"]
        list_emails = nodes["list_emails"]
        generate_email_draft = nodes["generate_email_draft"]
        human_review = nodes["human_review"]
        send_email = nodes["send_email"]
        sniff_contacts = nodes["sniff_contacts"]
        format_result = nodes["format_result"]
        should_continue = nodes["should_continue"]
    
    # 创建图
    graph = StateGraph(ContactState)
    
    # 添加节点
    graph.add_node("parse_intent", parse_intent)
    graph.add_node("list_contacts", list_contacts)
    graph.add_node("add_contact", add_contact)
    graph.add_node("list_emails", list_emails)
    graph.add_node("generate_email_draft", generate_email_draft)
    graph.add_node("human_review", human_review)
    graph.add_node("send_email", send_email)
    graph.add_node("sniff_contacts", sniff_contacts)
    graph.add_node("format_result", format_result)
    
    # 添加边
    # 从START开始
    graph.add_edge(START, "parse_intent")
    
    # 从parse_intent根据条件路由
    graph.add_conditional_edges(
        "parse_intent",
        should_continue,
        {
            "list_contacts": "list_contacts",
            "add_contact": "add_contact",
            "list_emails": "list_emails",
            "generate_email_draft": "generate_email_draft",
            "sniff_contacts": "sniff_contacts",
        }
    )
    
    # 从各个操作节点到format_result
    graph.add_edge("list_contacts", "format_result")
    graph.add_edge("add_contact", "format_result")
    graph.add_edge("list_emails", "format_result")
    graph.add_edge("sniff_contacts", "format_result")
    
    # 邮件发送的特殊流程
    graph.add_edge("generate_email_draft", "human_review")
    
    # 从human_review根据条件路由
    graph.add_conditional_edges(
        "human_review",
        should_continue,
        {
            "send_email": "send_email",
            "format_result": "format_result",
        }
    )
    
    # 发送邮件后到格式化
    graph.add_edge("send_email", "format_result")
    
    # 最终到END
    graph.add_edge("format_result", END)
    
    return graph