基于图结构的Agent推理链构建

传统的Agent推理链以线性或树形结构为主，链式思维CoT是一条直线，思维树ToT是一棵树。但真实的推理过程往往比这两种结构更复杂——推理步骤之间可能存在交叉引用、反馈循环和非层次化的依赖关系。图结构的推理链（Graph of Thoughts, GoT）将推理过程建模为有向图，打破了树形结构的限制，允许更灵活的推理模式。本文将系统阐述图思维GoT的原理、推理图构建、节点关系建模、图遍历推理、循环推理处理，并提供完整的代码实现。

一、从树到图：推理结构的进化

树形结构有一个根本限制：每个节点只有一个父节点。这意味着每条推理路径是独立的，不能合并或交叉。但在实际推理中，不同路径经常需要合并——两条独立推理得出的中间结论可能共同支撑同一个最终结论。树结构无法表达这种合并关系，导致推理冗余。

图结构打破了这一限制。在有向图中，一个节点可以有多个前驱节点，表示该节点的结论由多个推理路径共同支撑。这种结构更自然地表达了推理中的"综合"和"聚合"操作。

图结构还支持反馈循环。在某些推理场景中，后续推理的结果可以反馈到前面的步骤，触发重新推理。比如在迭代优化中，第一轮推理产生的草案经过评估后，反馈到推理起点触发第二轮改进。树结构无法表达循环，图结构可以。

from dataclasses import dataclass, field
from typing import List, Optional, Dict, Any, Set, Callable, Tuple
from enum import Enum
from collections import defaultdict, deque
import time

class NodeType(Enum):
    INPUT = "input"
    REASONING = "reasoning"
    AGGREGATION = "aggregation"
    EVALUATION = "evaluation"
    OUTPUT = "output"
    FEEDBACK = "feedback"

class EdgeType(Enum):
    FORWARD = "forward"
    AGGREGATE = "aggregate"
    FEEDBACK = "feedback"
    CONDITIONAL = "conditional"

@dataclass
class ReasoningNode:
    """推理图节点"""
    id: str
    node_type: NodeType
    content: str
    state: Dict[str, Any]
    score: float = 0.0
    depth: int = 0
    is_visited: bool = False
    is_active: bool = True
    metadata: Dict[str, Any] = field(default_factory=dict)
    predecessors: Set[str] = field(default_factory=set)
    successors: Set[str] = field(default_factory=set)

@dataclass
class ReasoningEdge:
    """推理图边"""
    source: str
    target: str
    edge_type: EdgeType
    weight: float = 1.0
    condition: Optional[Callable] = None
    label: str = ""

