Day13：迈向智能交互：从 ReAct Agent 开发到优化的全历程

一、引言

在当今人工智能应用的浪潮中，开发智能 Agent 成为解锁大语言模型潜力的关键。本周我们将深入学习如何开发一个完整的 ReAct Agent，并在后续对其进行复盘、调试与优化。以通义千问大模型为依托，逐步构建一个功能强大、交互友好的智能 Agent。

二、第一个完整 ReAct Agent 开发

核心任务：整合知识开发完整 ReAct Agent

1. 工具选择与整合：我们需要整合多种工具，使 Agent 能够根据用户问题自主选择合适工具。例如，我们选择计算器、搜索和文件读取作为初始的三个工具。对于计算器工具，用于处理数值计算问题；搜索工具，可针对用户信息查询需求；文件读取工具，则能满足读取本地文件内容的要求。

2. 工具函数定义：

   • 计算器工具：

class CalculatorParams(BaseModel):
    num1: float
    num2: float
    operation: str

    @validator('operation')
    def valid_operation(cls, v):
        valid_ops = ['+', '-', '*', '/']
        if v not in valid_ops:
            raise ValueError('不支持的运算操作')
        return v


def calculate(params: CalculatorParams):
    if params.operation == '+':
        return params.num1 + params.num2
    elif params.operation == '-':
        return params.num1 - params.num2
    elif params.operation == '*':
        return params.num1 * params.num2
    elif params.operation == '/':
        if params.num2 == 0:
            print("除数不能为零")
            return None
        return params.num1 / params.num2

搜索工具：假设使用一个模拟的搜索 API，实际需替换为真实 API。

class SearchParams(BaseModel):
    query: str
    api_key: str

    @validator('query')
    def query_not_empty(cls, v):
        if not v.strip():
            raise ValueError('查询内容不能为空')
        return v

    @validator('api_key')
    def api_key_not_empty(cls, v):
        if not v.strip():
            raise ValueError('api_key不能为空')
        return v


def search(params: SearchParams):
    base_url = "https://api.example.com/search"
    params_dict = {
        "key": params.api_key,
        "q": params.query
    }
    try:
        response = requests.get(base_url, params=params_dict)
        response.raise_for_status()
        return response.json()
    except requests.RequestException as e:
        print(f"搜索失败: {e}")
        return None

文件读取工具：

class ReadFileParams(BaseModel):
    file_path: str

    @validator('file_path')
    def file_path_exists(cls, v):
        if not os.path.exists(v):
            raise ValueError('文件路径不存在')
        return v


def read_file(params: ReadFileParams):
    try:
        with open(params.file_path, 'r', encoding='utf - 8') as f:
            return f.read()
    except Exception as e:
        print(f"读取文件失败: {e}")
        return None

工具选择逻辑：借助通义千问大模型的推理能力，判断用户问题所需工具。通过向通义千问发送用户问题和工具描述，模型决定调用哪个工具。

def choose_tool(user_input, api_key):
    tools = [
        {
            "name": "calculate",
            "parameters": {
                "type": "object",
                "properties": {
                    "num1": {"type": "number"},
                    "num2": {"type": "number"},
                    "operation": {"type": "string"}
                },
                "required": ["num1", "num2", "operation"]
            },
            "description": "用于执行简单的数学运算"
        },
        {
            "name": "search",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string"},
                    "api_key": {"type": "string"}
                },
                "required": ["query", "api_key"]
            },
            "description": "用于搜索信息"
        },
        {
            "name": "read_file",
            "parameters": {
                "type": "object",
                "properties": {
                    "file_path": {"type": "string"}
                },
                "required": ["file_path"]
            },
            "description": "用于读取本地文件内容"
        }
    ]
    prompt = {
        "user_input": user_input,
        "functions": tools
    }
    response = call_qwen_api(api_key, json.dumps(prompt))
    if "function_call" in response:
        function_call = response["function_call"]
        return function_call["name"], function_call["parameters"]
    return None, None

结果解析与回答生成：不同工具返回结果格式不同，需要针对性解析。例如，计算器返回简单数值，搜索可能返回 JSON 格式数据，文件读取返回文本内容。解析后，构建提示让通义千问生成自然语言回答。

def parse_calculator_result(result):
    if result is not None:
        return f"计算结果为{result}"
    return ""


def parse_search_result(result):
    if result:
        # 假设搜索结果结构，实际需根据真实API调整
        return f"搜索结果: {result.get('results', [])}"
    return ""


