从"回答问题"到"解决问题"，打造能自主使用工具的AI智能体

开篇回顾与本期聚焦

在前两篇文章中，我们已经掌握了LangChain的两大核心能力：

• Model I/O：与模型对话的"正确姿势"（链接地址）

• Retrieval：为模型注入外部知识（链接地址）

现在，我们的AI应用已经能够"知道很多事情"并"基于知识回答问题"。但这还不够——真正的智能不应该仅仅局限于知道什么，更在于能做什么。

想象一下下面的这些场景（如果AI能做，那是不是可以好好摸鱼了！~~）：

• 一个能帮你查天气、订会议、发邮件的个人助理

• 一个能自动分析数据并生成报告的分析师

• 一个能根据用户需求自主操作软件的数字员工

这就是Agents（智能体） 的用武之地。在本篇中，我们将探索LangChain最令人兴奋的模块，打造能够自主思考、使用工具、解决问题的AI智能体。

---

一、核心理论：Agents——从"工具人"到"思考者"

1. 什么是Agents？

Agents（智能体） 的核心思想是：让大语言模型成为决策大脑，能够自主选择和使用各种工具来完成任务。

与之前我们构建的"知识问答系统"不同，智能体不是简单地回答问题，而是执行任务。它能够：

• 分析问题：理解用户的目标和需求

• 制定计划：拆解任务步骤，决定使用哪些工具

• 执行行动：调用合适的工具获取信息或执行操作

• 观察结果：分析工具返回的结果，决定下一步行动

• 循环迭代：重复上述过程直到任务完成

Al Agents是基于LLM的能够自主理解、自主规划决策、执行复杂任务的智能体，Agents不是ChatGPT的升级版，它不仅告诉你“如何做”，更会帮你去做。若各种Copilot是副驾驶，那Agents就是主驾驶。

Agents = LLM + 规划技能 + 记忆 + 工具使用。本质上Agents是一个LLM的编排与执行系统：

2. Agents的核心组件

a) Tools（工具）

• 定义：任何可以被AI调用的功能模块

• 示例：搜索引擎、计算器、API接口、数据库查询、文件操作等

• 关键：每个工具都有清晰的描述，让模型知道何时使用它

b) Agent（智能体）

• 角色：决策引擎，基于当前状态决定下一步行动

• 类型：ReAct、Plan-and-Execute、Self-ask-with-search等

• 核心：遵循"思考-行动-观察"的循环

c) Agent Executor（执行器）

• 作用：管理智能体的执行流程，处理错误，防止无限循环

• 功能：限制最大步骤数、处理异常、管理中间状态

3. ReAct框架：智能体的"思考模式"

ReAct（Reasoning + Acting） 是最流行的智能体框架：

传统上，对 LLM 的研究将推理（如通过思想链 CoT）和行动（如生成动作规划）作为两个独立的领域。CoT 在其“内部思考”中可能产生事实幻觉或错误传播；而单纯的行动模型则缺乏对复杂任务的规划、分解和异常处理能力。

ReAct 的核心思想在于打破这种隔阂。它通过 Prompting 的方式，让 LLM 生成一种**交错（interleaved）**的轨迹，该轨迹包含三个关键元素：

• Reason（思考）：模型的内部语言推理过程。它不直接影响外部环境，而是用于分析问题、分解任务、制定或调整计划、处理异常情况，以及从观察中提取关键信息。

• Act（行动）：模型生成的与外部环境交互的具体指令。例如，调用一个搜索引擎 API、在一个虚拟环境中执行某个动作等。

• Observe（观察）：执行行动后从外部环境返回的结果。这些结果作为新的上下文，供模型在下一步进行思考。

这种”思考 → 行动 → 观察 → 思考…”的循环，模拟了人类解决问题时“边想边做、边做边想”的认知过程，形成了一个动态、可回溯的闭环。

总而言之，ReAct是一种结合了推理(Reasoning)和行动(Acting)的新型提示技术，旨在提升大模型(LLMs)在解决语言理解和交互式决策任务方面的性能。这项技术通过在模型生成回答的过程中，交替生成推理轨迹和特定任务的行动，从而实现两者之间的协同效应。具体来说，推理轨迹帮助模型诱导、跟踪和更新行动计划，处理异常情况；而行动则允许模型与外部知识源（如知识库或环境）接口，并从中获取额外信息。

