理解Agents

目前主流的 AI 工具（如 ChatGPT、文心一言、MidJourney）均处于2-3 级协作阶段。
在Chain中行动序列是 硬编码的、固定流程的 ，像是“线性流水线”，而Agent是一个通过动态协调 大语言模型（LLM） 和 工具（Tools） 来完成复杂任务的智能系统。它让LLM充当"决策大脑"，根据用户输入自主选择和执行工具（如搜索、计算、数据库查询等），最终生成精准的响应。
它可以根据任务动态决定：如何拆解任务、调用哪些工具、以什么顺序调用、如何利用中间结果推进任务。

Agent的组件

1）大模型(LLM)：作为大脑，提供推理、规划和知识理解能力。
比如：OpenaAI()、ChatOpenAI()
2）记忆(Memory)：具备短期记忆（上下文）和长期记忆（向量存储），支持快速知识检索。
比如：ConversationBufferMemory、ConversationSummaryMemory、ConversationBufferWindowMemory等
3）工具(Tools)：调用外部工具（如API、数据库）的执行单元
比如：SearchTool、CalculatorTool
因为大模型虽然非常强大，但是也具备一定的局限性。比如不能回答实时信息、处理复杂数学逻辑问题仍然非常初级等等。因此，可以借助第三方工具来辅助大模型的应用。
以MCP工具为例说明：
https://bailian.console.aliyun.com/?tab=mcp#/mcp-market
https://www.modelscope.cn/mcp
4）规划(Planning)：任务分解、反思与自省框架实现复杂任务处理
5）行动(Action)：实际执行决策的能力 （比如：检索、推理、编程 ）
6）协作：通过与其他智能体交互合作，完成更复杂的任务目标。

Agent与AgentExecutor

在 LangChain 中，Agent（智能体） 和 AgentExecutor（智能体执行器） 是协作完成 “自主任务闭环” 的两个核心组件 —— 前者负责 “思考决策”（如 “该用什么工具、怎么用”），后者负责 “流程调度”（如 “调用工具、传递参数、处理结果”）。二者的关系类似 “指挥官” 与 “参谋长”：Agent 决定 “做什么、怎么做”，AgentExecutor 确保 “按决策落地执行”。
首先需明确三者的依赖关系：Agent 绑定 LLM+Prompt（用于决策）；AgentExecutor 绑定 Agent+Tools+Memory（用于执行）；Tools 包含本次任务需用的工具（如WeatherTool）。
示例：

# 1. 定义工具（查天气）
def get_weather(city):
    return f"{city}当前天气：晴，25℃，风力3级"
weather_tool = Tool(
    name="WeatherTool",
    func=get_weather,
    description="查询指定城市的实时天气，需输入城市名"
)
tools = [weather_tool]

# 2. 初始化LLM和Memory
llm = ChatOpenAI(model="gpt-4o", temperature=0)
memory = ConversationBufferMemory(memory_key="chat_history")

# 3. 初始化Agent（决策层）：用initialize_agent简化创建
agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent=AgentType.CHAT_ZERO_SHOT_REACT_DESCRIPTION,  # 指定Agent类型（决策逻辑）
    memory=memory,
    verbose=True
)
executor = agent 
executor.invoke("查询上海天气")

# > Entering new AgentExecutor chain...
# Question: 查询上海天气
# Thought: I need to use the WeatherTool to find the current weather in Shanghai.
# Action:
# ```
# {
#   "action": "WeatherTool",
#   "action_input": "上海"
# }
# ```
# Observation: 上海当前天气：晴，25℃，风力3级
# Thought:I now know the final answer.
# 
# Final Answer: 上海当前天气：晴，25℃，风力3级。
# 
# > Finished chain.

Agents的核心类型

方式1：Funcation Call模式（函数调用）

• 基于 结构化函数调用 （如 OpenAI Function Calling）
• 直接生成工具调用参数（ JSON 格式 ）
• 效率更高，适合工具明确的场景

#第1种： 
AgentType.OPENAI_FUNCTIONS 
#第2种： 
AgentType.OPENAI_MULTI_FUNCTIONS

当任务仅需 “AI 调用单一工具 / 固定流程工具”，且无需复杂决策时，优先用 Function Call：