def parse_read_file_result(result):
    if result:
        return f"文件内容: {result}"
    return ""


def generate_answer(parsed_result, api_key):
    prompt = f"根据以下信息生成回答: {parsed_result}"
    response = call_qwen_api(api_key, prompt)
    if response:
        return response["choices"][0]["message"]["content"]
    return "无法生成回答"

完整 ReAct Agent 代码：

import requests
import json
import os
from pydantic import BaseModel, validator


# 计算器工具
class CalculatorParams(BaseModel):
    num1: float
    num2: float
    operation: str

    @validator('operation')
    def valid_operation(cls, v):
        valid_ops = ['+', '-', '*', '/']
        if v not in valid_ops:
            raise ValueError('不支持的运算操作')
        return v


def calculate(params: CalculatorParams):
    if params.operation == '+':
        return params.num1 + params.num2
    elif params.operation == '-':
        return params.num1 - params.num2
    elif params.operation == '*':
        return params.num1 * params.num2
    elif params.operation == '/':
        if params.num2 == 0:
            print("除数不能为零")
            return None
        return params.num1 / params.num2


# 搜索工具
class SearchParams(BaseModel):
    query: str
    api_key: str

    @validator('query')
    def query_not_empty(cls, v):
        if not v.strip():
            raise ValueError('查询内容不能为空')
        return v

    @validator('api_key')
    def api_key_not_empty(cls, v):
        if not v.strip():
            raise ValueError('api_key不能为空')
        return v


def search(params: SearchParams):
    base_url = "https://api.example.com/search"
    params_dict = {
        "key": params.api_key,
        "q": params.query
    }
    try:
        response = requests.get(base_url, params=params_dict)
        response.raise_for_status()
        return response.json()
    except requests.RequestException as e:
        print(f"搜索失败: {e}")
        return None


# 文件读取工具
class ReadFileParams(BaseModel):
    file_path: str

    @validator('file_path')
    def file_path_exists(cls, v):
        if not os.path.exists(v):
            raise ValueError('文件路径不存在')
        return v


def read_file(params: ReadFileParams):
    try:
        with open(params.file_path, 'r', encoding='utf - 8') as f:
            return f.read()
    except Exception as e:
        print(f"读取文件失败: {e}")
        return None


# 模拟通义千问API调用（实际需替换为真实API调用）
def call_qwen_api(api_key, prompt):
    url = "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {api_key}"
    }
    data = {
        "model": "qwen-plus",
        "messages": [
            {"role": "user", "content": prompt}
        ],
        "stream": False
    }
    try:
        response = requests.post(url, headers=headers, json=data)
        response.raise_for_status()
        return response.json()
    except requests.RequestException as e:
        print(f"调用通义千问API失败: {e}")
        return None


def choose_tool(user_input, api_key):
    tools = [
        {
            "name": "calculate",
            "parameters": {
                "type": "object",
                "properties": {
                    "num1": {"type": "number"},
                    "num2": {"type": "number"},
                    "operation": {"type": "string"}
                },
                "required": ["num1", "num2", "operation"]
            },
            "description": "用于执行简单的数学运算"
        },
        {
            "name": "search",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string"},
                    "api_key": {"type": "string"}
                },
                "required": ["query", "api_key"]
            },
            "description": "用于搜索信息"
        },
        {
            "name": "read_file",
            "parameters": {
                "type": "object",
                "properties": {
                    "file_path": {"type": "string"}
                },
                "required": ["file_path"]
            },
            "description": "用于读取本地文件内容"
        }
    ]
    prompt = {
        "user_input": user_input,
        "functions": tools
    }
    response = call_qwen_api(api_key, json.dumps(prompt))
    if "function_call" in response:
        function_call = response["function_call"]
        return function_call["name"], function_call["parameters"]
    return None, None


def parse_calculator_result(result):
    if result is not None:
        return f"计算结果为{result}"
    return ""


def parse_search_result(result):
    if result:
        return f"搜索结果: {result.get('results', [])}"
    return ""


def parse_read_file_result(result):
    if result:
        return f"文件内容: {result}"
    return ""


def generate_answer(parsed_result, api_key):
    prompt = f"根据以下信息生成回答: {parsed_result}"
    response = call_qwen_api(api_key, prompt)
    if response:
        return response["choices"][0]["message"]["content"]
    return "无法生成回答"