---

二、项目实战：构建全能个人助理智能体

好，现在，让我们构建一个能够使用多种工具的智能个人助理来帮我们理解Agents。

1、项目目标

    创建一个能理解复杂需求、自主选择工具、完成多步骤任务的智能助理。

    技术栈：LangChain Agents + 自定义Tools + HTTP请求调用

    完整代码如下：

import requests
import json
import math
from datetime import datetime, timedelta
from typing import Dict, Any, List
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain.schema import BaseOutputParser
import re

# ==================== 全局配置 ====================
# 聚合平台 http://www.ufunai.cn 配置
BASE_URL = "https://api.ufunai.cn/v1"
API_KEY = "sk-xxxxx"  # 替换为你的API密钥
MODEL_NAME = "gpt-4"  # 指定要测试的模型名称


# 1. 自定义HTTP模型类（复用之前代码）
class HTTPChatModel:
    def __init__(self, base_url: str, api_key: str, model: str = "gpt-4"):
        self.base_url = base_url
        self.api_key = api_key
        self.model = model

    def invoke(self, messages: list) -> str:
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.api_key}"
        }

        payload = {
            "model": self.model,
            "messages": messages,
            "temperature": 0.1,
            "max_tokens": 1200
        }

        try:
            response = requests.post(
                f"{self.base_url}/chat/completions",
                headers=headers,
                json=payload,
                timeout=60
            )
            response.raise_for_status()
            result = response.json()
            return result["choices"][0]["message"]["content"]

        except requests.exceptions.RequestException as e:
            print(f"HTTP请求错误: {e}")
            if hasattr(e, 'response') and e.response is not None:
                print(f"响应内容: {e.response.text}")
            raise


# 2. 自定义输出解析器，用于解析智能体的思考过程
class AgentOutputParser(BaseOutputParser):
    def parse(self, text: str) -> Dict[str, Any]:
        """解析智能体的输出，提取思考、行动和输入

        优先策略：
        - 如果模型给出了 action + action_input，则优先返回这两个字段（用于触发工具调用）
        - 如果没有 action，则返回 final_answer（直接结束）
        - 如果模型同时给出 action 和 final_answer，也会返回 action，以便执行工具
        """
        thought_match = re.search(r"思考[:：]\s*(.*?)(?=\n行动[:：]|\n最终答案[:：]|\n行动输入[:：]|\Z)", text, re.DOTALL)
        # 使用更宽松的正则以兼容中文、大小写以及包含空格或符号的工具名
        action_match = re.search(r"行动[:：]\s*([^\n]+)", text)
        action_input_match = re.search(r"行动输入[:：]\s*(.*?)(?=\n行动[:：]|\n最终答案[:：]|\Z)", text, re.DOTALL)
        final_answer_match = re.search(r"最终答案[:：]\s*(.*)", text, re.DOTALL)

        result = {}
        if thought_match:
            result["thought"] = thought_match.group(1).strip()

        # 优先解析 action，如果存在则返回 action + action_input（即便同时有 final_answer）
        if action_match and action_input_match:
            # 捕获更完整的工具名并清理两端的标点
            action = action_match.group(1).strip().strip(' .。，,')
            action_input = action_input_match.group(1).strip()
            result["action"] = action
            result["action_input"] = action_input

            # 仍然保留 final_answer（如果模型同时提供）以便在工具返回后参考，但不提前结束
            if final_answer_match:
                result["final_answer"] = final_answer_match.group(1).strip()

            return result

        # 如果没有 action，则返回最终答案（若有）
        if final_answer_match:
            result["final_answer"] = final_answer_match.group(1).strip()
            return result

        return result


# 3. 定义各种工具
class Tool:
    def __init__(self, name: str, description: str, function):
        self.name = name
        self.description = description
        self.function = function

    def run(self, input_text: str) -> str:
        try:
            return self.function(input_text)
        except Exception as e:
            return f"工具执行错误: {str(e)}"


