dataflow_agent/graphbuilder/graph_builder.py

# graphbuilder/graph_builder.py
from __future__ import annotations

import asyncio
from typing import Callable, Dict, List, Tuple, Any
from pydantic import BaseModel
from langgraph.graph import StateGraph


class GenericGraphBuilder:
    """
    增强版通用建图器，支持：
    1. @pre_tool 和 @post_tool 装饰器
    2. 链式调用添加节点、边、条件边
    3. 自动工具注册和管理
    """

    def __init__(self, state_model: type[BaseModel], entry_point: str = "start"):
        self.state_model = state_model
        self.entry_point = entry_point
        self.nodes: Dict[str, Tuple[Callable, str]] = {}  # name -> (func, role)
        self.edges: List[Tuple[str, str]] = []
        self.conditional_edges: Dict[str, Callable] = {}
        
        # 工具注册表
        self.pre_tool_registry: Dict[str, Dict[str, Callable]] = {}  # role -> {name: func}
        self.post_tool_registry: Dict[str, List[Callable]] = {}     # role -> [func]
        
        # 延迟导入 tool_manager 避免循环导入
        self.tool_manager = None

    def _get_tool_manager(self):
        """延迟导入 tool_manager"""
        if self.tool_manager is None:
            from dataflow_agent.toolkits.tool_manager import get_tool_manager
            self.tool_manager = get_tool_manager()
        return self.tool_manager

    def pre_tool(self, name: str, role: str):
        """装饰器：注册前置工具到指定角色"""
        def decorator(func: Callable):
            if role not in self.pre_tool_registry:
                self.pre_tool_registry[role] = {}
            self.pre_tool_registry[role][name] = func
            return func
        return decorator

    def post_tool(self, role: str):
        """装饰器：注册后置工具到指定角色"""
        def decorator(func: Callable):
            if role not in self.post_tool_registry:
                self.post_tool_registry[role] = []
            self.post_tool_registry[role].append(func)
            return func
        return decorator

    def add_node(self, name: str, func: Callable, role: str = None) -> 'GenericGraphBuilder':
        """添加单个节点，支持链式调用"""
        self.nodes[name] = (func, role or name)
        return self

    def add_nodes(self, nodes: Dict[str, Callable], role_mapping: Dict[str, str] = None) -> 'GenericGraphBuilder':
        """批量添加节点，支持角色映射"""
        role_mapping = role_mapping or {}
        for name, func in nodes.items():
            role = role_mapping.get(name, name)
            self.add_node(name, func, role)
        return self

    def add_edge(self, src: str, dst: str) -> 'GenericGraphBuilder':
        """添加单条边"""
        self.edges.append((src, dst))
        return self

    def add_edges(self, edges: List[Tuple[str, str]]) -> 'GenericGraphBuilder':
        """批量添加边"""
        self.edges.extend(edges)
        return self

    def add_conditional_edge(self, src: str, condition_func: Callable) -> 'GenericGraphBuilder':
        """添加单个条件边"""
        self.conditional_edges[src] = condition_func
        return self

    def add_conditional_edges(self, conditional_edges: Dict[str, Callable]) -> 'GenericGraphBuilder':
        """批量添加条件边"""
        self.conditional_edges.update(conditional_edges)
        return self

    def _register_tools_for_role(self, role: str, state: Any):
        """为指定角色注册工具"""
        tm = self._get_tool_manager()
        
        # 注册前置工具
        if role in self.pre_tool_registry:
            for tool_name, tool_func in self.pre_tool_registry[role].items():
                try:
                    tm.register_pre_tool(
                        name=tool_name,
                        role=role,
                        func=lambda s=state, f=tool_func: f(s),
                        override=True
                    )
                except TypeError:
                    # 兼容不支持 override 参数的版本
                    tm.register_pre_tool(
                        name=tool_name,
                        role=role,
                        func=lambda s=state, f=tool_func: f(s)
                    )

        # 注册后置工具
        if role in self.post_tool_registry:
            for tool_func in self.post_tool_registry[role]:
                tm.register_post_tool(tool_func, role=role)

    def _wrap_node_with_tools(self, node_func: Callable, role: str):
        """为节点包装自动工具注册逻辑"""
        async def wrapped_node(state):
            # 执行前自动注册该角色的工具
            self._register_tools_for_role(role, state)
            
            # 执行原始节点函数
            if asyncio.iscoroutinefunction(node_func):
                return await node_func(state)
            else:
                return node_func(state)
        
        return wrapped_node

    def build(self):
        """构建并返回编译后的图"""
        sg = StateGraph(self.state_model)
        
        # 添加节点（自动包装工具注册逻辑）
        for name, (func, role) in self.nodes.items():
            wrapped_func = self._wrap_node_with_tools(func, role)
            sg.add_node(name, wrapped_func)
        
        # 添加普通边
        for src, dst in self.edges:
            sg.add_edge(src, dst)
        
        # 添加条件边
        for src, cond_func in self.conditional_edges.items():
            sg.add_conditional_edges(src, cond_func)
        
        sg.set_entry_point(self.entry_point)
        return sg.compile()