基于LangGraph的自我改进智能体：Reflection与Reflexion技术详解与实现

作者：大模型RAG和Agent技术实践
文章目录

目录引言：什么是自我改进智能体？一. Reflection智能体：自我批评的艺术

核心概念工作原理Reflection模式工作流程关键技术特征LangGraph实现架构代码实现核心优势与特点

二. Reflexion智能体：知识检索的智慧

核心概念工作原理Reflexion模式工作流程关键技术特征LangGraph实现架构代码实现核心优势与特点

两种智能体模式对比

架构对比性能对比表选择策略

三. 系统整体架构四. 关键技术实现

1. 真正的流式输出架构2. 配置参数动态化3. LangGraph工作流状态管理4. Ollama本地模型集成

性能优化策略

1. 流式输出性能优化2. 模型调用优化

技术挑战与解决方案

挑战1：真正的流式输出实现挑战2：LangGraph状态管理挑战3：Ollama集成稳定性

应用输出示例

示例1：写作任务示例2：技术问题解答

总结

目录

引言：什么是自我改进智能体？

在人工智能快速发展的今天，智能体系统已经从简单的指令响应演进为具备自我改进和学习能力的复杂系统。本文将深入探讨两种核心的自我改进智能体模式：Reflection（反思型）和Reflexion（递进式），并详细介绍基于LangGraph框架的工程实现，包括系统架构设计、关键技术实现和性能优化策略。
传统的AI系统通常是一次性生成答案的模式：“输入 → 处理 → 输出”。而自我改进智能体采用循环优化的思路：“生成 → 评估 → 改进 → 再评估”，直到达到满意的结果。
这种设计理念来源于人类的学习过程：

我们不是第一次就写出完美的文章我们会先写个草稿，然后回顾、修改、再完善有时会查阅资料、寻求建议，然后重新组织思路

自我改进智能体正是将这种人类思维过程抽象为算法，让AI系统具备了持续改进的能力。
首先我们看下系统实现的界面，对这两个模式的Agent，有个感性认识：

一. Reflection智能体：自我批评的艺术

核心概念

Reflection智能体模拟了人类的自我批评过程。其核心思想是：让AI生成内容，然后让AI自己批评和改进这些内容。
工作原理

Reflection模式工作流程

关键技术特征

双重身份模式

生成者：负责创建初始内容批评者：负责评估和改进内容

迭代改进循环

每次迭代都会产生批评意见基于批评进行针对性改进持续优化直到收敛

质量评估机制

设置收敛标准（如迭代次数、质量分数）智能体自主判断是否需要继续改进

LangGraph实现架构

代码实现核心

从项目源码中可以看到，Reflection智能体的核心实现包含：

1. @dataclassclassReflectionState:"""Reflection智能体状态"""
2.         messages: List[BaseMessage]# 消息历史
3.         original_input:str# 原始输入
4.         iteration_count:int# 迭代次数
5.         final_output: Optional[str]# 最终输出
6.         critique_history: List[str]# 批评历史
7.         improvement_history: List[str]# 改进历史
8.         is_converged:bool# 是否收敛
9.         error_message: Optional[str]# 错误信息

复制代码

关键工作流程节点：
1. 生成节点（generate_content）

1. asyncdef_generate_content_node(self, state: ReflectionState)-> ReflectionState:"""内容生成节点"""
2.         messages =[
3.             SystemMessage(content=self.generation_prompt),
4.             HumanMessage(content=state.original_input)]
5.         
6.         response =await self.llm_manager.generate(
7.             LLMRequest(messages=messages,**self.config.get("generation",{})))
8.         
9.         state.messages.append(AIMessage(content=response.content))return state

复制代码

2. 批评节点（critique_content）

1. asyncdef_critique_content_node(self, state: ReflectionState)-> ReflectionState:"""自我批评节点"""
2.         critique_prompt = self._build_critique_prompt(state)
3.         
4.         response =await self.llm_manager.generate(
5.             LLMRequest(messages=[HumanMessage(content=critique_prompt)]))
6.         
7.         state.critique_history.append(response.content)return state

复制代码

3. 改进节点（improve_content）

1. asyncdef_improve_content_node(self, state: ReflectionState)-> ReflectionState:"""内容改进节点"""
2.         improvement_prompt = self._build_improvement_prompt(state)
3.         
4.         response =await self.llm_manager.generate(
5.             LLMRequest(messages=[HumanMessage(content=improvement_prompt)]))
6.         
7.         state.improvement_history.append(response.content)
8.         state.messages.append(AIMessage(content=response.content))
9.         state.iteration_count +=1return state

复制代码

优势与特点

优势：

简单直接，无需外部数据源迭代速度较快，资源消耗相对较低适合内容生成和优化任务

适用场景：