# 3.1 计算器工具
def calculator_tool(expression: str) -> str:
    """执行数学计算"""
    # 安全限制：只允许基本的数学运算
    allowed_chars = set('0123456789+-*/.() ')
    if not all(c in allowed_chars for c in expression):
        return "错误：表达式中包含不允许的字符"

    try:
        # 使用eval计算，但在生产环境中应该使用更安全的方式
        result = eval(expression)
        return f"计算结果: {result}"
    except Exception as e:
        return f"计算错误: {str(e)}"


# 3.2 时间工具
def time_tool(query: str) -> str:
    """获取当前时间信息"""
    now = datetime.now()
    if "明天" in query:
        tomorrow = now + timedelta(days=1)
        return f"明天是: {tomorrow.strftime('%Y年%m月%d日 %A')}"
    elif "昨天" in query:
        yesterday = now - timedelta(days=1)
        return f"昨天是: {yesterday.strftime('%Y年%m月%d日 %A')}"
    else:
        return f"当前时间: {now.strftime('%Y年%m月%d日 %H:%M:%S %A')}"


# 3.3 单位转换工具
def unit_converter_tool(input_text: str) -> str:
    """执行单位转换"""
    try:
        # 解析输入，格式如 "100 USD to CNY" 或 "10公里 to 英里"
        parts = input_text.lower().split()
        if len(parts) < 4 or parts[2] != "to":
            return "格式错误，请使用: '数值 原单位 to 目标单位'"

        value = float(parts[0])
        from_unit = parts[1]
        to_unit = parts[3]

        # 简单的转换率（实际应用中应该使用实时API）
        conversion_rates = {
            "usd_cny": 7.2,
            "cny_usd": 0.14,
            "km_mile": 0.6214,
            "mile_km": 1.6093,
            "kg_pound": 2.2046,
            "pound_kg": 0.4536,
            "c_f": lambda x: (x * 9 / 5) + 32,
            "f_c": lambda x: (x - 32) * 5 / 9
        }

        key = f"{from_unit}_{to_unit}"
        if key in conversion_rates:
            rate = conversion_rates[key]
            if callable(rate):
                result = rate(value)
            else:
                result = value * rate
            return f"转换结果: {value} {from_unit} = {result:.2f} {to_unit}"
        else:
            return f"不支持从 {from_unit} 到 {to_unit} 的转换"
    except Exception as e:
        return f"转换错误: {str(e)}"


# 3.4 模拟搜索工具
def search_tool(query: str) -> str:
    """模拟搜索引擎（实际应用中应该接入真实搜索API）"""
    # 这里模拟返回一些预设结果
    mock_results = {
        "天气": "根据模拟数据：今天晴天，温度25°C，湿度60%，东南风3级。",
        "新闻": "最新科技新闻：AI技术快速发展，多模态模型成为新趋势。",
        "股票": "模拟股票信息：AAPL $175.32 (+1.2%), GOOGL $145.67 (-0.5%)",
        "默认": f"已搜索: '{query}'。这是模拟搜索结果，实际应用中应接入真实搜索引擎API。"
    }

    for key in mock_results:
        if key in query:
            return mock_results[key]

    return mock_results["默认"]


