基于ReAct模式的智能助手开发实践

1. 项目概述

最近在开发一个智能助手项目时，我遇到了传统大模型对话系统的局限性。当用户询问"今天北京的天气怎么样？"时，模型只能回答"我无法获取实时天气信息"。这让我开始研究ReAct模式，它通过"思考-行动-观察"的循环机制，让AI能够主动使用工具获取信息，而不仅仅是依赖训练数据中的知识。

这个项目从零开始实现了一个完整的ReAct Agent系统，包含以下核心组件：

• Agent核心框架：管理对话历史和API调用
• 工具函数集合：提供计算器和价格查询等基础功能
• ReAct执行引擎：驱动"思考-行动-观察"的循环

2. 核心架构设计

2.1 Agent核心框架

Agent类是整个系统的基础，负责与大模型API交互并维护对话历史。核心代码如下：

python复制class Agent:
    """Agent推理框架"""

    def __init__(self, system_prompt=""):
        self._messages = []
        if system_prompt:
            self._messages.append({"role": "system", "content": system_prompt})

    def invoke(self, query: str) -> str:
        """调用Agent进行推理"""
        self._messages.append({"role": "user", "content": query})
        result = self.exec()
        self._messages.append({"role": "assistant", "content": result})
        return result

    def exec(self) -> str:
        """执行推理，返回结果"""
        completion = client.chat.completions.create(
            model=DEFAULT_MODEL,
            messages=self._messages,
            temperature=0  # 推理场景需要确定性输出
        )
        return completion.choices[0].message.content

关键设计点：

1. temperature参数：设为0确保工具调用的准确性
2. 对话历史管理：_messages列表保存完整对话上下文
3. API封装：exec方法统一处理与大模型的交互

2.2 工具系统设计

工具是ReAct Agent扩展能力的关键。我们实现了两个基础工具：

python复制def calculate(expression: str) -> float:
    """计算数学表达式"""
    return eval(expression)

def ask_fruit_unit_price(fruit: str) -> str:
    """查询水果单价"""
    if fruit.casefold() == "apple":
        return "Apple unit price is 10/kg"
    if fruit.casefold() == "banana":
        return "Banana unit price is 6/kg"
    return f"{fruit} unit price is 20/kg"

工具设计原则：

1. 单一职责：每个工具只做一件事
2. 明确接口：输入输出类型清晰
3. 错误处理：返回友好的错误信息

3. ReAct执行引擎实现

3.1 Prompt设计

ReAct模式的核心在于Prompt设计。我们的Prompt包含以下关键部分：

python复制react_prompt = """
You run in a loop of Thought, Action, PAUSE, Observation.
At the end of the loop you output an Answer

Available actions:

calculate:
e.g. calculate: 4 * 7 / 3
Runs a calculation and returns the number

ask_fruit_unit_price:
e.g. ask_fruit_unit_price: apple
Asks for the price of a fruit

Example session:

Question: What is the unit price of apple?
Thought: I need to ask the user for the price of an apple.
Action: ask_fruit_unit_price: apple
PAUSE

Observation: Apple unit price is 10/kg

Answer: The unit price of apple is 10 per kg.
""".strip()

Prompt设计要点：

1. 明确执行流程：Thought → Action → PAUSE → Observation → Answer
2. 提供完整示例：展示从问题到答案的完整路径
3. 工具描述清晰：说明每个工具的用途和调用方式

3.2 执行流程

执行引擎的核心是一个循环，不断执行以下步骤：

1. 调用Agent获取思考结果
2. 解析Action指令
3. 执行对应工具
4. 将观察结果作为下一轮输入

python复制def react(query: str, max_turns: int = 5) -> None:
    agent = Agent(system_prompt=react_prompt)
    question = query

    for turn in range(max_turns):
        result = agent.invoke(question)
        actions = [action_re.match(a) for a in result.split('\n') if action_re.match(a)]
        
        if not actions:
            print("Finish")
            return

        tool, params = actions[0].groups()
        observation = known_tools[tool](params)
        question = f"Observation: {observation}\n\n"

4. 实战经验与优化

4.1 常见问题解决

问题：Agent输出格式不规范
解决方案：

1. 增强Prompt中的格式示例
2. 添加容错解析逻辑

python复制def parse_action(result: str):
    patterns = [
        r'^Action: (\w+): (.*)$',  # 标准格式
        r'^Action: (\w+)\s+(.*)$',  # 容错格式
        r'^(\w+):\s*(.*)$',         # 简化格式
    ]
    for pattern in patterns:
        match = re.search(pattern, result, re.MULTILINE)
        if match:
            return match.groups()
    return None

问题：工具调用失败
解决方案：

1. 添加工具错误处理
2. 返回结构化错误信息

python复制def calculate(expression: str) -> str:
    try:
        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"Error: {str(e)}. Please check your expression."

4.2 性能优化

优化1：工具调用缓存

python复制from functools import lru_cache

@lru_cache(maxsize=100)
def calculate(expression: str) -> str:
    return str(eval(expression))

优化2：并行工具调用

python复制import concurrent.futures

def parallel_tool_calls(tools_and_params):
    with concurrent.futures.ThreadPoolExecutor() as executor:
        futures = []
        for tool_name, params in tools_and_params:
            future = executor.submit(known_tools[tool_name], params)
            futures.append(future)

        results = []
        for future in concurrent.futures.as_completed(futures):
            results.append(future.result())
        return results

5. 扩展方向

这个基础实现可以进一步扩展：

1. 更多工具：添加网络搜索、数据库查询等功能
2. 记忆机制：实现跨会话的记忆存储
3. 可视化：展示推理链路和工具调用过程
4. 错误恢复：自动重试和降级策略
5. 权限控制：限制敏感工具的访问

6. 环境配置

6.1 依赖安装

bash复制pip install openai python-dotenv

6.2 环境变量

创建.env文件：

code复制OPENAI_API_KEY=your_api_key
OPENAI_API_BASE=https://api.deepseek.com/v1
AI_MODEL=deepseek-chat

7. 技术要点总结

1. ReAct循环：Thought → Action → Observation的循环是核心机制
2. 工具设计：每个工具应该简单、专注、有明确接口
3. Prompt工程：清晰的指令和示例对模型表现至关重要
4. 错误处理：需要考虑各种异常情况并妥善处理
5. 性能优化：缓存、并行等技巧可以提升系统效率

在实现过程中，最大的收获是理解了Agent系统如何通过组合简单工具来完成复杂任务。这种模块化的设计思路不仅适用于AI系统，对普通软件开发也有很好的借鉴意义。