def react_agent(user_input, api_key):
    tool_name, tool_params = choose_tool(user_input, api_key)
    if tool_name == 'calculate':
        params = CalculatorParams(**tool_params)
        result = calculate(params)
        parsed_result = parse_calculator_result(result)
    elif tool_name =='search':
        params = SearchParams(**tool_params)
        result = search(params)
        parsed_result = parse_search_result(result)
    elif tool_name =='read_file':
        params = ReadFileParams(**tool_params)
        result = read_file(params)
        parsed_result = parse_read_file_result(result)
    else:
        return "无法确定要使用的工具"
    return generate_answer(parsed_result, api_key)

补充任务：优化交互体验

1. 用户输入提示：在程序开始时，提示用户输入问题，使交互更友好。

2. 执行步骤打印：在 Agent 运行过程中，打印选择的工具、传入的参数以及中间结果，方便追溯问题。

if __name__ == "__main__":
    api_key = "your_api_key"
    user_input = input("请输入你的问题: ")
    print(f"用户输入: {user_input}")
    tool_name, tool_params = choose_tool(user_input, api_key)
    print(f"选择的工具: {tool_name}")
    print(f"工具参数: {tool_params}")
    if tool_name == 'calculate':
        params = CalculatorParams(**tool_params)
        result = calculate(params)
        print(f"计算结果: {result}")
        parsed_result = parse_calculator_result(result)
    elif tool_name =='search':
        params = SearchParams(**tool_params)
        result = search(params)
        print(f"搜索结果: {result}")
        parsed_result = parse_search_result(result)
    elif tool_name =='read_file':
        params = ReadFileParams(**tool_params)
        result = read_file(params)
        print(f"文件读取结果: {result}")
        parsed_result = parse_read_file_result(result)
    else:
        print("无法确定要使用的工具")
        exit()
    answer = generate_answer(parsed_result, api_key)
    print(f"生成的回答: {answer}")

三、 Agent 调试优化

核心任务：复盘本周学习内容

1. Function Call：回顾如何使用 Function Call 让大模型调用外部工具，包括工具函数定义、参数校验以及与大模型的交互流程。

2. ReAct 范式：总结 ReAct 范式的 “推理→动作→观察→再推理” 流程，以及它如何与 Function Call 结合，使 Agent 能够根据用户问题合理选择并调用工具。

3. Agent 开发：梳理开发完整 ReAct Agent 的过程，从工具选择、工具函数实现、结果解析到回答生成的各个环节，分析每个环节的作用和实现方法。

补充任务：调试优化 Agent

1. 修复已知 bug：

   • 工具选择错误：检查工具选择逻辑，确保通义千问返回的工具选择与用户问题匹配。可能需要调整工具描述，使模型更准确理解工具适用场景。

   • 结果解析失败：针对不同工具返回结果，仔细检查解析函数。例如，若搜索 API 返回结果结构变化，相应调整 parse_search_result 函数。

2. 增加新工具（文件写入）：

   • 工具函数定义：

     class WriteFileParams(BaseModel):
         file_path: str
         content: str
     
         @validator('file_path')
         def file_path_valid(cls, v):
             # 简单检查路径合法性，实际可更复杂
             if not v.strip():
                 raise ValueError('文件路径不能为空')
             return v
     
     
     def write_file(params: WriteFileParams):
         try:
             with open(params.file_path, 'w', encoding='utf - 8') as f:
                 f.write(params.content)
             return "文件写入成功"
         except Exception as e:
             print(f"文件写入失败: {e}")
             return None
     
     

   • 工具描述添加：在工具选择逻辑中，添加文件写入工具的描述。

     tools = [
         {
             "name": "calculate",
             "parameters": {
                 "type": "object",
                 "properties": {
                     "num1": {"type": "number"},
                     "num2": {"type": "number"},
                     "operation": {"type": "string"}
                 },
                 "required": ["num1", "num2", "operation"]
             },
             "description": "用于执行简单的数学运算"
         },
         {
             "name": "search",
             "parameters": {
                 "type": "object",
                 "properties": {
                     "query": {"type": "string"},
                     "api_key": {"type": "string"}
                 },
                 "required": ["query", "api_key"]
             },
             "description": "用于搜索信息"
         },
         {
             "name": "read_file",
             "parameters": {
                 "type": "object",
                 "properties": {
                     "file_path": {"type": "string"}
                 },
                 "required": ["file_path"]
             },
             "description": "用于读取本地文件内容"
         },
         {
             "name": "write_file",
             "parameters": {
                 "type": "object",
                 "properties": {
                     "file_path": {"type": "string"},
                     "content": {"type": "string"}
                 },
                 "required": ["file_path", "content"]
             },
             "description": "用于将内容写入本地文件"
         }
     ]
     
     

   • 结果解析与回答生成调整：添加文件写入工具的结果解析函数，并在 react_agent 函数中处理文件写入工具的调用和结果解析。

     def parse_write_file_result(result):
         if result:
             return result
         return "文件写入失败"
     
     
     def react_agent(user_input, api_key):
         tool_name, tool_params = choose_tool(user_input,api_key):
             if tool_name == 'calculate':
                 params = CalculatorParams(**tool_params)
                 result = calculate(params)
                 parsed_result = parse_calculator_result(result)
             elif tool_name =='search':
                 params = SearchParams(**tool_params)
                 result = search(params)
                 parsed_result = parse_search_result(result)
             elif tool_name =='read_file':
                 params = ReadFileParams(**tool_params)
                 result = read_file(params)
                 parsed_result = parse_read_file_result(result)
             elif tool_name == 'write_file':
                 params = WriteFileParams(**tool_params)
                 result = write_file(params)
                 parsed_result = parse_write_file_result(result)
             else:
                 return "无法确定要使用的工具"
             return generate_answer(parsed_result, api_key)
     
     

   • 优化后的完整代码：