# 4. 智能体执行器
class SimpleAgentExecutor:
    def __init__(self, model, tools: Dict[str, Tool], max_steps: int = 5):
        self.model = model
        self.tools = tools
        self.max_steps = max_steps
        self.parser = AgentOutputParser()

        # 构建工具描述字符串
        self.tool_descriptions = "\n".join([
            f"- {name}: {tool.description}"
            for name, tool in tools.items()
        ])

        # 智能体提示模板 - 在初始化时就完成格式化
        template_str = """
        你是一个智能个人助理，可以使用各种工具来帮助用户解决问题。

        你可以使用的工具：
        {tool_descriptions}

        请按照以下格式回应：
        思考: 首先分析用户的问题，决定是否需要使用工具以及使用哪个工具
        行动: 工具名称（如果需要使用工具）
        行动输入: 工具的输入参数（如果需要使用工具）
        最终答案: 直接回答用户的问题（如果不需要工具或已经通过工具获得足够信息）

        注意：
        1. 每次只能使用一个工具
        2. 根据工具返回的结果，决定下一步行动
        3. 当你认为已经获得足够信息来回答用户问题时，使用"最终答案:"

        当前对话：
        用户: {input}

        请开始回应：
        """
        self.prompt_template = ChatPromptTemplate.from_template(template_str)


    def run(self, user_input: str) -> str:
        """执行智能体任务"""
        conversation_history = []
        steps = 0

        print(f"用户: {user_input}")
        print("-" * 50)

        while steps < self.max_steps:
            # 构建当前对话上下文
            current_context = user_input
            if conversation_history:
                history_text = "\n".join([
                    f"步骤{i + 1}: {step['thought']} -> 结果: {step['result'][:100]}..."
                    for i, step in enumerate(conversation_history)
                ])
                current_context = f"{user_input}\n\n执行历史:\n{history_text}"

            # 调用模型获取智能体响应
            prompt = self.prompt_template.format(tool_descriptions=self.tool_descriptions, input=current_context)
            messages = [{"role": "user", "content": prompt}]
            response = self.model.invoke(messages)

            print(f"步骤 {steps + 1}:")
            print(f"AI响应:\n{response}")
            print("-" * 30)

            # 解析响应
            parsed = self.parser.parse(response)
            print(f"解析结果: {parsed}")

            # 优先处理 action（如果模型给出了工具调用指令），即使同时包含 final_answer 也不应提前返回
            if "action" in parsed and "action_input" in parsed:
                action = parsed["action"]
                action_input = parsed["action_input"]

                # 尝试不区分大小写匹配工具名
                action_key = None
                for key in self.tools:
                    if action.lower() == key.lower():
                        action_key = key
                        break

                # 如果仍未找到，尝试一些常见中文映射
                if action_key is None:
                    mapping = {
                        "计算器": "Calculator",
                        "计算": "Calculator",
                        "时间": "Time",
                        "时间查询": "Time",
                        "单位转换": "UnitConverter",
                        "换算": "UnitConverter",
                        "搜索": "Search",
                        "查找": "Search"
                    }
                    # 先精确匹配映射键，再尝试不区分大小写匹配映射值
                    if action in mapping:
                        action_key = mapping[action]
                    else:
                        for k, v in mapping.items():
                            if action.lower() == k.lower():
                                action_key = v
                                break

                if action_key and action_key in self.tools:
                    # 执行工具
                    print(f"使用工具: {action_key}, 输入: {action_input}")
                    tool_result = self.tools[action_key].run(action_input)
                    print(f"工具结果: {tool_result}")

                    # 记录执行历史
                    conversation_history.append({
                        "thought": parsed.get("thought", ""),
                        "action": action_key,
                        "action_input": action_input,
                        "result": tool_result
                    })

                    steps += 1
                    # 在使用工具后继续循环，让模型基于工具结果决定下一步
                    continue
                else:
                    print(f" 未知工具: {action}")
                    return f"错误: 未知工具 '{action}'"

            # 如果没有 action，则检查是否存在最终答案
            if "final_answer" in parsed:
                print(" 任务完成!")
                return parsed["final_answer"]

            # 如果既没有 action 也没有 final_answer，则响应格式不正确
            print(" 响应格式错误")
            return "错误: 智能体响应格式不正确"

        return "错误: 达到最大执行步骤数，任务未完成"


# 5. 主程序
def main():
    # 初始化模型
    chat_model = HTTPChatModel(
        base_url = BASE_URL,
        api_key = API_KEY,
        model = MODEL_NAME
    )

    # 初始化工具
    tools = {
        "Calculator": Tool(
            name="Calculator",
            description="执行数学计算，支持加减乘除和括号",
            function=calculator_tool
        ),
        "Time": Tool(
            name="Time",
            description="获取当前时间、日期信息，可以查询今天、明天、昨天",
            function=time_tool
        ),
        "UnitConverter": Tool(
            name="UnitConverter",
            description="执行单位转换，支持货币、长度、重量、温度等",
            function=unit_converter_tool
        ),
        "Search": Tool(
            name="Search",
            description="搜索信息，可以查询天气、新闻、股票等",
            function=search_tool
        )
    }

    # 初始化智能体执行器
    agent = SimpleAgentExecutor(chat_model, tools, max_steps=5)

    print("智能个人助理已启动!")
    print("支持功能: 计算、时间查询、单位转换、信息搜索")
    print("输入'退出'结束对话\n")

    # 测试用例
    test_cases = [
        "100美元可以兑换多少人民币？",
        "计算(15 + 23) * 4 - 18 / 3的结果",
        "今天天气怎么样？明天呢？",
        "10公里等于多少英里？25摄氏度是多少华氏度？"
    ]

    print("试试这些问题:")
    for i, case in enumerate(test_cases, 1):
        print(f"{i}. {case}")
    print()

    # 交互式对话
    while True:
        user_input = input("您的问题: ").strip()
        if user_input.lower() in ['退出', 'exit', 'quit']:
            break

        if not user_input:
            continue

        print("\n" + "=" * 50)
        result = agent.run(user_input)
        print(f"\n 最终回答: {result}")
        print("=" * 50 + "\n")