• 场景示例：
  a. 调用计算器计算 “3+5×2”（工具固定，参数明确）；
  b. 调用翻译工具将 “Hello” 译为中文（工具固定，需求明确）；
  c. 调用 API 获取指定股票的实时价格（工具固定，参数仅需股票代码）。
• 核心原因：任务简单，无需 AI “思考为什么选这个工具”，仅需按格式调用即可。

方式2：ReAct 模式（思考 - 行动）

• 基于 文本推理 的链式思考（Reasoning + Acting），具备反思和自我纠错能力。
  • 推理（Reasoning）：分析当前状态，决定下一步行动
  • 行动（Acting）：调用工具并返回结果
• 通过 自然语言描述决策过程

当任务需要 “AI 自主拆解步骤、选择工具、调整策略” 时，必须用 ReAct：

• 场景示例：
  a. 规划 “北京 3 日游”（需拆解 “查天气→选景点→订酒店→规划路线”，多工具协同，需根据天气调整景点）；
  b. 撰写 “2024 年 Q3 手机市场调研报告”（需拆解 “查数据→分析趋势→对比竞品→生成结论”，需判断数据是否足够，不足则补充查询）；
  c. 解决 “用户电脑无法联网”（需拆解 “检查网络连接→ping 网关→查 DNS→判断故障原因”，需根据每步结果调整下一步操作）。
• 核心原因：任务复杂且步骤不固定，需 AI 通过 “思考 - 行动 - 观察” 闭环自主决策，避免遗漏或错误。

从技术实现上，ReAct 模式并非 “自己直接调用工具”，而是通过 “Prompt 引导决策”+“AgentExecutor 调度执行”两步，将 “工具调用” 嵌入闭环：

1. 第一步：Prompt 引导 AI “主动想到调用工具”
   ReAct 模式的核心是 “用 Prompt 规范 AI 的思考逻辑”，其中会明确提示 AI：“当需要外部信息 / 执行操作时，必须调用工具”。
   例如，LangChain 中CHAT_ZERO_SHOT_REACT_DESCRIPTION Agent 的默认 Prompt 包含这样的引导：

“你可以使用以下工具来帮助你回答问题：{tools}。思考步骤：
分析问题，判断是否需要调用工具；
若需要，选择合适的工具，并传入正确的参数；
根据工具返回的结果，判断是否需要继续调用工具，或直接输出答案。”

这种 Prompt 会 “强制” AI 在思考时优先考虑 “是否用工具”—— 比如用户问 “今天上海股市收盘指数”，AI 会根据 Prompt 引导，意识到 “这是实时数据，必须调用股票查询工具”，而不是直接输出旧数据。

2. 第二步：AgentExecutor 负责 “执行工具调用，反馈结果”
   ReAct 模式的 AI（决策层）只负责 “输出‘调用哪个工具、传什么参数’的指令”，不直接操作工具；AgentExecutor（执行层） 会接收指令，完成工具调用的落地：

• 解析指令：从 AI 的 “行动” 中提取 “工具名”（如 “StockQueryTool”）和 “参数”（如 “上海，2024-10-20”）；
• 调用工具：找到对应的工具实例，传入参数执行（如调用股票 API 获取收盘指数）；
• 反馈结果：将工具返回的结果（如 “上海综指收盘 3050 点”）整理后，回传给 AI，供 AI 进行 “再思考”。

例如：AI（ReAct 决策层）输出指令{“action”:“StockQueryTool”,“action_input”:“上海，2024-10-20”}，AgentExecutor 会：
3. 找到StockQueryTool实例；
4. 调用StockQueryTool.run(“上海，2024-10-20”)；
5. 将结果 “上海综指收盘 3050 点” 返回给 AI；
6. AI 基于这个结果，判断 “无需再调用工具，可直接输出答案”。

ReAct + Function Call = 强大 Agent

在实际开发中，两者并非互斥，而是常结合使用：

• ReAct 提供 “决策逻辑”：AI 通过 “思考 - 行动 - 观察” 闭环，确定 “什么时候调用工具、调用哪个工具”；
• Function Call 提供 “技术实现”：AI 按 Function Call 的结构化格式，生成工具可解析的指令，确保调用成功。

例如，在 LangChain 的CHAT_ZERO_SHOT_REACT_DESCRIPTION Agent 中：

• ReAct 是 “决策框架”：Agent 按 “思考→行动→观察” 流程决策；
• Function Call 是 “行动指令格式”：Agent 输出的Action部分，本质是符合 Function Call 规范的结构化指令（如{“action”:“工具名”,“action_input”:“参数”}）。