class ReasoningGraph:
    """推理图数据结构"""

    def __init__(self):
        self.nodes: Dict[str, ReasoningNode] = {}
        self.edges: Dict[str, ReasoningEdge] = {}
        self.adjacency: Dict[str, List[str]] = defaultdict(list)
        self.reverse_adjacency: Dict[str, List[str]] = defaultdict(list)
        self._node_counter = 0

    def add_node(self, content: str, node_type: NodeType,
                 state: Dict[str, Any] = None, score: float = 0.0) -> ReasoningNode:
        node_id = f"node_{self._node_counter}"
        self._node_counter += 1
        node = ReasoningNode(id=node_id, node_type=node_type, content=content,
                             state=state or {}, score=score)
        self.nodes[node_id] = node
        return node

    def add_edge(self, source: str, target: str,
                 edge_type: EdgeType = EdgeType.FORWARD,
                 weight: float = 1.0, condition: Callable = None,
                 label: str = "") -> ReasoningEdge:
        edge_id = f"{source}->{target}"
        edge = ReasoningEdge(source=source, target=target, edge_type=edge_type,
                             weight=weight, condition=condition, label=label)
        self.edges[edge_id] = edge
        self.adjacency[source].append(target)
        self.reverse_adjacency[target].append(source)
        self.nodes[source].successors.add(target)
        self.nodes[target].predecessors.add(source)
        return edge

    def get_predecessors(self, node_id: str) -> List[ReasoningNode]:
        return [self.nodes[p] for p in self.reverse_adjacency[node_id]
                if p in self.nodes and self.nodes[p].is_active]

    def get_successors(self, node_id: str) -> List[ReasoningNode]:
        return [self.nodes[s] for s in self.adjacency[node_id]
                if s in self.nodes and self.nodes[s].is_active]

    def get_roots(self) -> List[ReasoningNode]:
        return [n for n in self.nodes.values()
                if not n.predecessors and n.is_active]

    def topological_sort(self) -> List[str]:
        """拓扑排序（忽略反馈边）"""
        in_degree: Dict[str, int] = {}
        for nid, node in self.nodes.items():
            if not node.is_active:
                continue
            forward_preds = [p for p in self.reverse_adjacency[nid]
                             if p in self.nodes and self.nodes[p].is_active
                             and self.edges.get(f"{p}->{nid}")
                             and self.edges[f"{p}->{nid}"].edge_type != EdgeType.FEEDBACK]
            in_degree[nid] = len(forward_preds)
        queue = deque([nid for nid, d in in_degree.items() if d == 0])
        result = []
        while queue:
            nid = queue.popleft()
            result.append(nid)
            for succ in self.adjacency[nid]:
                if succ not in in_degree:
                    continue
                edge = self.edges.get(f"{nid}->{succ}")
                if edge and edge.edge_type == EdgeType.FEEDBACK:
                    continue
                in_degree[succ] -= 1
                if in_degree[succ] == 0:
                    queue.append(succ)
        return result

    def get_path(self, start: str, end: str) -> Optional[List[str]]:
        visited: Set[str] = set()
        queue = deque([(start, [start])])
        while queue:
            current, path = queue.popleft()
            if current == end:
                return path
            if current in visited:
                continue
            visited.add(current)
            for succ in self.adjacency[current]:
                if succ not in visited:
                    queue.append((succ, path + [succ]))
        return None

    def detect_cycles(self) -> List[List[str]]:
        """检测图中的所有环"""
        cycles = []
        visited: Set[str] = set()
        rec_stack: Set[str] = set()
        for nid in self.nodes:
            if nid not in visited:
                self._detect_cycles_dfs(nid, visited, rec_stack, [], cycles)
        return cycles

    def _detect_cycles_dfs(self, node: str, visited: Set[str],
                           rec_stack: Set[str], path: List[str],
                           cycles: List[List[str]]):
        visited.add(node)
        rec_stack.add(node)
        path.append(node)
        for succ in self.adjacency[node]:
            if succ not in visited:
                self._detect_cycles_dfs(succ, visited, rec_stack, path, cycles)
            elif succ in rec_stack:
                cycle_start = path.index(succ)
                cycles.append(path[cycle_start:] + [succ])
        path.pop()
        rec_stack.discard(node)

    def get_summary(self) -> Dict[str, Any]:
        return {
            "total_nodes": len(self.nodes),
            "active_nodes": sum(1 for n in self.nodes.values() if n.is_active),
            "total_edges": len(self.edges),
            "has_cycles": len(self.detect_cycles()) > 0,
        }

推理图数据结构是整个GoT框架的基础。ReasoningGraph 支持节点和边的增删查改，提供拓扑排序（自动忽略反馈边以处理循环）、路径查找和环检测等核心操作。detect_cycles 方法使用DFS和递归栈检测图中的所有环，这对于循环推理的处理至关重要。

二、推理图构建

推理图的构建是一个增量过程。Agent从输入节点开始，逐步生成推理节点并通过边连接它们。与ToT不同，GoT允许在构建过程中添加聚合边（合并多条路径）和反馈边（创建循环）。

推理图构建有三种基本操作：扩展（从现有节点生成新的后继节点）、聚合（将多个节点的推理结果合并到一个新节点）和反馈（从后续节点创建回到前面节点的边，触发重新推理）。

