ReAct 范式深度解析：从理论到 LangGraph 实践

# 传统AI方法：要么纯推理，要么纯行动
def traditional_reasoning_only(question):
    """纯推理方法：仅基于训练数据回答"""
    return "基于我的训练数据，今天可能是晴天"
    
def traditional_action_only(question):
    """纯行动方法：直接调用API，缺乏思考"""
    # 模拟直接调用API，没有推理过程
    if "天气" in question:
        return "晴天，温度25°C"  # 硬编码结果，没有推理
    return "无法处理"

# ReAct方法：推理和行动交替进行
def react_approach(question):
    """ReAct方法：推理和行动交替进行"""
    # 第1步：推理 - 分析问题
    reasoning = "用户问的是今天某城市的天气，我需要查询实时天气信息"
    
    # 第2步：行动 - 执行查询
    weather_result = weather_api("某城市")
    
    # 第3步：推理 - 分析查询结果
    reasoning = "查询结果显示今天某城市是晴天，温度25度，这是实时准确信息"
    
    # 第4步：行动 - 生成最终答案
    return"今天某城市是晴天，温度25度，适合外出"

from langgraph.prebuilt import create_react_agent
from langchain_core.tools import tool
from langchain_ollama import ChatOllama
from langchain_core.messages import HumanMessage

# 1. 定义工具
@tool
def search_weather(location: str) -> str:
    """搜索指定地点的天气信息"""
    # 模拟天气查询API
    weather_data = {
        "A城市": "晴天，温度25°C，湿度60%",
        "B城市": "多云，温度22°C，湿度70%"
    }
    return weather_data.get(location, f"{location}的天气信息暂时无法获取")

@tool
def calculate_distance(city1: str, city2: str) -> str:
    """计算两个城市之间的距离"""
    # 模拟距离计算API
    distances = {
        ("A城市", "B城市"): "约500公里",
        ("B城市", "A城市"): "约500公里"
    }
    return distances.get((city1, city2), f"{city1}到{city2}的距离信息暂时无法获取")

# 2. 创建模型
model = ChatOllama(model="qwen3:8b", temperature=0.1)

# 3. 创建ReAct Agent
agent = create_react_agent(
    model=model,
    tools=[search_weather, calculate_distance],
    prompt="你是一个天气助手，可以帮助用户查询天气和计算距离。",
    version="v2"
)

# 4. 使用Agent
result = agent.invoke({
    "messages": [HumanMessage(content="A城市和B城市的天气怎么样？距离有多远？")]
})

# 文件路径: libs/langgraph/langgraph/prebuilt/agent_executor.py
def create_react_agent(
    model: BaseChatModel,
    tools: Sequence[BaseTool],
    prompt: Optional[BaseMessage] = None,
    response_format: Optional[Union[Dict, Type[BaseModel]]] = None,
    pre_model_hook: Optional[Callable] = None,
    post_model_hook: Optional[Callable] = None,
    state_schema: Optional[Type[TypedDict]] = None,
    version: Literal["v1", "v2"] = "v2",
) -> CompiledGraph:
    """创建ReAct Agent的核心实现"""
    
    # 1. 定义状态结构
    class AgentState(TypedDict):
        messages: Annotated[Sequence[BaseMessage], add_messages]
        remaining_steps: NotRequired[RemainingSteps]
    
    # 2. 创建图结构
    workflow = StateGraph(AgentState)
    
    # 3. 定义节点
    def agent_node(state: AgentState):
        """Agent节点：调用模型进行推理"""
        model_with_tools = model.bind_tools(tools)
        response = model_with_tools.invoke(state["messages"])
        return {"messages": [response]}
    
    def tools_node(state: AgentState):
        """工具节点：执行工具调用"""
        return tool_node.invoke(state)
    
    # 4. 定义路由逻辑
    def should_continue(state: AgentState) -> str:
        """判断是否需要继续执行工具"""
        last_message = state["messages"][-1]
        if hasattr(last_message, 'tool_calls') and last_message.tool_calls:
            return"tools"
        return"end"
    
    # 5. 组装图结构
    workflow.add_node("agent", agent_node)
    workflow.add_node("tools", tools_node)
    workflow.add_conditional_edges(
        "agent", 
        should_continue, 
        {"tools": "tools", "end": END}
    )
    workflow.add_edge("tools", "agent")
    
    return workflow.compile()