AgentExecutor创建方式

传统方式：initialize_agent()

特点：

• 内置一些标准化模板（如 ZERO_SHOT_REACT_DESCRIPTION ）
• Agent的创建：使用AgentType

优点：快速上手（3行代码完成配置）
缺点：定制化能力较弱（如提示词固定）

from langchain.agents import initialize_agent
#第1步：创建AgentExecutor
agent_executor = initialize_agent(
llm=llm,
agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
tools=[search_tool],
verbose=True
)
#第2步：执行
agent_executor.invoke({"xxxx"})

通用方式：AgentExecutor构造方法

特点：

• Agent的创建：使用create_xxx_agent

优点：

• 可自定义提示词（如从远程hub获取或本地自定义）
• 清晰分离Agent逻辑与执行逻辑

缺点：

• 需要更多代码
• 需理解底层组件关系

prompt = hub.pull("hwchase17/react")
tools = [search_tool]
#第1步：创建Agent实例
agent = create_react_agent(
    llm=llm,
    prompt=prompt,
    tools=tools
)
#第2步：创建AgentExecutor实例
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools
)
#第3步：执行
agent_executor.invoke({"input":"xxxxx"})

AgentType

LangChain与Agent示例：

import datetime
import os
import random
import dotenv
from langchain.agents import initialize_agent, AgentType
from langchain.memory import ConversationBufferMemory
from langchain_core.tools import Tool
from langchain_openai import ChatOpenAI
dotenv.load_dotenv()
os.environ['OPENAI_API_KEY'] = os.getenv("OPENAI_API_KEY")
os.environ['OPENAI_BASE_URL'] = os.getenv("OPENAI_BASE_URL")
# 自定义工具1：查询当前时间
def get_current_time(query: str) -> str:
    """获取当前的日期和时间"""
    now = datetime.datetime.now()
    return f"当前时间是：{now.strftime('%Y年%m月%d日 %H:%M:%S')}"

# 自定义工具2：查询天气（模拟）
def get_weather(query: str) -> str:
    """查询指定城市的天气情况，输入格式应为"城市名" """
    # 模拟一些天气数据
    weather_conditions = ["晴朗", "多云", "小雨", "中雨", "阴天"]
    temperatures = [18, 19, 20, 21, 22, 23, 24, 25, 26, 27]

    # 提取城市名（简单处理）
    city = query.strip()
    if not city:
        return "请提供要查询天气的城市名"

    # 随机生成天气数据（实际应用中会调用真实API）
    condition = random.choice(weather_conditions)
    temperature = random.choice(temperatures)
    return f"{city}当前天气：{condition}，气温{temperature}°C"

# 定义工具集合
tools = [
    Tool(
        name="CurrentTime",
        func=get_current_time,
        description="用于获取当前的日期和时间，当用户问现在几点或今天是什么日期时使用"
    ),
    Tool(
        name="WeatherQuery",
        func=get_weather,
        description="用于查询指定城市的天气情况，需要用户提供城市名作为参数"
    )
]

llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 初始化内存，用于多轮对话
memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)

# 初始化Agent
agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.CHAT_CONVERSATIONAL_REACT_DESCRIPTION,
    memory=memory,
    verbose=True  # 输出详细的思考过程
)

# 测试Agent
response = agent.invoke("现在几点了？")
print(response["output"])

print("\n")
response = agent.invoke("北京的天气怎么样？")
print(response["output"])

print("\n")
response = agent.invoke("那上海呢？")
print(response["output"])

print("\n")
response = agent.invoke("还记得我刚才问了哪些城市的天气吗？")
print(response["output"])

================输出：