class GraphBuilder:
    """推理图构建器"""

    def __init__(self, graph: ReasoningGraph, llm_call: Callable):
        self.graph = graph
        self.llm_call = llm_call
        self.build_history: List[Dict] = []

    def initialize(self, problem: str) -> ReasoningNode:
        node = self.graph.add_node(
            content=problem, node_type=NodeType.INPUT,
            state={"problem": problem, "step": 0},
        )
        self.build_history.append({"action": "init", "node_id": node.id})
        return node

    def expand(self, source_id: str, problem: str,
               num_branches: int = 2) -> List[ReasoningNode]:
        source = self.graph.nodes[source_id]
        children = []
        for i in range(num_branches):
            prompt = (
                f"问题: {problem}\n当前推理: {source.content}\n"
                f"请给出第{i+1}种不同的下一步推理:"
            )
            response = self.llm_call(prompt, temperature=0.6 + 0.1 * i,
                                     max_tokens=200)
            thought = response.strip().split("\n")[0]
            child = self.graph.add_node(
                content=thought, node_type=NodeType.REASONING,
                state={**source.state, "last_thought": thought,
                       "step": source.state.get("step", 0) + 1, "branch": i},
            )
            self.graph.add_edge(source_id, child.id, EdgeType.FORWARD,
                                label=f"branch_{i}")
            children.append(child)
        self.build_history.append(
            {"action": "expand", "source": source_id,
             "children": [c.id for c in children]})
        return children

    def aggregate(self, source_ids: List[str], problem: str) -> ReasoningNode:
        sources = [self.graph.nodes[sid] for sid in source_ids]
        contents = [s.content for s in sources]
        contents_text = "\n".join(f"  路径{i+1}: {c}" for i, c in enumerate(contents))
        prompt = (
            f"问题: {problem}\n多条推理路径:\n{contents_text}\n"
            f"请综合以上路径，给出统一的推理结论:"
        )
        response = self.llm_call(prompt, temperature=0.3, max_tokens=300)
        node = self.graph.add_node(
            content=response.strip(), node_type=NodeType.AGGREGATION,
            state={"aggregated_from": source_ids,
                   "step": max(s.state.get("step", 0) for s in sources) + 1},
        )
        for sid in source_ids:
            self.graph.add_edge(sid, node.id, EdgeType.AGGREGATE)
        self.build_history.append(
            {"action": "aggregate", "sources": source_ids, "result": node.id})
        return node

    def evaluate(self, node_id: str, problem: str) -> float:
        node = self.graph.nodes[node_id]
        prompt = (f"问题: {problem}\n推理步骤: {node.content}\n"
                  f"评估前景(0-1)，只输出数字:")
        response = self.llm_call(prompt, temperature=0.0, max_tokens=10)
        try:
            node.score = max(0.0, min(1.0, float(response.strip())))
        except ValueError:
            node.score = 0.5
        return node.score

    def add_feedback(self, from_id: str, to_id: str, label: str = "feedback"):
        self.graph.add_edge(from_id, to_id, EdgeType.FEEDBACK, label=label)
        self.build_history.append(
            {"action": "feedback", "from": from_id, "to": to_id})

图构建器实现了四种核心操作。expand 从一个源节点生成多个推理分支，类似ToT的思维生成。aggregate 将多个节点的推理内容合并为一个综合结论，这是GoT区别于ToT的关键操作。evaluate 对节点进行评估。add_feedback 添加反馈边创建推理循环。

三、节点关系建模

推理图中的节点关系远比树结构丰富。除了简单的前驱-后继关系外，还存在聚合关系（多个前驱共同支撑一个后继）、条件关系（只有满足条件才激活的边）和反馈关系（从后继指向前驱的边）。

