Graph-of-Thought(思维图谱)
Graph-of-Thought(GoT) 是 ToT 的进一步扩展,由 Besta et al. 于 2023 年提出。其核心思想是将推理过程建模为有向图,节点代表思维单元,边代表推理关系。相比 ToT 的树结构,GoT 支持更复杂的推理模式,包括聚合、分解和循环。
一、核心原理
1.1 GoT 的设计哲学
GoT 解决了 ToT 无法处理复杂依赖关系的问题:
┌─────────────────────────────────────────────────────────────┐
│ CoT → ToT → GoT 演进 │
├─────────────────────────────────────────────────────────────┤
│ │
│ Chain-of-Thought(线性): │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ A ─→ B ─→ C ─→ D │ │
│ │ ❌ 单路径,无分支 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ Tree-of-Thought(树状): │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ A │ │
│ │ ↙ ↘ │ │
│ │ B C │ │
│ │ ↙ ↘ ↙ ↘ │ │
│ │ D E F G │ │
│ │ ✅ 分支,但无法聚合 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ Graph-of-Thought(图状): │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ A │ │
│ │ ↙ ↘ │ │
│ │ B C ←────┐ │ │
│ │ ↓ ↓ │ │ │
│ │ D ─→ E ────→F (聚合) │ │
│ │ ↑ │ │ │
│ │ └──────┘ (循环) │ │
│ │ ✅ 分支 + 聚合 + 循环 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘1.2 GoT 核心操作
| 操作 | 描述 | 示意图 |
|---|---|---|
| 分解 | 将一个思维拆分为多个子思维 | A → B, C, D |
| 聚合 | 将多个思维合并为一个 | B, C, D → E |
| 转换 | 对思维进行修改/细化 | A → A' |
| 循环 | 思维之间的迭代推理 | A → B → A' |
┌─────────────────────────────────────────────────────────────┐
│ GoT 四种核心操作 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 1. 分解(Decomposition) │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ │ │
│ │ ┌──→ 子问题1 │ │
│ │ 问题 ─┼──→ 子问题2 │ │
│ │ └──→ 子问题3 │ │
│ │ │ │
│ │ 示例:分析市场份额 │ │
│ │ 市场份额 ─→ 竞争对手分析 │ │
│ │ ─→ 用户群体分析 │ │
│ │ ─→ 产品差异化分析 │ │
│ │ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 2. 聚合(Aggregation) │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ │ │
│ │ 子问题1 ──┐ │ │
│ │ 子问题2 ──┼──→ 综合结论 │ │
│ │ 子问题3 ──┘ │ │
│ │ │ │
│ │ 示例:综合分析报告 │ │
│ │ 竞争对手分析 ─┐ │ │
│ │ 用户群体分析 ─┼──→ 市场份额报告 │ │
│ │ 产品差异分析 ─┘ │ │
│ │ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 3. 转换(Transformation) │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ │ │
│ │ 思维A ──→ 思维A'(改进版) │ │
│ │ │ │
│ │ 示例:迭代优化 │ │
│ │ 初稿 ──→ 修改稿 ──→ 终稿 │ │
│ │ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 4. 循环(Iteration) │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ │ │
│ │ ┌──────────┐ │ │
│ │ │ ↓ │ │
│ │ 假设 ─→ 验证 ─→ 修正假设 ─→ 再验证... │ │
│ │ │ │
│ │ 示例:科学推理 │ │
│ │ 提出假设 ─→ 实验验证 ─→ 修正假设 ─→ ... │ │
│ │ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘1.3 与 CoT/ToT 的对比
| 维度 | CoT | ToT | GoT |
|---|---|---|---|
| 结构 | 线性链 | 树 | 图 |
| 分支 | ❌ | ✅ | ✅ |
| 聚合 | ❌ | ❌ | ✅ |
| 循环 | ❌ | ❌ | ✅ |
| 复杂度 | 低 | 中 | 高 |
| 适用场景 | 简单推理 | 多路径探索 | 复杂依赖推理 |
二、工作流程
2.1 完整工作流程
┌─────────────────────────────────────────────────────────────────────┐
│ GoT 完整工作流程 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ │
│ │ 用户输入 │ │
│ │ (Query) │ │
│ └──────┬──────┘ │
│ │ │
│ ↓ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ 初始化思维图 │ │
│ │ graph = {nodes: [], edges: []} │ │
│ │ root = create_node(problem, type="problem") │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │ │
│ ↓ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ GoT 推理循环 │ │
│ │ │ │
│ │ while not solved and not max_iterations: │ │
│ │ │ │
│ │ ┌─────────────────────────────────────────────────┐ │ │
│ │ │ 1. 选择待处理的思维节点 │ │ │
│ │ │ node = select_node(graph) │ │ │
│ │ └─────────────────────────────────────────────────┘ │ │
│ │ │ │ │
│ │ ↓ │ │
│ │ ┌─────────────────────────────────────────────────┐ │ │
│ │ │ 2. 决定操作类型 │ │ │
│ │ │ op = decide_operation(node) │ │ │
│ │ │ // 分解/聚合/转换/循环 │ │ │
│ │ └─────────────────────────────────────────────────┘ │ │
│ │ │ │ │
│ │ ↓ │ │
│ │ ┌─────────────────────────────────────────────────┐ │ │
│ │ │ 3. 执行操作 │ │ │
│ │ │ new_nodes = execute_operation(node, op) │ │ │
│ │ └─────────────────────────────────────────────────┘ │ │
│ │ │ │ │
│ │ ↓ │ │
│ │ ┌─────────────────────────────────────────────────┐ │ │
│ │ │ 4. 更新思维图 │ │ │
│ │ │ add_nodes(graph, new_nodes) │ │ │
│ │ │ add_edges(graph, node, new_nodes) │ │ │
│ │ └─────────────────────────────────────────────────┘ │ │
│ │ │ │ │
│ │ ↓ │ │
│ │ ┌─────────────────────────────────────────────────┐ │ │
│ │ │ 5. 检查是否解决 │ │ │
│ │ │ if is_solution(new_nodes): │ │ │
│ │ │ return extract_solution() │ │ │
│ │ └─────────────────────────────────────────────────┘ │ │
│ │ │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │ │
│ ↓ │
│ ┌─────────────────────┐ │
│ │ 返回最优解 │ │
│ └─────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘2.2 思维图结构
┌─────────────────────────────────────────────────────────────┐
│ 思维图数据结构 │
├─────────────────────────────────────────────────────────────┤
│ │
│ Graph = { │
│ nodes: [ │
│ { │
│ id: "node_1", │
│ type: "problem", // 问题节点 │
│ content: "...", │
│ state: "pending", // 状态 │
│ score: 0.0 // 评分 │
│ }, │
│ { │
│ id: "node_2", │
│ type: "thought", // 思维节点 │
│ content: "...", │
│ state: "expanded", │
│ score: 4.0 │
│ }, │
│ { │
│ id: "node_3", │
│ type: "aggregation", // 聚合节点 │
│ content: "...", │
│ inputs: ["node_2", "node_4"], │
│ state: "completed", │
│ score: 4.5 │
│ } │
│ ], │
│ edges: [ │
│ {from: "node_1", to: "node_2", type: "decompose"},│
│ {from: "node_2", to: "node_3", type: "aggregate"},│
│ {from: "node_4", to: "node_3", type: "aggregate"} │
│ ] │
│ } │
│ │
└─────────────────────────────────────────────────────────────┘三、代码实现
3.1 基础 GoT 实现
"""
Graph-of-Thought 基础实现
支持分解、聚合、转换、循环四种操作
"""
from typing import List, Dict, Optional, Set
from dataclasses import dataclass, field
from enum import Enum
import uuid
from langchain_openai import ChatOpenAI
class NodeType(Enum):
"""节点类型"""
PROBLEM = "problem" # 问题节点
THOUGHT = "thought" # 思维节点
AGGREGATION = "aggregation" # 聚合节点
SOLUTION = "solution" # 解节点
class EdgeType(Enum):
"""边类型"""
DECOMPOSE = "decompose" # 分解
AGGREGATE = "aggregate" # 聚合
TRANSFORM = "transform" # 转换
ITERATE = "iterate" # 循环
@dataclass
class ThoughtNode:
"""思维节点"""
id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
type: NodeType = NodeType.THOUGHT
content: str = ""
state: str = "pending" # pending, expanded, completed
score: float = 0.0
inputs: List[str] = field(default_factory=list) # 输入节点 ID
def __hash__(self):
return hash(self.id)
@dataclass
class ThoughtEdge:
"""思维边"""
from_id: str
to_id: str
type: EdgeType
class GraphOfThought:
"""
Graph-of-Thought 推理器
支持分解、聚合、转换、循环四种操作
"""
DECOMPOSE_PROMPT = """
请将以下问题分解为若干子问题:
问题:{problem}
请输出 2-4 个子问题,每个子问题应该:
1. 是原问题的组成部分
2. 可以独立解决
3. 解决后有助于回答原问题
格式:
子问题1: ...
子问题2: ...
"""
AGGREGATE_PROMPT = """
请将以下多个思考结果整合为一个综合结论:
{thoughts}
请输出一个综合性的结论,整合所有要点。
"""
TRANSFORM_PROMPT = """
请改进以下思考内容:
当前思考:{thought}
请输出一个改进后的版本,要求:
1. 保持核心意思不变
2. 增加细节或修正错误
3. 提高清晰度和准确性
改进后的思考:
"""
def __init__(
self,
model_name: str = "gpt-4",
max_iterations: int = 10
):
"""
初始化 GoT 推理器
Args:
model_name: 模型名称
max_iterations: 最大迭代次数
"""
self.llm = ChatOpenAI(model=model_name, temperature=0.7)
self.max_iterations = max_iterations
# 图结构
self.nodes: Dict[str, ThoughtNode] = {}
self.edges: List[ThoughtEdge] = []
def add_node(self, node: ThoughtNode) -> str:
"""添加节点"""
self.nodes[node.id] = node
return node.id
def add_edge(self, from_id: str, to_id: str, edge_type: EdgeType):
"""添加边"""
self.edges.append(ThoughtEdge(from_id, to_id, edge_type))
def get_children(self, node_id: str) -> List[ThoughtNode]:
"""获取子节点"""
children = []
for edge in self.edges:
if edge.from_id == node_id:
if edge.to_id in self.nodes:
children.append(self.nodes[edge.to_id])
return children
def get_parents(self, node_id: str) -> List[ThoughtNode]:
"""获取父节点"""
parents = []
for edge in self.edges:
if edge.to_id == node_id:
if edge.from_id in self.nodes:
parents.append(self.nodes[edge.from_id])
return parents
def decompose(self, node: ThoughtNode) -> List[ThoughtNode]:
"""分解操作"""
prompt = self.DECOMPOSE_PROMPT.format(problem=node.content)
response = self.llm.invoke(prompt)
# 解析子问题
sub_thoughts = []
for line in response.content.split('\n'):
if line.strip().startswith('子问题'):
parts = line.split(':', 1)
if len(parts) > 1:
content = parts[1].strip()
sub_node = ThoughtNode(
type=NodeType.THOUGHT,
content=content
)
sub_thoughts.append(sub_node)
self.add_node(sub_node)
self.add_edge(node.id, sub_node.id, EdgeType.DECOMPOSE)
return sub_thoughts
def aggregate(self, nodes: List[ThoughtNode]) -> ThoughtNode:
"""聚合操作"""
thoughts_text = "\n\n".join([
f"思考{i+1}:{n.content}"
for i, n in enumerate(nodes)
])
prompt = self.AGGREGATE_PROMPT.format(thoughts=thoughts_text)
response = self.llm.invoke(prompt)
# 创建聚合节点
agg_node = ThoughtNode(
type=NodeType.AGGREGATION,
content=response.content,
inputs=[n.id for n in nodes]
)
self.add_node(agg_node)
# 添加聚合边
for n in nodes:
self.add_edge(n.id, agg_node.id, EdgeType.AGGREGATE)
return agg_node
def transform(self, node: ThoughtNode) -> ThoughtNode:
"""转换操作"""
prompt = self.TRANSFORM_PROMPT.format(thought=node.content)
response = self.llm.invoke(prompt)
# 创建转换后的节点
new_node = ThoughtNode(
type=NodeType.THOUGHT,
content=response.content
)
self.add_node(new_node)
self.add_edge(node.id, new_node.id, EdgeType.TRANSFORM)
return new_node
def solve(self, problem: str) -> dict:
"""
使用 GoT 解决问题
Args:
problem: 问题文本
Returns:
包含解题过程的字典
"""
# 创建根节点
root = ThoughtNode(
type=NodeType.PROBLEM,
content=problem
)
self.add_node(root)
# 简化实现:分解 -> 处理子问题 -> 聚合
# 第一步:分解问题
sub_thoughts = self.decompose(root)
# 第二步:处理每个子问题(这里简化为转换)
processed = []
for thought in sub_thoughts:
# 可以继续分解或转换
if len(thought.content) > 50: # 较长的继续分解
sub_subs = self.decompose(thought)
if sub_subs:
agg = self.aggregate(sub_subs)
processed.append(agg)
else:
processed.append(thought)
else:
processed.append(thought)
# 第三步:聚合结果
if len(processed) > 1:
final = self.aggregate(processed)
else:
final = processed[0] if processed else root
# 标记为解
final.type = NodeType.SOLUTION
return {
"problem": problem,
"solution": final.content,
"graph": self._graph_to_dict()
}
def _graph_to_dict(self) -> dict:
"""将图转换为字典格式"""
return {
"nodes": [
{
"id": n.id,
"type": n.type.value,
"content": n.content[:100] + "..." if len(n.content) > 100 else n.content,
"score": n.score
}
for n in self.nodes.values()
],
"edges": [
{
"from": e.from_id,
"to": e.to_id,
"type": e.type.value
}
for e in self.edges
]
}
# 使用示例
if __name__ == "__main__":
got = GraphOfThought()
result = got.solve(
"分析影响电动汽车普及的主要因素,并提出推广建议"
)
print(f"问题:{result['problem']}")
print(f"\n解决方案:\n{result['solution']}")
print("\n思维图结构:")
print(f"节点数:{len(result['graph']['nodes'])}")
print(f"边数:{len(result['graph']['edges'])}")3.2 使用 LangGraph 实现
"""
使用 LangGraph 实现 GoT
更适合复杂的状态管理
"""
from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
# 定义状态
class GoTState(TypedDict):
"""GoT 状态"""
problem: str # 原始问题
current_thoughts: List[str] # 当前思维列表
aggregated_result: str # 聚合结果
iteration: int # 迭代次数
is_solved: bool # 是否已解决
def create_got_graph():
"""创建 GoT 图"""
llm = ChatOpenAI(model="gpt-4", temperature=0.7)
# 定义节点函数
def decompose(state: GoTState) -> GoTState:
"""分解节点"""
prompt = f"""
将以下问题分解为 3 个子问题:
{state['problem']}
格式:
1. ...
2. ...
3. ...
"""
response = llm.invoke(prompt)
# 解析子问题
thoughts = []
for line in response.content.split('\n'):
line = line.strip()
if line and line[0].isdigit() and '.' in line:
thought = line.split('.', 1)[1].strip()
thoughts.append(thought)
return {
**state,
"current_thoughts": thoughts,
"iteration": state["iteration"] + 1
}
def process_thought(state: GoTState) -> GoTState:
"""处理每个子问题"""
processed = []
for thought in state["current_thoughts"]:
prompt = f"""
请分析以下子问题并给出结论:
{thought}
"""
response = llm.invoke(prompt)
processed.append(response.content)
return {
**state,
"current_thoughts": processed
}
def aggregate(state: GoTState) -> GoTState:
"""聚合节点"""
thoughts_text = "\n\n".join([
f"分析{i+1}:{t}"
for i, t in enumerate(state["current_thoughts"])
])
prompt = f"""
请将以下分析结果整合为综合结论:
{thoughts_text}
请给出一个完整的解决方案。
"""
response = llm.invoke(prompt)
return {
**state,
"aggregated_result": response.content,
"is_solved": True
}
def should_continue(state: GoTState) -> str:
"""判断是否继续"""
if state["is_solved"]:
return "end"
if state["iteration"] >= 3:
return "end"
return "continue"
# 创建图
workflow = StateGraph(GoTState)
# 添加节点
workflow.add_node("decompose", decompose)
workflow.add_node("process", process_thought)
workflow.add_node("aggregate", aggregate)
# 设置入口
workflow.set_entry_point("decompose")
# 添加边
workflow.add_edge("decompose", "process")
workflow.add_edge("process", "aggregate")
workflow.add_edge("aggregate", END)
return workflow.compile()
# 使用示例
if __name__ == "__main__":
graph = create_got_graph()
result = graph.invoke({
"problem": "如何提高团队的工作效率?",
"current_thoughts": [],
"aggregated_result": "",
"iteration": 0,
"is_solved": False
})
print(f"问题:{result['problem']}")
print(f"\n解决方案:\n{result['aggregated_result']}")四、适用场景
4.1 最佳适用场景
| 场景类型 | 具体示例 | GoT 优势 |
|---|---|---|
| 复杂问题分解 | 系统架构设计、战略规划 | 支持多级分解和聚合 |
| 多维度分析 | 市场调研、竞品分析 | 并行分析后聚合 |
| 迭代优化 | 方案改进、文档润色 | 支持循环操作 |
| 综合推理 | 复杂决策、综合报告 | 整合多个思考结果 |
4.2 场景详解
┌─────────────────────────────────────────────────────────────┐
│ GoT 典型应用场景 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 1. 多维度分析场景 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 问题:评估一个创业项目的可行性 │ │
│ │ │ │
│ │ 项目可行性 │ │
│ │ │ │ │
│ │ ┌──────────────┼──────────────┐ │ │
│ │ ↓ ↓ ↓ │ │
│ │ 市场分析 技术可行性 财务分析 │ │
│ │ │ │ │ │ │
│ │ ↓ ↓ ↓ │ │
│ │ 市场规模 技术难度 成本预算 │ │
│ │ 竞争格局 团队能力 收益预测 │ │
│ │ │ │ │ │ │
│ │ └──────────────┼──────────────┘ │ │
│ │ ↓ │ │
│ │ 综合评估报告 │ │
│ │ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 2. 迭代优化场景 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 任务:优化一份产品文案 │ │
│ │ │ │
│ │ 初稿 ─→ 评估 ─→ 修改建议 ─→ 修改稿 │ │
│ │ ↑ │ │ │
│ │ └───────────────────────┘ │ │
│ │ 循环迭代 │ │
│ │ │ │
│ │ 优点:可以持续改进直到满意 │ │
│ │ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘4.3 与 CoT/ToT 的适用场景对比
| 场景 | CoT | ToT | GoT | 推荐 |
|---|---|---|---|---|
| 简单计算 | ✅ | ❌ | ❌ | CoT |
| 多步推理 | ✅ | ⚠️ | ❌ | CoT |
| 多路径探索 | ❌ | ✅ | ⚠️ | ToT |
| 多维度分析 | ❌ | ⚠️ | ✅ | GoT |
| 迭代优化 | ❌ | ❌ | ✅ | GoT |
| 复杂依赖 | ❌ | ❌ | ✅ | GoT |
五、GoT 的优势与挑战
5.1 主要优势
| 优势 | 描述 |
|---|---|
| 灵活性 | 支持分支、聚合、循环多种推理模式 |
| 表达能力 | 可以建模复杂的思维关系 |
| 可扩展 | 易于添加新的操作类型 |
| 并行处理 | 子问题可以并行处理 |
5.2 主要挑战
| 挑战 | 描述 | 应对策略 |
|---|---|---|
| 复杂度高 | 图结构比树更难管理 | 使用图数据库或专用库 |
| 开销大 | 节点和边数量增长快 | 剪枝和缓存策略 |
| 调试困难 | 图的执行路径不直观 | 可视化工具和日志 |
| 评估困难 | 中间节点评估更复杂 | 设计合适的评估函数 |
六、面试高频问题
Q1: GoT 与 ToT 的核心区别是什么?
答案要点:
- 结构:ToT 是树,GoT 是图
- 操作:ToT 只有分支,GoT 支持聚合、转换、循环
- 能力:GoT 可以处理更复杂的依赖关系
- 复杂度:GoT 实现和管理更复杂
Q2: GoT 的四种核心操作是什么?
答案要点:
- 分解(Decomposition):将问题拆分为子问题
- 聚合(Aggregation):将多个结果合并
- 转换(Transformation):改进或细化思维
- 循环(Iteration):迭代推理
Q3: 什么时候应该使用 GoT 而不是 CoT/ToT?
答案要点:
- 需要将多个分析结果聚合时
- 需要迭代优化直到满意时
- 问题有复杂的依赖关系时
- 需要多维度并行分析时
Q4: GoT 的主要挑战是什么?
答案要点:
- 图结构复杂,管理难度大
- 节点和边数量增长快,开销大
- 执行路径不直观,调试困难
- 中间节点评估更复杂
Q5: 如何优化 GoT 的性能?
答案要点:
- 剪枝:移除低价值节点
- 缓存:避免重复计算
- 并行:子问题并行处理
- 增量:只计算新增节点
七、小结
| 概念 | 一句话总结 | 面试关键词 |
|---|---|---|
| GoT | 图状推理,支持聚合、分解、循环 | 思维图谱、有向图 |
| 分解 | 将问题拆分为子问题 | Decomposition |
| 聚合 | 将多个结果合并 | Aggregation |
| 循环 | 迭代推理优化 | Iteration |
一句话总结:GoT 是 CoT/ToT 的终极形态,通过图结构支持分解、聚合、转换、循环四种操作,适合复杂的推理场景。
最后更新:2026年3月19日