# 文件路径: libs/langgraph/langgraph/prebuilt/agent_executor.py
# 状态是系统的核心
class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]  # 消息历史
    remaining_steps: NotRequired[RemainingSteps]  # 剩余步数

# 文件路径: libs/langgraph/langgraph/prebuilt/agent_executor.py
def should_continue(state: AgentState) -> str:
    """智能路由决策"""
    last_message = state["messages"][-1]
    ifhasattr(last_message, 'tool_calls')and last_message.tool_calls:
        return "tools"  # 需要工具
    return "end"        # 直接结束

# 文件路径: libs/langgraph/langgraph/prebuilt/agent_executor.py
# 工具绑定到模型
model_with_tools = model.bind_tools(tools)
response = model_with_tools.invoke(messages)

# 用户输入
user_input = "A城市和B城市的天气怎么样？距离有多远？"

# 执行过程
result = agent.invoke({
    "messages": [{"role": "user", "content": user_input}]
})

# 执行步骤：
# 1. Agent接收用户消息
# 2. 模型分析：需要查询天气和计算距离
# 3. 调用search_weather工具
# 4. 调用calculate_distance工具  
# 5. 基于工具结果生成最终回复

# 文件路径: libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py
def create_react_agent(
    model: Union[str, LanguageModelLike, Callable],
    tools: Union[Sequence[BaseTool], ToolNode],
    prompt: Optional[Prompt] = None,
    response_format: Optional[StructuredResponseSchema] = None,
    state_schema: Optional[StateSchemaType] = None,
    version: Literal["v1", "v2"] = "v2",
) -> CompiledStateGraph:
    """
    创建ReAct Agent的核心函数
    
    参数说明：
    - model: 语言模型实例
    - tools: 工具列表或工具节点
    - prompt: 自定义提示词
    - response_format: 结构化输出格式
    - state_schema: 状态模式
    - version: 版本选择
    """

# 文件路径: libs/langgraph/langgraph/prebuilt/agent_executor.py
if state_schema is None:
    state_schema = (
        AgentStateWithStructuredResponse
        if response_format is not None
        else AgentState
    )

# 文件路径: libs/langgraph/langgraph/prebuilt/agent_executor.py
if _should_bind_tools(model, tool_classes, num_builtin=len(llm_builtin_tools)):
    model = model.bind_tools(tool_classes + llm_builtin_tools)

# 文件路径: libs/langgraph/langgraph/prebuilt/agent_executor.py
def should_continue(state: StateSchema) -> Union[str, list[Send]]:
    messages = _get_state_value(state, "messages")
    last_message = messages[-1]
    
    ifnot isinstance(last_message, AIMessage) ornot last_message.tool_calls:
        return END
    else:
        return"tools"

# 传统方法：全局状态，难以管理
classTraditionalAgent:
    def __init__(self):
        self.messages = []
        self.tool_results = []
        self.current_step = 0
        # ... 更多状态变量

# LangGraph方法：状态模式，类型安全
# 文件路径: libs/langgraph/langgraph/prebuilt/agent_executor.py
class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]
    remaining_steps: NotRequired[RemainingSteps]

# 文件路径: libs/langgraph/langgraph/prebuilt/agent_executor.py
def should_continue(state: AgentState) -> str:
    messages = state["messages"]
    last_message = messages[-1]
    
    # 关键设计：不是检查所有消息，而是只检查最后一条
    if isinstance(last_message, AIMessage) and last_message.tool_calls:
        return"tools"
    return"end"

# 文件路径: libs/langgraph/langgraph/prebuilt/agent_executor.py
# 核心设计：工具绑定到模型，而不是单独管理
model_with_tools = model.bind_tools(tools)
response = model_with_tools.invoke(state["messages"])

# 最初的简单方案
def generate_outline(user_input):
    """纯大模型方案：直接生成PPT大纲"""
    prompt = f"""
    用户需求：{user_input}
    请生成一份完整的PPT大纲，包含：
    1. 封面页
    2. 目录页
    3. 产品介绍
    4. 功能特性
    5. 应用案例
    6. 总结页
    """
    return llm.invoke(prompt)