class RelationAnalyzer:
    """节点关系分析器"""

    def __init__(self, graph: ReasoningGraph):
        self.graph = graph

    def get_aggregation_groups(self) -> List[Dict[str, Any]]:
        groups = defaultdict(list)
        for edge in self.graph.edges.values():
            if edge.edge_type == EdgeType.AGGREGATE:
                groups[edge.target].append(edge.source)
        return [{"aggregation_node": t, "sources": s}
                for t, s in groups.items()]

    def get_feedback_loops(self) -> List[Dict[str, Any]]:
        cycles = self.graph.detect_cycles()
        loops = []
        for cycle in cycles:
            feedback_edges = []
            for i in range(len(cycle) - 1):
                edge = self.graph.edges.get(f"{cycle[i]}->{cycle[i+1]}")
                if edge and edge.edge_type == EdgeType.FEEDBACK:
                    feedback_edges.append(f"{cycle[i]}->{cycle[i+1]}")
            if feedback_edges:
                loops.append({"path": cycle, "feedback_edges": feedback_edges,
                              "length": len(cycle) - 1})
        return loops

    def compute_node_importance(self) -> Dict[str, float]:
        """基于PageRank变体计算节点重要性"""
        nodes = [n for n in self.graph.nodes.values() if n.is_active]
        if not nodes:
            return {}
        node_ids = [n.id for n in nodes]
        n = len(node_ids)
        scores = {nid: 1.0 / n for nid in node_ids}
        damping = 0.85
        for _ in range(20):
            new_scores = {}
            for nid in node_ids:
                incoming = self.graph.reverse_adjacency[nid]
                incoming_score = 0.0
                for pred_id in incoming:
                    if pred_id in scores:
                        pred = self.graph.nodes[pred_id]
                        out_count = len([s for s in pred.successors if s in scores])
                        if out_count > 0:
                            edge = self.graph.edges.get(f"{pred_id}->{nid}")
                            weight = edge.weight if edge else 1.0
                            incoming_score += scores[pred_id] * weight / out_count
                new_scores[nid] = (1 - damping) / n + damping * incoming_score
            scores = new_scores
        total = sum(scores.values())
        if total > 0:
            scores = {k: v / total for k, v in scores.items()}
        return scores

    def find_redundant_nodes(self, threshold: float = 0.01) -> List[str]:
        importance = self.compute_node_importance()
        return [nid for nid, score in importance.items() if score < threshold]

关系分析器提供了聚合组识别、反馈循环分析、节点重要性计算和冗余节点检测四种功能。compute_node_importance 使用PageRank变体计算每个节点在推理图中的重要性，帮助识别关键推理步骤。

四、图遍历推理

图遍历推理是GoT的执行引擎，负责按照图结构推进推理过程。与ToT的树搜索不同，图遍历需要处理聚合节点（等待所有前驱完成才能执行）和反馈边（触发循环推理）。

class GraphTraversalEngine:
    """图遍历推理引擎"""

    def __init__(self, graph: ReasoningGraph, evaluator: Callable):
        self.graph = graph
        self.evaluator = evaluator
        self.execution_log: List[Dict] = []
        self.max_iterations = 50

    def traverse(self, problem: str) -> Dict[str, Any]:
        topo_order = self.graph.topological_sort()
        completed: Set[str] = set()
        for iteration in range(self.max_iterations):
            progress = False
            for node_id in topo_order:
                if node_id in completed:
                    continue
                node = self.graph.nodes[node_id]
                if not node.is_active:
                    completed.add(node_id)
                    continue
                if not self._is_ready(node_id, completed):
                    continue
                self._execute_node(node_id, problem)
                completed.add(node_id)
                progress = True
                if node.node_type == NodeType.OUTPUT:
                    return self._build_result(node_id, completed)
            if not progress:
                break
            if self._process_feedback(completed):
                completed = self._reset_for_feedback(completed)
                topo_order = self.graph.topological_sort()
        best = self._find_best_output()
        return self._build_result(best, completed) if best else \
            {"success": False, "reason": "no_output"}

    def _is_ready(self, node_id: str, completed: Set[str]) -> bool:
        for pred_id in self.graph.reverse_adjacency[node_id]:
            edge = self.graph.edges.get(f"{pred_id}->{node_id}")
            if not edge or edge.edge_type in (EdgeType.FEEDBACK, EdgeType.CONDITIONAL):
                continue
            if pred_id not in completed:
                return False
        return True

    def _execute_node(self, node_id: str, problem: str):
        node = self.graph.nodes[node_id]
        node.is_visited = True
        if node.node_type == NodeType.AGGREGATION:
            preds = self.graph.reverse_adjacency[node_id]
            node.state["aggregated_contents"] = [
                self.graph.nodes[p].content for p in preds
                if p in self.graph.nodes]
        score = self.evaluator(node.content, node.state, problem)
        node.score = score
        self.execution_log.append(
            {"node_id": node_id, "type": node.node_type.value,
             "score": score, "timestamp": time.time()})
        if score < 0.1 and node.node_type == NodeType.REASONING:
            node.is_active = False

    def _process_feedback(self, completed: Set[str]) -> bool:
        triggered = False
        for edge in self.graph.edges.values():
            if edge.edge_type != EdgeType.FEEDBACK:
                continue
            if edge.source in completed and edge.target in completed:
                source = self.graph.nodes[edge.source]
                if source.score < 0.5:
                    target = self.graph.nodes[edge.target]
                    target.state["feedback"] = source.content
                    target.is_visited = False
                    triggered = True
        return triggered

    def _reset_for_feedback(self, completed: Set[str]) -> Set[str]:
        new_completed: Set[str] = set()
        for nid in completed:
            node = self.graph.nodes[nid]
            if node.node_type in (NodeType.INPUT, NodeType.OUTPUT):
                new_completed.add(nid)
        return new_completed

    def _find_best_output(self) -> Optional[str]:
        outputs = [n for n in self.graph.nodes.values()
                   if n.node_type == NodeType.OUTPUT and n.is_active]
        if not outputs:
            return None
        return max(outputs, key=lambda n: n.score).id

    def _build_result(self, node_id: str, completed: Set[str]) -> Dict[str, Any]:
        node = self.graph.nodes[node_id]
        roots = self.graph.get_roots()
        path = self.graph.get_path(roots[0].id, node_id) if roots else [node_id]
        return {"success": True, "answer": node.content, "score": node.score,
                "completed_nodes": len(completed),
                "total_nodes": len(self.graph.nodes), "path": path}