文章写作和编辑代码生成和优化创意内容产出简单的问题解答

二. Reflexion智能体：知识检索的智慧

核心概念

Reflexion智能体在Reflection基础上增加了外部知识检索能力。其核心思想是：不仅自我批评，还要主动寻求外部知识和专业建议。
工作原理

Reflexion模式工作流程

关键技术特征

三元角色模式

生成者：创建初始响应自我反思者：识别知识缺口外部知识整合者：搜集和整合外部信息

知识检索循环

自动识别需要补充的知识点主动搜索相关资料将外部知识整合到响应中

智能查询生成

分析当前响应的不足生成精准的搜索查询评估知识源的可靠性

LangGraph实现架构

Reflexion智能体的LangGraph实现更加复杂，包含知识检索子图：

代码实现核心

Reflexion智能体的状态管理更加复杂：

1. @dataclassclassReflexionState:"""Reflexion智能体状态"""
2.         messages: List[BaseMessage]# 消息历史
3.         original_input:str# 原始输入
4.         current_iteration:int# 当前迭代次数
5.         initial_response: Optional[str]# 初始响应
6.         self_reflection: Optional[str]# 自省内容
7.         search_queries: List[str]# 搜索查询
8.         external_knowledge: List[Dict]# 外部知识
9.         final_response: Optional[str]# 最终响应
10.         knowledge_sources: List[str]# 知识来源
11.         is_converged:bool# 是否收敛
12.         error_message: Optional[str]# 错误信息

复制代码

核心工作节点包括：
1. 自省分析节点（self_reflect）

1. asyncdef_self_reflect_node(self, state: ReflexionState)-> ReflexionState:"""自省分析节点"""
2.         reflection_prompt =f"""
3.         请分析以下响应的不足之处：
4.         
5.         用户输入：{state.original_input}
6.         当前响应：{state.initial_response}
7.         
8.         请识别：
9.         1. 响应中的知识缺口
10.         2. 需要进一步查证的细节
11.         3. 可以改进的具体方面
12.         """
13.         
14.         response =await self.llm_manager.generate(
15.             LLMRequest(messages=[HumanMessage(content=reflection_prompt)]))
16.         
17.         state.self_reflection = response.content
18.         return state

复制代码

2. 知识检索节点（search_knowledge）

1. asyncdef_search_knowledge_node(self, state: ReflexionState)-> ReflexionState:"""知识检索节点"""ifnot state.search_queries:return state
2.         
3.         # 使用知识检索器搜索外部信息
4.         knowledge_results =await self.knowledge_retriever.search_external_knowledge(
5.             state.search_queries
6.         )
7.         
8.         state.external_knowledge.extend(knowledge_results)return state

复制代码

3. 知识整合节点（integrate_knowledge）

1. asyncdef_integrate_knowledge_node(self, state: ReflexionState)-> ReflexionState:"""知识整合节点"""
2.         integration_prompt = self._build_integration_prompt(state)
3.         
4.         response =await self.llm_manager.generate(
5.             LLMRequest(messages=[HumanMessage(content=integration_prompt)]))
6.         
7.         state.final_response = response.content
8.         return state

复制代码

优势与特点

优势：

能够获取最新的外部知识响应更加准确和全面适合复杂和专业化的问题

适用场景：

技术问题解答学术研究和分析专业领域咨询事实性问题的回答

两种智能体模式对比

架构对比

性能对比表

特性	Reflection	Reflexion
响应速度	较快	较慢
准确性	较高	高
知识深度	基础	深度
资源消耗	较低	较高
适用范围	通用	专业
实时性	高	中等

选择策略

选择Reflection的场景：

内容创作和编辑创意性工作快速响应需求资源受限环境

选择Reflexion的场景：

专业技术问题需要最新信息的查询学术研究支持准确性要求极高的场景

三. 系统整体架构

四. 关键技术实现

1. 真正的流式输出架构

我们实现了逐token的实时流式输出，而不是传统的分段式伪流式：

核心代码实现：

1. # backend/src/agents/reflection_agent.pyasyncdefprocess_stream(self, input_text:str, domain:str="general")-> AsyncGenerator[str,None]:# 流式生成内容asyncfor chunk in self.llm_manager.generate_stream(request):
2.             generated_content += chunk
3.             yield json.dumps({"type":"generate_chunk","iteration": current_iteration,"chunk": chunk
4.             })+"\n"

复制代码

2. 配置参数动态化

系统从配置文件动态读取所有LLM参数，避免硬编码：

配置示例：

1. # configs/config.yamlllm:ollama:model:"qwen3:8b"temperature:0.7max_tokens:5000agents:reflection:max_iterations:3reflexion:max_iterations:5

复制代码

3. LangGraph工作流状态管理