if __name__ == "__main__":
    main()

2、代码深度解读

1. 自定义输出解析器 (AgentOutputParser)

   • 解析智能体的思考过程，提取关键信息

   • 识别"思考"、"行动"、"最终答案"等关键部分

   • 处理智能体的结构化输出

2. 工具系统 (Tool类)

   • 统一工具接口，每个工具都有名称、描述和功能

   • 支持错误处理和异常管理

   • 易于扩展新的工具

3. 智能体执行引擎 (SimpleAgentExecutor)

   • 管理"思考-行动-观察"的完整循环

   • 维护对话历史和执行上下文

   • 防止无限循环和错误传播

4. 多种实用工具

   • 计算器：执行数学运算

   • 时间工具：获取时间信息

   • 单位转换：货币、长度、温度等转换

   • 搜索工具：模拟信息检索（可替换为真实API）

运行效果演示：

智能个人助理已启动!
支持功能: 计算、时间查询、单位转换、信息搜索
输入'退出'结束对话

试试这些问题:
1. 100美元可以兑换多少人民币？
2. 计算(15 + 23) * 4 - 18 / 3的结果
3. 今天天气怎么样？明天呢？
4. 10公里等于多少英里？25摄氏度是多少华氏度？

您的问题: 90美元可以兑换多少人民币, 另外今天天气怎么样

==================================================
用户: 90美元可以兑换多少人民币, 另外今天天气怎么样
--------------------------------------------------
步骤 1:
AI响应:
思考: 用户询问了两个问题，一个是货币兑换的问题，另一个是关于天气的问题。根据规则，每次只能使用一个工具，因此我将首先解决货币兑换的问题。

行动: UnitConverter
行动输入: 转换90美元到人民币
最终答案: 请稍等，我正在计算90美元兑换成人民币的金额。
------------------------------
解析结果: {'thought': '用户询问了两个问题，一个是货币兑换的问题，另一个是关于天气的问题。根据规则，每次只能使用一个工具，因此我将首先解决货币兑换的问题。', 'action': 'UnitConverter', 'action_input': '转换90美元到人民币', 'final_answer': '请稍等，我正在计算90美元兑换成人民币的金额。'}
使用工具: UnitConverter, 输入: 转换90美元到人民币
工具结果: 格式错误，请使用: '数值 原单位 to 目标单位'
步骤 2:
AI响应:
思考: 用户询问了两个问题，一个是货币兑换的问题，另一个是关于天气的问题。根据规则，每次只能使用一个工具，因此我将首先解决货币兑换的问题。
行动: UnitConverter
行动输入: 90 USD to CNY
最终答案: 请稍等，我正在帮您查询90美元兑换成人民币的金额。
------------------------------
解析结果: {'thought': '用户询问了两个问题，一个是货币兑换的问题，另一个是关于天气的问题。根据规则，每次只能使用一个工具，因此我将首先解决货币兑换的问题。', 'action': 'UnitConverter', 'action_input': '90 USD to CNY', 'final_answer': '请稍等，我正在帮您查询90美元兑换成人民币的金额。'}
使用工具: UnitConverter, 输入: 90 USD to CNY
工具结果: 转换结果: 90.0 usd = 648.00 cny
步骤 3:
AI响应:
思考: 用户的第二个问题是询问今天的天气情况，我将使用搜索工具来获取天气信息。
行动: Search
行动输入: 今天天气
最终答案: 90美元可以兑换648.00人民币。关于今天的天气，我需要使用搜索工具来获取具体信息，请稍等。
------------------------------
解析结果: {'thought': '用户的第二个问题是询问今天的天气情况，我将使用搜索工具来获取天气信息。', 'action': 'Search', 'action_input': '今天天气', 'final_answer': '90美元可以兑换648.00人民币。关于今天的天气，我需要使用搜索工具来获取具体信息，请稍等。'}
使用工具: Search, 输入: 今天天气
工具结果: 根据模拟数据：今天晴天，温度25°C，湿度60%，东南风3级。
步骤 4:
AI响应:
最终答案: 90美元可以兑换648人民币。今天的天气是晴天，温度25°C，湿度60%，东南风3级。
------------------------------
解析结果: {'final_answer': '90美元可以兑换648人民币。今天的天气是晴天，温度25°C，湿度60%，东南风3级。'}
 任务完成!

 最终回答: 90美元可以兑换648人民币。今天的天气是晴天，温度25°C，湿度60%，东南风3级。