图遍历引擎按照拓扑排序执行节点，遇到聚合节点时等待所有前驱完成，遇到低分节点时剪枝。反馈处理是GoT的特色功能：当反馈边的源节点分数较低时，触发目标节点的重新推理。

五、循环推理处理

循环推理是GoT区别于ToT的核心能力。在很多推理场景中，单次推理的结果不够好，需要基于评估反馈进行多轮迭代优化。GoT通过反馈边显式建模这种循环，使得推理图可以表达"推理-评估-改进-再推理"的迭代过程。

class CyclicReasoningHandler:
    """循环推理处理器"""

    def __init__(self, graph: ReasoningGraph, builder: GraphBuilder,
                 evaluator: Callable):
        self.graph = graph
        self.builder = builder
        self.evaluator = evaluator
        self.iteration_history: List[Dict] = []
        self.termination = {
            "max_iterations": 5, "score_threshold": 0.85,
            "convergence_delta": 0.02, "convergence_count": 3,
            "stagnation_count": 3,
        }

    def run_cyclic_reasoning(self, problem: str, input_id: str,
                             output_id: str) -> Dict[str, Any]:
        max_iter = self.termination["max_iterations"]
        threshold = self.termination["score_threshold"]
        conv_delta = self.termination["convergence_delta"]
        conv_count = self.termination["convergence_count"]
        stag_count = self.termination["stagnation_count"]
        best_score = 0.0
        best_content = ""
        conv_streak = 0
        stag_streak = 0
        prev_score = 0.0
        reason = "max_iterations"

        for iteration in range(max_iter):
            output = self.graph.nodes[output_id]
            score = output.score
            content = output.content
            if score > best_score:
                best_score = score
                best_content = content
            self.iteration_history.append(
                {"iteration": iteration + 1, "score": score})
            if score >= threshold:
                reason = "score_threshold_reached"
                break
            if iteration > 0:
                delta = abs(score - prev_score)
                if delta < conv_delta:
                    conv_streak += 1
                    if conv_streak >= conv_count:
                        reason = "convergence"
                        break
                else:
                    conv_streak = 0
                if score <= prev_score:
                    stag_streak += 1
                    if stag_streak >= stag_count:
                        reason = "stagnation"
                        break
                else:
                    stag_streak = 0
            prev_score = score
            self._trigger_refinement(input_id, output_id, problem, iteration)
        return {"best_score": best_score, "best_content": best_content,
                "total_iterations": len(self.iteration_history),
                "terminated_reason": reason,
                "history": self.iteration_history}

    def _trigger_refinement(self, input_id: str, output_id: str,
                            problem: str, iteration: int):
        output = self.graph.nodes[output_id]
        feedback_node = self.graph.add_node(
            content=f"改进建议(第{iteration+1}轮): 分数{output.score}",
            node_type=NodeType.FEEDBACK,
            state={"iteration": iteration + 1, "previous_score": output.score})
        self.graph.add_edge(output_id, feedback_node.id, EdgeType.FORWARD)
        self.graph.add_edge(feedback_node.id, input_id, EdgeType.FEEDBACK,
                            label=f"refine_{iteration+1}")
        input_node = self.graph.nodes[input_id]
        input_node.state["feedback"] = feedback_node.content
        input_node.state["iteration"] = iteration + 1
        input_node.is_visited = False