使用LangGraph的StateGraph管理复杂的智能体工作流：

LangGraph工作流构建

1. defbuild_reflection_graph(self)-> StateGraph:"""构建Reflection智能体工作流图"""
2.         workflow = StateGraph(ReflectionState)# 添加节点
3.         workflow.add_node("generate", self._generate_content_node)
4.         workflow.add_node("critique", self._critique_content_node)
5.         workflow.add_node("improve", self._improve_content_node)
6.         workflow.add_node("finalize", self._finalize_node)# 定义边和条件
7.         workflow.add_edge(START,"generate")
8.         workflow.add_edge("generate","critique")
9.         
10.         workflow.add_conditional_edges("critique",
11.             self._should_continue,{"continue":"improve","finish":"finalize"})
12.         
13.         workflow.add_conditional_edges("improve",
14.             self._check_iteration_limit,{"continue":"critique","finish":"finalize"})
15.         
16.         workflow.add_edge("finalize", END)return workflow

复制代码

4. Ollama本地模型集成

完整支持Ollama本地模型，无需API密钥：

Ollama客户端实现：

1. # backend/src/utils/llm_manager.pyclassOllamaClient:def__init__(self, base_url:str, model:str,**kwargs):
2.             self.base_url = base_url
3.             self.model = model
4.             self.temperature = kwargs.get('temperature',0.7)
5.             self.max_tokens = kwargs.get('max_tokens',4000)asyncdefgenerate_stream(self, prompt:str)-> AsyncGenerator[str,None]:# 实现真正的流式生成asyncwith aiohttp.ClientSession()as session:asyncwith session.post(f"{self.base_url}/api/generate",
6.                                       json={"model": self.model,"prompt": prompt,"stream":True}, timeout=None)as response:asyncfor line in response.content:if line:yield json.loads(line)["response"]

复制代码

性能优化策略

1. 流式输出性能优化

2. 模型调用优化

温度参数分层：生成阶段使用较高温度(0.7)，批评阶段使用较低温度(0.3)

• Token分配策略：
  
  内容生成：100% max_tokens自我批评：75% max_tokens内容改进：100% max_tokens
  
  
  

技术挑战与解决方案

挑战1：真正的流式输出实现

问题：传统实现是等待完整响应后分段发送，用户体验不佳。
解决方案：

使用Python的异步生成器(AsyncGenerator)实现Server-Sent Events(SSE)长连接前端使用ReadableStream处理流式数据

挑战2：LangGraph状态管理

问题：复杂的状态转换和条件判断。
解决方案：

使用TypedDict定义严格的状态结构实现条件边(conditional edges)处理分支逻辑采用dataclass进行状态封装

挑战3：Ollama集成稳定性

问题：本地模型加载、超时处理、错误恢复。
解决方案：

实现连接池和重试机制添加健康检查和自动重启支持多模型fallback策略

应用输出示例

示例1：写作任务

**任务：**写一篇关于人工智能的简短介绍
Reflection智能体输出：

1. 第一次生成：AI是计算机科学的一个分支。
3. 自我批评：这太简单了，没有提供具体信息。
5. 改进后：人工智能是计算机科学的一个分支，致力于创建能够执行通常需要人类智能的任务的机器。
7. 自我批评：还是没有深入说明AI的具体应用。
9. 最终输出：人工智能（AI）是计算机科学的一个分支，致力于创建能够模拟人类智能行为的机器系统。AI技术包括机器学习、深度学习、自然语言处理等，广泛应用于图像识别、语音助手、自动驾驶等领域，为现代社会带来了革命性的变化。

复制代码

示例2：技术问题解答

**任务：**解释区块链技术的工作原理
Reflexion智能体输出：

1. 初始响应：区块链是一种分布式账本技术...
3. 自省分析：需要更详细的技术细节和实际应用案例
5. 知识缺口：分布式共识机制、具体实现原理
7. 外部搜索：
8. - "区块链分布式共识算法"
9. - "区块链工作原理详解"
11. 整合后响应：区块链是一种分布式账本技术，其核心工作原理包括：
12. 1. 数据结构：区块按时间顺序链接，形成链式结构
13. 2. 分布式共识：通过Proof of Work或Proof of Stake等算法达成网络共识
14. 3. 密码学安全：使用哈希函数和非对称加密确保数据安全
15. 4. 去中心化：无单一控制点，增强系统可靠性
17. 实际应用包括比特币、以太坊、智能合约等...

复制代码

总结

本项目成功构建了一个功能完整的自我改进智能体系统，通过LangGraph实现了Reflection和Reflexion两种核心模式。系统的真正流式输出、参数动态化和Ollama集成等特性，为用户提供了优秀的使用体验。

原文地址：https://blog.csdn.net/whoamidd/article/details/155135420