==================================================

三、高级扩展：连接真实API

上面的搜索工具是模拟的，现在让我们将其升级为真实的API调用：

# 真实搜索工具（需要API密钥）
def real_search_tool(query: str) -> str:
    """使用真实搜索引擎API"""
    # 这里以SerpAPI为例（需要注册获取API密钥）
    try:
        params = {
            'q': query,
            'api_key': 'your-serpapi-key-here',  # 替换为你的API密钥
            'engine': 'google'
        }
        
        response = requests.get('https://serpapi.com/search', params=params)
        data = response.json()
        
        # 提取有机搜索结果
        if 'organic_results' in data and data['organic_results']:
            first_result = data['organic_results'][0]
            return f"搜索结果: {first_result.get('title', '')} - {first_result.get('snippet', '')}"
        else:
            return "未找到相关结果"
    except Exception as e:
        return f"搜索错误: {str(e)}"

# 天气API工具
def weather_tool(location: str) -> str:
    """使用真实天气API"""
    try:
        # 这里使用OpenWeatherMap API示例（需要注册获取API密钥）
        api_key = "your-weather-api-key"
        url = f"http://api.openweathermap.org/data/2.5/weather?q={location}&appid={api_key}&units=metric&lang=zh_cn"
        
        response = requests.get(url)
        data = response.json()
        
        if response.status_code == 200:
            temp = data['main']['temp']
            desc = data['weather'][0]['description']
            humidity = data['main']['humidity']
            return f"{location}天气: {desc}, 温度{temp}°C, 湿度{humidity}%"
        else:
            return f"无法获取{location}的天气信息"
    except Exception as e:
        return f"天气查询错误: {str(e)}"

四、总结与展望

通过本篇的学习，我们已经成功构建了一个能够自主思考、使用工具解决问题的AI智能体。

本篇核心收获：

• Agents理念：理解了智能体从"回答问题"到"解决问题"的范式转变

• ReAct框架：掌握了思考-行动-观察的智能体工作循环

• 工具系统：学会了如何创建和管理各种功能工具

• 执行引擎：理解了智能体执行器的设计原理和实现方式

生产环境建议：

• 添加工具使用权限控制和安全性检查

• 实现更复杂的错误处理和重试机制

• 添加执行步骤的监控和日志记录

• 考虑工具调用的成本控制和限流

LangChain系列总结：经过这三篇文章的学习，我们已经完整掌握了LangChain的核心模块：

1. Model I/O：与模型对话的标准方式

2. Retrieval：为模型注入外部知识

3. Agents：让模型自主使用工具解决问题

这三个模块构成了构建复杂AI应用的完整技术栈，从基础对话到知识增强，再到自主行动，层层递进，功能强大。

---

下一篇预告：

现在，我们的智能体已经能够使用工具解决问题，但在真实的人机交互中，我们期望的不仅仅是单次任务处理，而是连续的、有上下文的对话，是能够记住历史、理解语境的智能体验。

在《LangChain实战入门（四）：融合篇——打造有记忆、能协作的AI应用》中，我们将探索LangChain的Memory与Chains模块，打造能够记住对话历史、协调多个工具完成复杂任务的智能助理！你将学会如何为AI添加"记忆力"，构建真正理解上下文、能够长期协作的智能研究助手！

另：需要源码的同学，请关注微信公众号（优趣AI）在留言区评论获取！！！

创作不易，码字更不易，如果觉得这篇文章对你有帮助，记得点个关注、在看或收藏，给作者一点鼓励吧~