# 改进方案：增加知识检索
def generate_outline_with_knowledge(user_input):
    # 检索客户信息
    customer_info = search_customer_knowledge(user_input)
    # 检索产品信息
    product_info = search_product_knowledge(user_input)
    # 检索行业信息
    industry_info = search_industry_knowledge(user_input)
    
    prompt = f"""
    用户需求：{user_input}
    客户信息：{customer_info}
    产品信息：{product_info}
    行业信息：{industry_info}
    
    请生成一份完整的PPT大纲...
    """
    return llm.invoke(prompt)

# ReAct方案：智能判断
def react_outline_generation(user_input):
    # 1. 推理：分析用户真正需要什么
    reasoning = analyze_user_intent(user_input)
    
    # 2. 行动：根据推理结果决定是否检索
    if reasoning.needs_customer_info:
        customer_info = search_customer_knowledge(user_input)
    if reasoning.needs_product_info:
        product_info = search_product_knowledge(user_input)
    
    # 3. 观察：基于检索结果调整策略
    # 4. 调整：生成符合用户真实需求的大纲
    return generate_adaptive_outline(reasoning, retrieved_info)

classSmartOutlineReActAgent:
    """智能大纲生成ReAct Agent - 核心设计思路"""
    
    def __init__(self):
        # 核心组件：工具 + 模型 + Agent
        self.tools = self._load_tools()           # 分层工具设计
        self.llm = self._init_llm()              # 模型初始化
        self.agent = self._create_react_agent()   # ReAct Agent
    
    def _load_tools(self):
        """工具分层设计：通用工具 + 专业工具"""
        general_tools = load_general_tools()      # 时间、搜索等
        internal_tools = load_internal_tools()    # 内部知识库
        return general_tools + internal_tools
    
    def _create_react_agent(self):
        """构建ReAct Agent - 核心执行引擎"""
        return create_react_agent(
            model=self.llm,
            tools=self.tools,
            prompt=self.system_prompt,
            version="v2"  # 支持并行工具调用
        )
    
    def generate_outline(self, user_input: str) -> str:
        """生成PPT大纲 - 核心业务逻辑"""
        # 1. 构建输入
            messages = [HumanMessage(content=user_input)]
            
        # 2. 执行ReAct循环
        result = self.agent.invoke({
            "messages": messages,
            "recursion_limit": 50,
            "max_iterations": 20
        })
        
        # 3. 返回结果
        return result["messages"][-1].content

def create_react_agent(model, tools, prompt=None, version="v2"):
    """创建ReAct Agent的核心实现"""
    
    # 1. 状态模式定义
    class AgentState(TypedDict):
        messages: Annotated[Sequence[BaseMessage], add_messages]
        remaining_steps: int
    
    # 2. 图结构构建
    workflow = StateGraph(AgentState)
    
    # 3. 节点定义
    def agent_node(state: AgentState):
        """Agent节点 - 核心推理引擎"""
        # 绑定工具到模型
        model_with_tools = model.bind_tools(tools)
        
        # 调用模型
        response = model_with_tools.invoke(state["messages"])
        
        return {"messages": [response]}
    
    def tools_node(state: AgentState):
        """Tools节点 - 工具执行引擎"""
        if version == "v1":
            # v1版本：串行执行所有工具
            return tool_node.invoke(state)
        else:
            # v2版本：并行执行工具（使用Send API）
            return tool_node.invoke(state)
    
    # 4. 条件路由逻辑
    def should_continue(state: AgentState) -> str:
        """智能路由决策"""
        messages = state["messages"]
        last_message = messages[-1]
        
        # 检查是否包含工具调用
        if isinstance(last_message, AIMessage) and last_message.tool_calls:
            return"tools"
        return"end"
    
    # 5. 图结构组装
    workflow.add_node("agent", agent_node)
    workflow.add_node("tools", tools_node)
    workflow.add_conditional_edges(
        "agent", 
        should_continue, 
        {"tools": "tools", "end": END}
    )
    workflow.add_edge("tools", "agent")
    
    # 6. 编译并返回
    return workflow.compile()

