Graph of Thoughts (GoT) 框架：超越思维链与思维树的复杂推理引擎

1. 从链式到图式为什么我们需要超越CoT与ToT如果你已经尝试过用大语言模型LLM解决一些稍微复杂的问题比如逻辑推理、代码生成或者数学计算那你大概率接触过“思维链”Chain-of-Thought, CoT和“思维树”Tree-of-Thoughts, ToT这些提示工程技术。CoT让模型一步步推导ToT则让模型在多个可能的推理路径上探索这已经比直接提问强太多了。但在我实际用它们处理一些工业级复杂度的任务时比如分析一份冗长的技术报告并提取结构化结论或者优化一个存在多个约束条件的算法时我发现了它们的瓶颈推理结构太“死板”了。CoT是一条单行道错了就得从头再来ToT像一棵分叉的树虽然能探索但分支之间是孤立的一个分支上产生的精彩中间结论无法被另一个分支直接利用。这就像一群专家在各自的小房间里埋头苦干无法交流。而现实中解决复杂问题往往需要这种动态的、非线性的思维协作与信息整合。这就是Graph of ThoughtsGoT框架吸引我的地方。它不再局限于链或树而是将整个推理过程建模为一个有向图。图中的节点是“思考”Thoughts也就是LLM产生的各种文本状态如问题分解、部分答案、评估结果边是“操作”Operations定义了思考之间如何转换、合并、评估或迭代。你可以把这个图想象成一个灵活的工作流引擎LLM是执行每个步骤的“工人”而GoT框架则是那个统筹全局、动态调度任务的“项目经理”。我花了几周时间深入研究了spcl/graph-of-thoughts这个官方实现。它不是一个简单的概念演示而是一个设计精巧、可用于实际生产的Python框架。最让我欣赏的是它的双重灵活性一方面你可以用它实现论文中那种复杂的、带合并与回溯的图推理另一方面它的架构也完全兼容和简化了CoT、ToT等传统模式的实现让你可以在同一个框架下对比不同策略。接下来我将结合源码和实战拆解如何利用GoT框架将LLM从“高级打字机”变成真正的“复杂问题求解引擎”。2. 核心架构拆解控制器、操作图与语言模型如何协同要玩转GoT首先得理解它的三个核心组件这就像理解一台精密仪器的传动系统。很多教程只教你怎么调包但如果不明白背后原理一旦出问题或者需要自定义你就会寸步难行。2.1 控制器Controller推理过程的总指挥控制器是框架的大脑位于graph_of_thoughts/controller/。它的核心职责是驱动整个推理图的执行。你可能会想执行一个图不就是按某种顺序遍历节点吗但实际复杂得多因为LLM调用是异步且昂贵的而且后续节点的执行可能依赖于前面多个节点的结果。在GoT的控制器实现中我发现了几个关键设计状态管理控制器维护一个全局的“思考状态”字典。每个思考Thought都是一个对象包含了其ID、内容、分数以及与其他思考的关联关系。控制器负责在操作执行过程中更新和传递这些状态。操作调度控制器并非简单地遍历图。它检查图中每个操作的“前置条件”。例如一个Aggregate聚合操作可能需要等待其所有前置的思考节点都执行完毕并有了结果后才能触发。控制器实现了这种依赖感知的调度逻辑。与LLM交互控制器并不直接与LLM对话而是通过LanguageModel抽象接口。它将需要处理的思考状态和当前操作上下文交给Prompter模块去构建具体的提示词Prompt然后调用LLM得到响应后再通过Parser模块解析更新思考状态。实操心得阅读controller.py中的run()方法是理解整个流程的最佳入口。你会发现它内部有一个主循环不断检查图中是否有“可执行”的操作即所有前置依赖已满足然后执行它们直到没有更多操作可执行为止。这种设计使得框架可以处理带循环和条件分支的复杂图结构。2.2 操作图Graph of Operations, GoO定义你的推理蓝图这是GoT框架的灵魂代码在graph_of_thoughts/operations/。一个GoO由一系列Operation操作节点和它们之间的边组成。框架内置了几种基础操作足以构建绝大多数推理模式Generate生成最基础的操作。给定一些输入思考调用LLM生成新的思考。在CoT中你可能会串联多个Generate操作来实现一步步推导。Score评分对一个或多个思考进行评估并为其打分。分数可以用于后续的路径选择如ToT中的剪枝或聚合时的权重计算。你需要自己实现评分函数。Aggregate聚合这是GoT超越链和树的关键。它可以将多个思考例如从不同角度分析问题的结果合并成一个新的、更全面的思考。合并的逻辑如直接拼接、基于分数的加权汇总可以由你自定义。GroundTruth验证这是一个特殊的操作通常用于实验或测试。它将LLM产生的思考与已知的正确答案进行比较并给出一个“终极分数”。在实际生产环境中你可能用不到它或者将其替换为你自己的验证逻辑。KeepBestN保留最优N个常用于类似ToT的搜索场景。在所有当前思考中根据分数只保留最好的N个其余剪枝掉以控制计算成本。构建图的方式非常直观通过append_operation方法添加节点框架会自动按添加顺序建立边形成链。对于更复杂的图结构如合并、分支你需要使用add_operation并手动指定前置操作。注意事项操作图的执行顺序并不完全等同于你添加操作的顺序。控制器会根据操作的依赖关系动态调度。例如你定义了一个Generate后面跟着一个Score那么Score会等待其前置的Generate操作执行完毕并产生思考后才会被触发。理解这种隐式的依赖关系对于调试复杂图至关重要。2.3 语言模型、提示器与解析器与LLM对话的桥梁这三者共同封装了与LLM交互的细节让控制器和操作图无需关心具体的模型API。Language Models(language_models.py)提供了对不同LLM后端如OpenAI ChatGPT、Azure OpenAI、甚至是本地LLM的抽象。目前官方主要支持OpenAI系列。配置的关键在于一个config.json文件里面需要包含你的API密钥和基础URL如果使用Azure或代理。{ openai_api_key: your-api-key-here, openai_api_base: https://api.openai.com/v1 // 可选的用于自定义端点 }Prompter这是一个需要你为每个特定任务实现的类。它的generate_prompt方法接收当前的思考状态和操作上下文然后返回一个发送给LLM的字符串提示词。这是提示工程的核心所在。设计一个好的Prompter决定了LLM是否能理解在当前步骤该做什么。Parser同样需要针对任务实现。它的parse_thoughts方法接收LLM的返回文本和当前状态负责从这段文本中提取出结构化的“思考”内容并可能为其分配一个初始分数。一个健壮的Parser必须能处理LLM输出的各种可能格式和意外情况。避坑技巧在实现自定义的Prompter和Parser时一定要加入充分的错误处理和日志。LLM的输出是不可控的可能包含无关内容、格式错误或直接拒绝。你的Parser不能假设输出是完美的需要用正则表达式、关键字匹配或尝试-捕获try-catch逻辑来优雅地处理异常并设置一个默认的失败状态防止整个推理图崩溃。3. 实战演练用GoT框架实现一个复杂文档分析器看懂了架构我们来动手实现一个比官方排序例子更贴近实际应用的场景分析一篇技术博客并生成一份包含要点总结、技术栈提取和难度评估的结构化报告。我们将设计一个比简单链式更复杂的图结构。3.1 问题定义与图结构设计假设我们的输入是一篇关于“微服务架构设计”的长博客。我们希望LLM帮我们完成分解将文章分解成几个核心部分如概述、挑战、解决方案、案例。并行分析对每个核心部分同时进行两项分析a) 提取关键论点b) 识别提到的技术工具。聚合与评估将所有部分的“关键论点”聚合形成全文摘要将所有“技术工具”聚合去重后形成技术栈列表。最后基于全文内容评估文章的阅读难度。这个任务天然适合用图来建模一个Generate操作将文章分解成部分。对每个部分触发两个并行的Generate操作提取论点、识别工具。两个Aggregate操作分别汇总所有论点和所有工具。一个最终的Generate操作基于聚合后的内容评估难度。3.2 实现自定义Prompter与Parser首先我们定义思考状态。我们可以用一个字典来跟踪不同阶段的内容initial_state { original_text: 这里放入整篇博客内容..., parts: [], # 存储分解后的部分 key_points_per_part: [], # 列表的列表存储每部分的论点 tech_per_part: [], # 列表的列表存储每部分的技术 summary: , tech_stack: [], difficulty: }然后我们实现BlogAnalyzerPrompter。这里以“分解”操作为例class BlogAnalyzerPrompter: def generate_prompt(self, current_state, **kwargs): operation kwargs.get(operation) if operation decompose: # 提示词让模型将文章分解成逻辑部分 return f你是一位技术文档分析师。请将以下关于微服务的文章分解为3-5个核心逻辑部分例如引言、核心挑战、设计模式、实施建议、总结。为每个部分提供一个简短的小标题。 文章内容 {current_state[original_text]} 请严格按照以下格式输出 部分1: [小标题] 部分2: [小标题] ... elif operation extract_points: part_title kwargs.get(part_title) part_content kwargs.get(part_content) # 提示词针对特定部分提取关键论点 return f针对文章中的“{part_title}”部分提取3-5个最核心的论点或主张。 部分内容 {part_content} 请以列表形式输出关键论点 # ... 为其他操作extract_tech, aggregate_summary, evaluate_difficulty实现类似的提示词生成逻辑接着实现BlogAnalyzerParserclass BlogAnalyzerParser: def parse_thoughts(self, response_text, current_state, **kwargs): operation kwargs.get(operation) new_thoughts [] if operation decompose: # 解析模型返回的部分标题 lines response_text.strip().split(\n) parts [line.split(: )[1] for line in lines if line.startswith(部分)] # 更新状态并创建新的“思考”对象 current_state[parts] parts # 这里简化处理实际框架中需要创建Thought对象并返回 thought Thought(contentjson.dumps({parts: parts}), score0.0) new_thoughts.append(thought) elif operation extract_points: # 解析关键论点列表 points [line.strip(- ).strip() for line in response_text.strip().split(\n) if line.strip()] # 更新状态... # ... 解析其他操作的输出 return new_thoughts3.3 构建并执行操作图现在我们用代码把图搭建起来。这里会涉及到更复杂的图结构我们需要使用add_operation并指定前置依赖。from graph_of_thoughts import controller, language_models, operations # 1. 初始化组件 lm language_models.ChatGPT(config.json, model_namegpt-4) # 建议使用能力更强的模型 prompter BlogAnalyzerPrompter() parser BlogAnalyzerParser() # 2. 构建操作图 gop operations.GraphOfOperations() # 操作1: 分解文章 decompose_op operations.Generate() gop.add_operation(decompose_op) # 假设我们根据分解结果知道有3个部分为每个部分创建并行分析分支 # 在实际中这需要动态生成这里为演示简化 extract_ops [] for i in range(3): # 假设3个部分 # 操作2a: 为第i部分提取关键论点 op_points operations.Generate() # 设置它的前置操作是 decompose_op并且通过上下文传递部分是哪个 gop.add_operation(op_points, [decompose_op]) extract_ops.append((op_points, fpoints_for_part_{i})) # 操作2b: 为第i部分提取技术栈 op_tech operations.Generate() gop.add_operation(op_tech, [decompose_op]) extract_ops.append((op_tech, ftech_for_part_{i})) # 操作3: 聚合所有关键论点 aggregate_points_op operations.Aggregate(aggregation_functionmy_points_agg_func) # 它的前置是所有提取论点的操作 gop.add_operation(aggregate_points_op, [op for op, tag in extract_ops if points in tag]) # 操作4: 聚合所有技术栈 aggregate_tech_op operations.Aggregate(aggregation_functionmy_tech_agg_func) gop.add_operation(aggregate_tech_op, [op for op, tag in extract_ops if tech in tag]) # 操作5: 基于聚合结果评估难度 (依赖于操作3和4) evaluate_op operations.Generate() gop.add_operation(evaluate_op, [aggregate_points_op, aggregate_tech_op]) # 3. 创建控制器并运行 ctrl controller.Controller( lm, gop, prompter, parser, initial_state # 包含原始文本的初始状态字典 ) ctrl.run() ctrl.output_graph(blog_analysis_graph.json)这个图结构比简单的链或树复杂得多它清晰地表达了任务的并行性与信息聚合需求。控制器会负责调度先执行分解然后并行执行所有部分的提取操作等所有提取完成再触发两个聚合操作最后进行难度评估。3.4 关键函数实现聚合与评分上面的代码中提到了自定义的聚合函数my_points_agg_func和my_tech_agg_func。这是GoT框架强大灵活性的体现。例如技术栈聚合函数可能需要去重和分类def my_tech_agg_func(thoughts): thoughts: 一个Thought对象列表每个Thought的content字段是之前提取的技术列表的JSON字符串。 all_tech [] for t in thoughts: tech_list json.loads(t.content) all_tech.extend(tech_list) # 假设content是[Docker, K8s]这样的列表 # 去重并可以做一些简单的分类或排序 unique_tech list(set(all_tech)) # 也许可以按类别排序这里简单按字母排序 unique_tech.sort() # 返回一个新的Thought内容 return json.dumps({aggregated_tech_stack: unique_tech})评分函数也同样重要例如在ToT模式中你可以用评分来筛选最佳推理路径。评分函数接收一个思考内容返回一个分数通常是浮点数。def score_solution_completeness(thought_content): 评估一个解决方案思考的完整性。 thought_content: 思考的文本内容。 返回一个0-1之间的分数。 # 简单的启发式评分检查关键要素是否出现 content_lower thought_content.lower() score 0.0 if database in content_lower: score 0.2 if api in content_lower: score 0.2 if scalability in content_lower: score 0.3 if security in content_lower: score 0.3 # 确保分数不超过1 return min(score, 1.0)4. 性能调优与生产级部署的考量将GoT用于实验和用于生产环境关注点截然不同。以下是我在尝试将其集成到实际项目时总结的经验和坑点。4.1 成本与延迟管理GoT框架会多次调用LLM成本Token消耗和总延迟是首要问题。操作合并与剪枝在设计图时问自己每个Generate操作都是必要的吗有些信息能否通过一次提问获取在Aggregate之前可以用KeepBestN或基于Score的过滤操作提前淘汰低质量的思考分支避免为无效路径支付后续的LLM调用成本。模型选型策略并非所有步骤都需要最强大的模型如GPT-4。你可以设计一个混合模型策略。例如让成本低的模型如GPT-3.5-Turbo执行简单的文本分解、提取任务让高成本、高能力的模型如GPT-4执行需要深度推理、聚合和评估的最终步骤。GoT框架的LanguageModel抽象层可以支持这一点你需要在不同操作中配置使用不同的模型实例。异步与批处理框架的控制器执行在默认情况下可能是同步的。对于可以并行执行的操作如我们例子中分析文章不同部分研究源码后发现其调度是顺序的但LLM调用本身可以通过异步库如asyncio来优化。你可以考虑修改控制器或操作执行逻辑将独立的Generate操作批量发送给LLM API如果API支持批处理以显著减少总等待时间。4.2 稳定性与错误处理LLM的输出具有不确定性一个节点的失败不应导致整个图崩溃。Parser的鲁棒性如前所述这是第一道防线。除了格式解析还要考虑内容合理性检查。例如如果解析出的“技术栈”里出现了明显无关的词语如“爱情”、“哲学”应该将其过滤或标记为低置信度。操作的重试与回退框架本身没有内置重试机制。在生产环境中你需要在Controller调用LLM的地方包裹重试逻辑例如使用tenacity库针对网络超时、API限速等问题进行自动重试。对于持续的内容错误如LLM始终不按格式输出可能需要设计一个回退操作例如使用更简单、约束更强的提示词再试一次或者记录错误并跳过当前分支。状态检查点与恢复对于执行时间很长的复杂图可以考虑定期将控制器和操作图的状态包括所有思考的内容和分数序列化保存。如果进程意外中断可以从检查点恢复避免从头开始浪费已消耗的API费用。4.3 监控、评估与持续改进可视化与调试框架提供的output_graph(“output.json”)功能非常有用。这个JSON文件包含了整个推理过程的完整轨迹。你可以编写脚本将其可视化查看每个思考的内容、分数以及操作之间的流向。这对于调试复杂的图逻辑和优化提示词至关重要。评估流水线如何衡量你的GoT工作流是否比简单的CoT或直接提问更好你需要建立一套评估标准。对于摘要任务可以是ROUGE分数对于分类任务可以是准确率。在GroundTruth操作中不仅可以对比最终答案还可以记录中间步骤的准确性帮助你定位是哪个环节的提示词或图设计出了问题。提示词的版本化管理你的Prompter类中的提示词模板是核心资产。应该将它们抽取到配置文件如YAML或数据库中进行版本控制。这样便于A/B测试不同的提示词变体对最终效果的影响。5. 常见问题排查与进阶技巧在实际使用中你肯定会遇到各种问题。下面是我遇到的一些典型情况及其解决方法。5.1 图执行卡住或陷入循环症状程序长时间不结束或者某个操作反复执行。排查检查依赖环确保你的操作图没有形成循环依赖。例如操作A依赖操作B的结果操作B又依赖操作A的结果。控制器无法解决这种死锁。检查操作完成条件每个操作执行后都应该产生新的思考或更新状态。检查你的Parser是否正确地创建并返回了Thought对象。如果某个操作总是解析失败返回空列表可能导致后续依赖它的操作永远无法满足“前置条件”。启用详细日志在框架代码中关键位置如控制器调度循环、操作执行前后添加日志打印当前状态、正在执行的操作ID等这是最直接的调试手段。5.2 LLM输出格式不稳定导致Parser失败症状Parser频繁抛出异常或者解析出的内容牛头不对马嘴。解决强化提示词约束在Prompter中使用更严格的格式指令例如“请务必以’关键词‘开头后面跟逗号分隔的列表”甚至提供输出范例Few-shot。采用多轮解析在Parser中实现一个“宽容”模式。首先尝试用严格的规则如正则表达式解析。如果失败则尝试用更宽松的方法如寻找关键词、使用LLM自己来解析上一轮LLM的输出——虽然有点套娃但有时有效。最后再设置一个默认值。后处理清洗解析后对内容进行清洗。去除多余的标记、纠正明显的拼写错误、过滤掉长度异常或包含禁用词的结果。5.3 效果不如简单的CoT症状设计了复杂的图但最终结果的质量或准确性还不如一个精心设计的单链CoT提示。分析与优化复杂度与收益权衡不是所有问题都需要图。对于逻辑线性强、步骤明确的问题一个健壮的CoT可能更高效可靠。GoT的优势在于处理需要信息整合、多路径探索与回溯、多角度分析的问题。重新评估你的任务是否真正需要这些特性。评估中间步骤用output_graph导出执行过程仔细检查每个中间“思考”的质量。问题可能出在某个具体的操作上。例如负责“分解”的Generate操作分出的部分不合理导致后续分析全部跑偏。优化那个薄弱环节的提示词。分数函数的误导性如果你使用了Score和KeepBestN来进行搜索剪枝确保你的评分函数与最终目标强相关。一个与最终目标偏离的评分函数会引导搜索走向错误的方向。5.4 扩展框架实现新的操作类型框架内置的操作是基础但你可能需要更特殊的操作。例如一个WebSearch操作调用搜索API获取外部信息或者一个CodeExecution操作在安全沙箱中运行模型生成的代码并获取结果。扩展起来很直观在operations/目录下新建一个Python文件。创建一个继承自Operation基类的新类。实现其execute方法。在这个方法里你可以访问控制器、语言模型、当前状态执行任何你需要的逻辑调用外部API、运行代码等然后返回新的思考列表。将这个新操作像内置操作一样添加到你的图中。这种可扩展性使得GoT框架能够成为连接LLM与外部工具和知识的强大枢纽真正走向实现智能体Agent的复杂工作流。经过这一番从理论到实战的深度探索GoT框架给我的感觉不再只是一个学术概念的实现而是一个具有坚实工程基础的问题解决工具箱。它的价值在于提供了一种范式一种将复杂问题拆解、编排并利用LLM逐步攻克的系统性方法。当你下次面对一个让单次提示感到棘手的复杂任务时不妨先拿起纸笔画一画这个任务的推理步骤是更像一条线、一棵树还是一张网如果是一张网那么Graph of Thoughts或许就是你一直在寻找的那把钥匙。