import requests
import json
import os
from pydantic import BaseModel, validator


# 计算器工具
class CalculatorParams(BaseModel):
    num1: float
    num2: float
    operation: str

    @validator('operation')
    def valid_operation(cls, v):
        valid_ops = ['+', '-', '*', '/']
        if v not in valid_ops:
            raise ValueError('不支持的运算操作')
        return v


def calculate(params: CalculatorParams):
    if params.operation == '+':
        return params.num1 + params.num2
    elif params.operation == '-':
        return params.num1 - params.num2
    elif params.operation == '*':
        return params.num1 * params.num2
    elif params.operation == '/':
        if params.num2 == 0:
            print("除数不能为零")
            return None
        return params.num1 / params.num2


# 搜索工具
class SearchParams(BaseModel):
    query: str
    api_key: str

    @validator('query')
    def query_not_empty(cls, v):
        if not v.strip():
            raise ValueError('查询内容不能为空')
        return v

    @validator('api_key')
    def api_key_not_empty(cls, v):
        if not v.strip():
            raise ValueError('api_key不能为空')
        return v


def search(params: SearchParams):
    base_url = "https://api.example.com/search"
    params_dict = {
        "key": params.api_key,
        "q": params.query
    }
    try:
        response = requests.get(base_url, params=params_dict)
        response.raise_for_status()
        return response.json()
    except requests.RequestException as e:
        print(f"搜索失败: {e}")
        return None


# 文件读取工具
class ReadFileParams(BaseModel):
    file_path: str

    @validator('file_path')
    def file_path_exists(cls, v):
        if not os.path.exists(v):
            raise ValueError('文件路径不存在')
        return v


def read_file(params: ReadFileParams):
    try:
        with open(params.file_path, 'r', encoding='utf - 8') as f:
            return f.read()
    except Exception as e:
        print(f"读取文件失败: {e}")
        return None


# 文件写入工具
class WriteFileParams(BaseModel):
    file_path: str
    content: str

    @validator('file_path')
    def file_path_valid(cls, v):
        if not v.strip():
            raise ValueError('文件路径不能为空')
        return v


def write_file(params: WriteFileParams):
    try:
        with open(params.file_path, 'w', encoding='utf - 8') as f:
            f.write(params.content)
        return "文件写入成功"
    except Exception as e:
        print(f"文件写入失败: {e}")
        return None


# 模拟通义千问API调用（实际需替换为真实API调用）
def call_qwen_api(api_key, prompt):
    url = "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {api_key}"
    }
    data = {
        "model": "qwen-plus",
        "messages": [
            {"role": "user", "content": prompt}
        ],
        "stream": False
    }
    try:
        response = requests.post(url, headers=headers, json=data)
        response.raise_for_status()
        return response.json()
    except requests.RequestException as e:
        print(f"调用通义千问API失败: {e}")
        return None