# LangGraph底层执行引擎核心代码分析
classPregel:
    """LangGraph的核心执行引擎 - 基于Pregel算法"""
   
    def __init__(self, nodes: Dict[str, Any], edges: List[Edge]):
        self.nodes = nodes
        self.edges = edges
        self.state_schema = self._build_state_schema()
   
    def invoke(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
        """执行图的核心方法"""
        # 1. 状态初始化
        state = self._initialize_state(inputs)
       
        # 2. 执行循环
        whilenot self._is_complete(state):
            # 3. 节点调度
            next_nodes = self._get_next_nodes(state)
           
            # 4. 并行执行节点
            updates = self._execute_nodes_parallel(next_nodes, state)
           
            # 5. 状态更新
            state = self._update_state(state, updates)
           
            # 6. 检查终止条件
            if self._should_terminate(state):
                break
       
        return state
   
    def _execute_nodes_parallel(self, nodes: List[str], state: Dict) -> Dict:
        """并行执行多个节点 - 性能优化的关键"""
        import asyncio
       
        async def execute_node(node_name: str):
            node_func = self.nodes[node_name]
            return await node_func(state)
       
        # 并行执行所有节点
        tasks = [execute_node(node) for node in nodes]
        results = asyncio.gather(*tasks)
       
        return self._merge_results(results)
   
    def _get_next_nodes(self, state: Dict) -> List[str]:
        """智能节点调度 - 基于条件路由"""
        next_nodes = []
       
        for edge in self.edges:
            if edge.condition(state):
                next_nodes.append(edge.target)
       
        return next_nodes

# 工具1：内部知识库搜索 - 解决信息获取问题
def search_internal_knowledge(query: str, knowledge_type: str = "all") -> str:
    """按需检索，避免信息过载"""
    # 根据知识类型选择性搜索
        if knowledge_type == "customer":
        results = customer_knowledge_base.search(query, limit=3)
        elif knowledge_type == "product":
        results = product_knowledge_base.search(query, limit=3)
        else:
        results = knowledge_base.search(query, limit=5)
    
    # 返回原始数据，让AI自己判断如何使用
    return format_search_results(results)

# 工具2：用户意图分析 - 解决"固化输出"问题
def analyze_user_intent(user_input: str) -> str:
    """让AI理解用户的真实需求"""
    intent = {
        "needs_customer_info": "客户" in user_input,
        "needs_product_info": "产品" in user_input,
        "output_scope": "partial"if"只要" in user_input else"full",
        "page_count": extract_page_count(user_input)
    }
    return json.dumps(intent)

# 工具设计核心原则
"""
1. 工具只负责获取信息，不负责分析
2. 工具只提供数据，不提供结论
3. 工具支持LLM推理，不替代LLM推理
4. 保持工具简单、纯净、单一职责
"""

@tool
def web_search(query: str) -> str:
    """获取互联网最新信息"""
    return search_engine.search(query)

@tool
def search_internal_knowledge(query: str) -> str:
    """获取内部文档和知识"""
    return knowledge_base.search(query)

## 核心工作原则

### 思考引导
- **智能分析**：分析用户输入的信息丰富度和需求复杂度
- **深度推理**：基于收集的信息进行多维度推理和关联分析
- **自主决策**：根据分析结果，自主选择执行策略和工具调用方式
- **价值导向**：每页内容都要体现客户价值和业务成果

### 深度思考框架
**多维度分析策略**：
- **客户维度**：规模、发展阶段、业务模式、技术成熟度
- **产品维度**：核心能力、技术优势、应用场景、实施复杂度
- **行业维度**：发展趋势、竞争格局、转型路径、成功要素
- **竞争维度**：竞品分析、差异化优势、价值主张、市场定位

# 错误的提示词设计
prompt = """
如果用户询问天气，调用天气API
如果用户询问时间，调用时间API
如果用户询问新闻，调用新闻API
"""

# 正确的提示词设计
prompt = """
分析用户需求，判断需要什么信息，然后选择合适的工具。
考虑以下因素：
1. 用户意图的明确程度
2. 所需信息的类型和来源
3. 当前上下文的相关性
4. 用户可能的后续需求
"""

# 错误的质量检查
quality_check = [
    "是否包含所有必要信息？",
    "是否格式正确？",
    "是否长度合适？"
]

# 正确的质量框架
quality_framework = """
评估内容质量：
1. 信息完整性：是否回答了用户的核心问题？
2. 逻辑一致性：内容是否逻辑清晰、前后一致？
3. 价值相关性：是否对用户有实际价值？
4. 可操作性：用户是否能够基于此内容采取行动？
"""

# 通用模板的问题
generic_prompt = "请分析以下内容并提供建议"

# 场景适配的优势
business_analysis_prompt = """
作为业务分析师，请分析以下内容：
1. 识别关键业务指标和趋势
2. 分析潜在风险和机会
3. 提供具体的改进建议
4. 考虑实施可行性和成本效益
"""

# 不好的提示词
prompt = """
你必须按照以下步骤：
1. 先分析问题
2. 再调用工具
3. 最后给出答案
"""

# 好的提示词
prompt = """
你是一个智能助手，能够帮助用户解决各种问题。
当遇到需要外部信息的问题时，你可以主动调用相关工具。
请根据问题的具体情况，灵活地选择最合适的解决方案。
"""

# 不好的设计
def fixed_workflow(question):
    if "天气" in question:
        returncall_weather_api()
    elif "订单" in question:
        returncall_order_api()
    else:
        return "我无法处理"

# 好的设计
def flexible_workflow(question):
    # 让AI自己决定需要什么工具
    return agent.invoke({"messages": [{"role": "user", "content": question}]})

# 不好的设计
def check_quality(response):
    checks = [
        "是否包含关键词",
        "是否超过100字",
        "是否包含标点符号"
    ]
    return all(checks)

# 好的设计
def assess_quality(response, context):
    # 基于上下文和用户需求评估质量
    return quality_score

# 解决方案：在提示词中强调判断能力
prompt = """
你是一个智能助手。当用户问题简单明确时，可以直接回答；
当需要外部信息时，才调用相关工具。
请根据问题的复杂度和信息需求，合理选择处理方式。
"""

# 解决方案：优雅的错误处理
def robust_tool_call(tool_name, tool_input):
    try:
        result = tool_name.invoke(tool_input)
        return result
    except Exception as e:
        return f"工具调用失败：{str(e)}，让我尝试其他方法"

# 解决方案：设置最大迭代次数
classReActAgent:
    def __init__(self, max_iterations=5):
        self.max_iterations = max_iterations
        self.current_iteration = 0
    
    def should_continue(self, state):
        self.current_iteration += 1
        if self.current_iteration >= self.max_iterations:
            return"end"
        # 其他逻辑...

# 问题：每次都要检索大量信息，即使不需要
def generate_outline_with_knowledge(user_input):
    # 问题：无论用户需要什么，都检索所有信息
    customer_info = search_customer_knowledge(user_input)  # 耗时2-3秒
    product_info = search_product_knowledge(user_input)   # 耗时2-3秒
    industry_info = search_industry_knowledge(user_input)  # 耗时2-3秒
    
    # 总耗时：6-9秒，用户体验很差
    return generate_outline(customer_info, product_info, industry_info)

# 解决方案：ReAct智能判断
def react_outline_generation(user_input):
    # 1. 先分析用户意图
    intent = analyze_user_intent(user_input)
    
    # 2. 按需检索
    retrieved_info = {}
    if intent.needs_customer_info:
        retrieved_info["customer"] = search_customer_knowledge(user_input)
    if intent.needs_product_info:
        retrieved_info["product"] = search_product_knowledge(user_input)
    if intent.needs_industry_info:
        retrieved_info["industry"] = search_industry_knowledge(user_input)
    
    # 3. 基于检索结果生成大纲
    return generate_adaptive_outline(intent, retrieved_info)

# 问题：总是生成"完整"的PPT大纲
用户说："我只要3页介绍产品功能"
系统回答：生成20页的完整PPT大纲  # 用户体验差

# 解决方案：智能判断输出范围
def generate_adaptive_outline(intent, retrieved_info):
    if intent.output_scope == "partial":
        # 只生成用户需要的部分
        return generate_partial_outline(intent.focus_areas, retrieved_info)
    elif intent.output_scope == "full":
        # 生成完整大纲
        return generate_full_outline(retrieved_info)
    else:
        # 根据页数要求生成
        return generate_specific_outline(intent.page_count, retrieved_info)

# 问题：AI过度依赖工具，即使问题很简单
用户问："什么是PPT？"
AI回答：我需要搜索一下PPT的定义...  # 不必要的工具调用

# 解决方案：智能判断是否需要工具
def should_use_tools(user_input, intent):
    # 简单问题直接回答
    if intent.complexity == "simple":
        return False
    
    # 需要外部信息才调用工具
    if intent.needs_external_info:
        return True
    
    return False

def get_adaptive_config(task_complexity: str) -> Dict[str, Any]:
    configs = {
        "simple": {
            "recursion_limit": 20,
            "max_iterations": 10,
            "timeout": 30
        },
        "complex": {
            "recursion_limit": 50,
            "max_iterations": 25,
            "timeout": 120
        }
    }
    return configs.get(task_complexity, configs["simple"])

# 支持并行工具调用
def parallel_tool_calls(tools: List[BaseTool], inputs: List[Dict]) -> List[str]:
    import asyncio
    
    async def call_tool(tool, input_data):
        return await tool.ainvoke(input_data)
    
    tasks = [call_tool(tool, input_data) for tool, input_data in zip(tools, inputs)]
    return asyncio.run(asyncio.gather(*tasks))

import time
from collections import defaultdict

classRateLimiter:
    def __init__(self, max_calls_per_minute=60):
        self.max_calls = max_calls_per_minute
        self.calls = defaultdict(list)
    
    def can_call(self, tool_name: str) -> bool:
        now = time.time()
        # 清理1分钟前的调用记录
        self.calls[tool_name] = [t for t in self.calls[tool_name] if now - t < 60]
        return len(self.calls[tool_name]) < self.max_calls
    
    def record_call(self, tool_name: str):
        self.calls[tool_name].append(time.time())

# 创建全局频率限制器实例
rate_limiter = RateLimiter()

def _handle_state_update(self, data: Any, step_count: int, outline_id: str, source: str = "", message_context: Dict[str, Any] = None):
    # 处理工具调用开始
    ifisinstance(last_message, AIMessage)and last_message.tool_calls:
        for tool_call in last_message.tool_calls:
            send_outline_tool_call_start(
                outline_id=outline_id,
                tool_name=tool_name,
                tool_args=tool_args,
                step_count=step_count
            )
    
    # 处理工具调用完成
    elif isinstance(last_message, ToolMessage):
        if is_error:
            send_outline_tool_call_error(...)
        else:
            send_outline_tool_call_complete(...)

# 手写ReAct Agent
# 安装：pip install langchain langchain-ollama

from langchain_ollama import ChatOllama
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from typing import Dict, List, Any, Optional
import re
import time

classMiniReActAgent:
    """手写ReAct Agent """
    
    def __init__(self):
        self.model = ChatOllama(model="qwen3:8b", temperature=0.1)
        self.tools = {
            "time": self._get_time,
            "calc": self._calculate,
            "search": self._search
        }
        # 状态管理
        self.state = {
            "messages": [],
            "current_step": "start",
            "tool_results": {},
            "reasoning": ""
        }
    
    def _get_time(self, param=""):
        """获取当前时间"""
        import datetime
        now = datetime.datetime.now()
        return f"现在是{now.strftime('%Y年%m月%d日 %H:%M')}"
    
    def _calculate(self, expression: str):
        """安全计算表达式"""
        try:
            ifnot re.match(r'^[0-9+\-*/.() ]+$', expression):
                return "计算错误：表达式包含非法字符"
            return str(eval(expression))
        except:
            return "计算错误：表达式无效"
    
    def _search(self, query: str):
        """模拟搜索功能"""
        return f"关于'{query}'的搜索结果..."
    
    def should_continue(self, state: Dict) -> str:
        """条件路由 """
        last_message = state["messages"][-1] if state["messages"] else None
        
        # 检查AI回复中是否包含工具调用
        if isinstance(last_message, AIMessage):
            content = last_message.content
            if "工具:" in content and "参数:" in content:
                return "tools"  # 需要工具
        return "end"        # 直接结束
    
    def agent_node(self, state: Dict) -> Dict:
        """Agent节点 """
        # 构建工具描述 - 让AI知道有哪些工具可用
        tool_descriptions = []
        for tool_name, tool_func in self.tools.items():
            if tool_name == "time":
                tool_descriptions.append("time: 获取当前时间(无需参数)")
            elif tool_name == "calc":
                tool_descriptions.append("calc: 计算数学表达式(需要表达式参数)")
            elif tool_name == "search":
                tool_descriptions.append("search: 搜索信息(需要搜索关键词)")
        
        # 构建提示词
        last_message = state["messages"][-1].content
        
        # 检查是否有工具执行结果
        if "工具执行结果:" in last_message:
            # 基于工具结果继续推理
        prompt = f"""
            用户问题: {state['messages'][0].content}
            工具执行结果: {last_message}
            
            请基于工具执行结果回答用户问题。
            如果还需要更多信息，请回答: 工具: [工具名] 参数: [参数]
            如果信息足够，请直接回答用户问题。
            """
        else:
            # 初始推理
            prompt = f"""
            用户问题: {last_message}
            
            可用工具: {', '.join(tool_descriptions)}
        
        请分析问题并决定是否需要使用工具。
            如果需要工具，请回答: 工具: [工具名] 参数: [参数]
            注意：time工具无需参数，直接写"工具: time 参数: 无"
            如果不需要工具，请直接回答用户问题。
        """
        
        # 调用模型
        response = self.model.invoke([HumanMessage(content=prompt)])
        
        # 更新状态
        new_state = state.copy()
        new_state["messages"] = state["messages"] + [AIMessage(content=response.content)]
        new_state["current_step"] = "agent"
        
        return new_state
    
    def tools_node(self, state: Dict) -> Dict:
        """Tools节点 """
        last_message = state["messages"][-1]
        
        if isinstance(last_message, AIMessage):
            content = last_message.content
            
            # 解析工具调用
            if "工具:" in content and "参数:" in content:
                # 提取工具名和参数
                tool_match = re.search(r'工具:\s*(\w+)', content)
                param_match = re.search(r'参数:\s*(.+)', content)
                
                if tool_match and param_match:
                    tool_name = tool_match.group(1)
                    param = param_match.group(1).strip()
                    
                    # 执行工具
                    if tool_name in self.tools:
                        # 处理time工具的无参数调用
                        if tool_name == "time" and (param == "无" or param == "" or param == "[]"):
                            result = self.tools[tool_name]("")
        else:
                            result = self.tools[tool_name](param)
                        
                        # 更新状态
                        new_state = state.copy()
                        new_state["messages"] = state["messages"] + [AIMessage(content=f"工具执行结果: {result}")]
                        new_state["current_step"] = "tools"
                        
                        return new_state
        
        return state
    
    def react_cycle(self, question: str):
        """ReAct核心循环 """
        print(f"用户问题: {question}")
        
        # 初始化状态
        self.state = {
            "messages": [HumanMessage(content=question)],
            "current_step": "start",
            "tool_results": {},
            "reasoning": ""
        }
        
        # 执行循环 
        max_iterations = 5
        for i in range(max_iterations):
            print(f"\n--- 第{i+1}轮执行 ---")
            
            # 1. Agent节点
            self.state = self.agent_node(self.state)
            print(f"Agent推理: {self.state['messages'][-1].content}")
            
            # 2. 条件路由 
            next_step = self.should_continue(self.state)
            print(f"路由决策: {next_step}")
            
            if next_step == "tools":
                # 3. Tools节点
                self.state = self.tools_node(self.state)
                print(f"工具执行结果: {self.state['messages'][-1].content}")
                
                # 4. 继续下一轮推理
                continue
            else:
                # 5. 结束
                print("执行完成")
                break
        
        return self.state["messages"][-1].content

# 使用示例
if __name__ == "__main__":
    agent = MiniReActAgent()
    
    # 测试不同场景
    agent.react_cycle("现在几点了？")
    print("\n" + "="*50 + "\n")
    agent.react_cycle("计算 15 + 27")
    print("\n" + "="*50 + "\n")
    agent.react_cycle("什么是人工智能？")