AI创想

标题: 基于LangGraph的自我改进智能体：Reflection与Reflexion技术详解与实现 [打印本页]

作者: 创想小编 时间: 昨天 22:53
标题: 基于LangGraph的自我改进智能体：Reflection与Reflexion技术详解与实现
作者：大模型RAG和Agent技术实践
文章目录

目录

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

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

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

(, 下载次数: 0)

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

核心概念

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

(, 下载次数: 0)

Reflection模式工作流程

(, 下载次数: 0)

关键技术特征

双重身份模式

迭代改进循环

质量评估机制

LangGraph实现架构

(, 下载次数: 0)

代码实现核心

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

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

复制代码

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

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

复制代码

2. 批评节点（critique_content）

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

复制代码

3. 改进节点（improve_content）

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

复制代码

优势与特点

优势：

适用场景：

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

核心概念

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

(, 下载次数: 0)

Reflexion模式工作流程

(, 下载次数: 0)

关键技术特征

三元角色模式

知识检索循环

智能查询生成

LangGraph实现架构

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

(, 下载次数: 0)

代码实现核心

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

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

复制代码

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

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

复制代码

2. 知识检索节点（search_knowledge）

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

复制代码

3. 知识整合节点（integrate_knowledge）

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

复制代码

优势与特点

优势：

适用场景：

两种智能体模式对比

架构对比

(, 下载次数: 0)

性能对比表

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

选择策略

选择Reflection的场景：

选择Reflexion的场景：

三. 系统整体架构

(, 下载次数: 0)

四. 关键技术实现

1. 真正的流式输出架构

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

(, 下载次数: 0)

核心代码实现：

# 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):
generated_content += chunk
yield json.dumps({"type":"generate_chunk","iteration": current_iteration,"chunk": chunk
})+"\n"

复制代码

2. 配置参数动态化

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

(, 下载次数: 0)

配置示例：

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

复制代码

3. LangGraph工作流状态管理

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

(, 下载次数: 0)

LangGraph工作流构建

defbuild_reflection_graph(self)-> StateGraph:"""构建Reflection智能体工作流图"""
workflow = StateGraph(ReflectionState)# 添加节点
workflow.add_node("generate", self._generate_content_node)
workflow.add_node("critique", self._critique_content_node)
workflow.add_node("improve", self._improve_content_node)
workflow.add_node("finalize", self._finalize_node)# 定义边和条件
workflow.add_edge(START,"generate")
workflow.add_edge("generate","critique")
workflow.add_conditional_edges("critique",
self._should_continue,{"continue":"improve","finish":"finalize"})
workflow.add_conditional_edges("improve",
self._check_iteration_limit,{"continue":"critique","finish":"finalize"})
workflow.add_edge("finalize", END)return workflow

复制代码

4. Ollama本地模型集成

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

(, 下载次数: 0)

Ollama客户端实现：

# backend/src/utils/llm_manager.pyclassOllamaClient:def__init__(self, base_url:str, model:str,**kwargs):
self.base_url = base_url
self.model = model
self.temperature = kwargs.get('temperature',0.7)
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",
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. 流式输出性能优化

(, 下载次数: 0)

2. 模型调用优化

温度参数分层

Token分配策略：

技术挑战与解决方案

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

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

挑战2：LangGraph状态管理

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

挑战3：Ollama集成稳定性

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

应用输出示例

示例1：写作任务

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

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

复制代码

示例2：技术问题解答

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

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

复制代码

总结

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

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

欢迎光临 AI创想 (https://www.llms-ai.com/)