循环推理处理器实现了完整的迭代优化循环。每轮迭代检查当前输出分数，根据终止条件决定是否继续。四种终止条件确保循环不会无限运行：目标分数达成、分数收敛、质量停滞和最大迭代次数。

六、完整图推理引擎集成

将推理图、图构建器、关系分析器、遍历引擎和循环处理器集成在一起。

class GraphReasoningEngine:
    """完整的图推理引擎"""

    def __init__(self, llm_call: Callable):
        self.llm_call = llm_call
        self.metrics: Dict[str, Any] = {
            "total_queries": 0, "successful": 0,
            "avg_nodes": 0.0, "avg_score": 0.0,
        }

    def solve(self, problem: str, use_aggregation: bool = True,
              use_feedback: bool = True) -> Dict[str, Any]:
        self.metrics["total_queries"] += 1
        graph = ReasoningGraph()
        builder = GraphBuilder(graph, self.llm_call)

        def evaluator(content: str, state: Dict, prob: str) -> float:
            prompt = f"问题: {prob}\n推理: {content}\n评估前景(0-1)，只输出数字:"
            resp = self.llm_call(prompt, temperature=0.0, max_tokens=10)
            try:
                return max(0.0, min(1.0, float(resp.strip())))
            except ValueError:
                return 0.5

        input_node = builder.initialize(problem)
        branches = builder.expand(input_node.id, problem, num_branches=2)
        for branch in branches:
            sub_branches = builder.expand(branch.id, problem, num_branches=2)
            for sb in sub_branches:
                builder.evaluate(sb.id, problem)
                builder.evaluate(branch.id, problem)

        output_node = None
        if use_aggregation and len(branches) >= 2:
            best_branches = sorted(branches, key=lambda n: n.score, reverse=True)[:2]
            agg_sources = []
            for b in best_branches:
                succs = graph.get_successors(b.id)
                if succs:
                    best_sub = max(succs, key=lambda n: n.score)
                    agg_sources.append(best_sub.id)
            if agg_sources:
                output_node = builder.aggregate(agg_sources, problem)
                output_node.node_type = NodeType.OUTPUT
        else:
            all_nodes = [n for n in graph.nodes.values()
                         if n.node_type == NodeType.REASONING]
            if all_nodes:
                best = max(all_nodes, key=lambda n: n.score)
                output_node = graph.add_node(
                    content=best.content, node_type=NodeType.OUTPUT,
                    state=best.state, score=best.score)
                graph.add_edge(best.id, output_node.id, EdgeType.FORWARD)

        if output_node:
            builder.evaluate(output_node.id, problem)

        result_score = output_node.score if output_node else 0.0
        if use_feedback and output_node and result_score < 0.85:
            handler = CyclicReasoningHandler(graph, builder, evaluator)
            cycle_result = handler.run_cyclic_reasoning(
                problem, input_node.id, output_node.id)
            result_score = cycle_result["best_score"]
            if cycle_result["best_content"]:
                output_node.content = cycle_result["best_content"]
                output_node.score = result_score

        analyzer = RelationAnalyzer(graph)
        importance = analyzer.compute_node_importance()
        redundant = analyzer.find_redundant_nodes()

        self.metrics["successful"] += 1
        n = self.metrics["total_queries"]
        self.metrics["avg_nodes"] = (
            self.metrics["avg_nodes"] * (n - 1) + len(graph.nodes)) / n
        self.metrics["avg_score"] = (
            self.metrics["avg_score"] * (n - 1) + result_score) / n

        return {
            "success": True,
            "answer": output_node.content if output_node else "",
            "score": result_score,
            "graph_summary": graph.get_summary(),
            "node_importance": importance,
            "redundant_nodes": redundant,
        }

    def get_metrics(self) -> Dict[str, Any]:
        return self.metrics