> Entering new AgentExecutor chain...
  agent = initialize_agent(

{
    "action": "CurrentTime",
    "action_input": ""
}

Observation: 当前时间是：2025年10月09日 23:32:22
Thought:
{
    "action": "Final Answer",
    "action_input": "当前时间是：2025年10月09日 23:32:22"
}

> Finished chain.
当前时间是：2025年10月09日 23:32:22

> Entering new AgentExecutor chain...

{
    "action": "WeatherQuery",
    "action_input": "北京"
}

Observation: 北京当前天气：晴朗，气温21°C
Thought:
{
    "action": "Final Answer",
    "action_input": "北京当前天气是晴朗，气温为21°C。"
}

> Finished chain.
北京当前天气是晴朗，气温为21°C。

> Entering new AgentExecutor chain...

{
    "action": "WeatherQuery",
    "action_input": "上海"
}

Observation: 上海当前天气：小雨，气温19°C
Thought:```json
{
    "action": "Final Answer",
    "action_input": "上海当前天气是小雨，气温19°C。"
}

> Finished chain.
上海当前天气是小雨，气温19°C。
> Entering new AgentExecutor chain...

{
    "action": "Final Answer",
    "action_input": "你刚才问了北京和上海的天气。"
}

> Finished chain.
你刚才问了北京和上海的天气。

LangChain与MCP结合：

1.先写一个mcp-server，在本地运行

from mcp.server.fastmcp import FastMCP
import logging

# 配置日志记录器
logging.basicConfig(
    level=logging.INFO,  # 设置日志级别为 INFO
    format="%(asctime)s - %(levelname)s - %(message)s"  # 日志格式
)
logger = logging.getLogger(__name__)

mcp = FastMCP("Weather")

@mcp.tool()
async def get_weather(location: str) -> str:
    """Get weather for location."""
    logger.info("The get_weather method is called: location=%s", location)
    return "天气阳光明媚，晴空万里。"

if __name__ == "__main__":
    logger.info("Start weather server through MCP")  # 记录服务启动日志
    mcp.run(transport="sse")

2.Agent与MCP-Client：

import asyncio
import os
import dotenv
from langchain.agents import initialize_agent, AgentType
from langchain.memory import ConversationBufferMemory
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_openai import ChatOpenAI

dotenv.load_dotenv()
os.environ['OPENAI_API_KEY'] = os.getenv("OPENAI_API_KEY")
os.environ['OPENAI_BASE_URL'] = os.getenv("OPENAI_BASE_URL")

# 初始化大模型
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 定义异步主函数
async def main():
    # MultiServerMCPClient 是用于连接多个 MCP 服务器的客户端。
    client = MultiServerMCPClient(
        {
            "weather": {
                "url": "http://localhost:8000/sse",
                "transport": "sse",
            }
        }
    )

    try:
        # 获取工具
        tools = await client.get_tools()
        print(f"已获取 {len(tools)} 个工具:")
        # 初始化内存，用于多轮对话
        memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)

        # 创建一个智能代理，思考 → 行动 → 观察 → 思考 → 行动 → ... → 最终答案
        agent = initialize_agent(
            tools,
            llm,
            agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,  # 支持结构化输入
            memory=memory,
            verbose=True,
            handle_parsing_errors=True
        )

        response = await agent.ainvoke({"input": "北京的天气怎么样？"})
        print(f"助手: {response['output']}")


    finally:
        # 清理资源
        if hasattr(client, 'close'):
            await client.close()

# 使用 asyncio 运行异步主函数
if __name__ == "__main__":
    asyncio.run(main())

输出：

已获取 1 个工具:

> Entering new AgentExecutor chain...
/Users/jiangjiang/Desktop/study/demo/langchain/mcp/client.py:37: LangChainDeprecationWarning: Please see the migration guide at: https://python.langchain.com/docs/versions/migrating_memory/
  memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)
/Users/jiangjiang/Desktop/study/demo/langchain/mcp/client.py:40: LangChainDeprecationWarning: LangChain agents will continue to be supported, but it is recommended for new use cases to be built with LangGraph. LangGraph offers a more flexible and full-featured framework for building agents, including support for tool-calling, persistence of state, and human-in-the-loop workflows. For details, refer to the `LangGraph documentation <https://langchain-ai.github.io/langgraph/>`_ as well as guides for `Migrating from AgentExecutor <https://python.langchain.com/docs/how_to/migrate_agent/>`_ and LangGraph's `Pre-built ReAct agent <https://langchain-ai.github.io/langgraph/how-tos/create-react-agent/>`_.
  agent = initialize_agent(
Action:

{
  "action": "get_weather",
  "action_input": {
    "location": "北京"
  }
}

Observation: 天气阳光明媚，晴空万里。
Thought:
Action:

{
  "action": "Final Answer",
  "action_input": "北京的天气阳光明媚，晴空万里。"
}

> Finished chain.
助手: 北京的天气阳光明媚，晴空万里。