def choose_tool(user_input, api_key):
    tools = [
        {
            "name": "calculate",
            "parameters": {
                "type": "object",
                "properties": {
                    "num1": {"type": "number"},
                    "num2": {"type": "number"},
                    "operation": {"type": "string"}
                },
                "required": ["num1", "num2", "operation"]
            },
            "description": "用于执行简单的数学运算"
        },
        {
            "name": "search",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string"},
                    "api_key": {"type": "string"}
                },
                "required": ["query", "api_key"]
            },
            "description": "用于搜索信息"
        },
        {
            "name": "read_file",
            "parameters": {
                "type": "object",
                "properties": {
                    "file_path": {"type": "string"}
                },
                "required": ["file_path"]
            },
            "description": "用于读取本地文件内容"
        },
        {
            "name": "write_file",
            "parameters": {
                "type": "object",
                "properties": {
                    "file_path": {"type": "string"},
                    "content": {"type": "string"}
                },
                "required": ["file_path", "content"]
            },
            "description": "用于将内容写入本地文件"
        }
    ]
    prompt = {
        "user_input": user_input,
        "functions": tools
    }
    response = call_qwen_api(api_key, json.dumps(prompt))
    if "function_call" in response:
        function_call = response["function_call"]
        return function_call["name"], function_call["parameters"]
    return None, None


def parse_calculator_result(result):
    if result is not None:
        return f"计算结果为{result}"
    return ""


def parse_search_result(result):
    if result:
        return f"搜索结果: {result.get('results', [])}"
    return ""


def parse_read_file_result(result):
    if result:
        return f"文件内容: {result}"
    return ""


def parse_write_file_result(result):
    if result:
        return result
    return "文件写入失败"


def generate_answer(parsed_result, api_key):
    prompt = f"根据以下信息生成回答: {parsed_result}"
    response = call_qwen_api(api_key, prompt)
    if response:
        return response["choices"][0]["message"]["content"]
    return "无法生成回答"


def react_agent(user_input, api_key):
    tool_name, tool_params = choose_tool(user_input, api_key)
    if tool_name == 'calculate':
        params = CalculatorParams(**tool_params)
        result = calculate(params)
        parsed_result = parse_calculator_result(result)
    elif tool_name =='search':
        params = SearchParams(**tool_params)
        result = search(params)
        parsed_result = parse_search_result(result)
    elif tool_name =='read_file':
        params = ReadFileParams(**tool_params)
        result = read_file(params)
        parsed_result = parse_read_file_result(result)
    elif tool_name == 'write_file':
        params = WriteFileParams(**tool_params)
        result = write_file(params)
        parsed_result = parse_write_file_result(result)
    else:
        return "无法确定要使用的工具"
    return generate_answer(parsed_result, api_key)


if __name__ == "__main__":
    api_key = "your_api_key"
    user_input = input("请输入你的问题: ")
    print(f"用户输入: {user_input}")
    tool_name, tool_params = choose_tool(user_input, api_key)
    print(f"选择的工具: {tool_name}")
    print(f"工具参数: {tool_params}")
    if tool_name == 'calculate':
        params = CalculatorParams(**tool_params)
        result = calculate(params)
        print(f"计算结果: {result}")
        parsed_result = parse_calculator_result(result)
    elif tool_name =='search':
        params = SearchParams(**tool_params)
        result = search(params)
        print(f"搜索结果: {result}")
        parsed_result = parse_search_result(result)
    elif tool_name =='read_file':
        params = ReadFileParams(**tool_params)
        result = read_file(params)
        print(f"文件读取结果: {result}")
        parsed_result = parse_read_file_result(result)
    elif tool_name == 'write_file':
        params = WriteFileParams(**tool_params)
        result = write_file(params)
        print(f"文件写入结果: {result}")
        parsed_result = parse_write_file_result(result)
    else:
        print("无法确定要使用的工具")
        exit()
    answer = generate_answer(parsed_result, api_key)
    print(f"生成的回答: {answer}")

总结

通过本周的学习，我们成功开发并优化了一个基于通义千问大模型的 ReAct Agent。从最初整合 Function Call、ReAct 范式等知识开发出支持多种工具的 Agent，到对其进行调试优化，包括修复工具选择和结果解析的问题，并增加新的文件写入工具，逐步提升了 Agent 的功能和稳定性。

在实际应用中，可根据具体需求进一步扩展工具种类，优化工具选择逻辑以及结果解析方式，以更好地满足用户多样化的需求。同时，记得将本周知识点笔记以及优化后的 Agent 代码整理保存，方便后续参考和改进。如果在学习过程中遇到问题，可回顾本周所学内容，或者查阅相关文档进行解决。