if __name__ == "__main__":
    def mock_llm(prompt: str, temperature: float = 0.7,
                 max_tokens: int = 200) -> str:
        if "评估前景" in prompt and "只输出数字" in prompt:
            return "0.8"
        if "综合以上路径" in prompt:
            return "综合两条路径的结论，最终答案是42"
        if "第1种" in prompt:
            return "通过数学计算推导"
        if "第2种" in prompt:
            return "通过逻辑推理推导"
        return "推理步骤完成"

    engine = GraphReasoningEngine(llm_call=mock_llm)
    result = engine.solve("23+19等于多少？", use_aggregation=True, use_feedback=False)
    print(f"答案: {result['answer']}")
    print(f"分数: {result['score']}")
    print(f"图摘要: {result['graph_summary']}")
    print(f"指标: {engine.get_metrics()}")

集成引擎将所有组件串联为完整的推理流程：初始化输入节点、扩展推理分支、评估各节点、聚合最优路径、可选的循环优化、关系分析和冗余检测。mock_llm 模拟了LLM调用，实际使用时替换为真实API即可。

七、GoT与ToT、CoT的对比

三种推理结构代表了不同的复杂度和能力层级。CoT是线性结构，适合简单推理；ToT是树形结构，适合需要搜索的复杂推理；GoT是图结构，适合需要聚合和迭代优化的复杂推理。

在表达能力上，GoT是CoT和ToT的超集——线性链和树都可以看作图的特例。GoT额外支持聚合操作（多路径合并）和循环操作（迭代优化），这两种能力是CoT和ToT无法表达的。

在计算开销上，GoT通常高于ToT，因为聚合操作需要额外的LLM调用来综合多条路径，循环操作需要多轮迭代。但GoT的聚合操作可以减少冗余推理——多条路径的公共部分只需推理一次，然后聚合各自独特的部分。

在适用场景上，GoT最适合那些需要多视角综合和迭代改进的任务。比如论文写作需要综合多个角度的论点，代码优化需要多轮迭代改进，决策制定需要综合多条分析路径的结论。

八、工程实践建议

在实际部署GoT时，有几个关键经验值得注意。

首先是聚合策略的选择。并非所有多路径推理都需要聚合——如果各路径的结论高度一致，聚合的边际价值很低但开销不小。建议设置一个差异度阈值，只有当路径结论的差异度超过阈值时才触发聚合。

其次是循环深度的控制。循环推理虽然能提升质量，但每轮循环的边际收益递减。建议设置最大循环次数为3到5轮，并配合收敛检测提前终止。

第三是图规模的控制。大型推理图虽然表达能力强，但遍历和维护开销也大。建议通过重要性分析定期剪枝低重要性节点，保持图的精简。

第四是并行化。GoT的扩展操作（同一节点的多个分支）和评估操作（多个独立节点）都可以并行执行。利用异步IO可以显著减少总推理延迟。

九、总结

图结构推理链GoT通过将推理过程建模为有向图，突破了树形结构的限制，支持聚合操作和循环推理两种高级推理模式。本文从推理图数据结构出发，详细实现了图构建器、关系分析器、图遍历引擎和循环推理处理器，并提供了完整的集成引擎。

GoT的核心价值在于它能够表达"多路径综合"和"迭代优化"这两种在复杂推理中常见但CoT和ToT无法处理的模式。通过聚合操作，GoT可以合并多条推理路径的优势结论；通过循环操作，GoT可以基于评估反馈持续改进推理质量。

在实际应用中，GoT最适合需要多视角综合和迭代改进的复杂推理任务。随着大模型推理能力的不断增强，图结构推理链有望成为Agent系统的标准推理框架，为更复杂、更可靠的AI推理提供基础设施支撑。