大模型学习笔记：Agent智能体

Leave a Comment / Passerby, Uncategorized /

概念：

智能体（Agent）是一种能够自主规划、决策、执行任务的组件，核心是让大预言模型（LLM）根据任务需求，选择并调用工具，完成单靠模型自身无法解决的复杂问题。

  • 没有Agent时，LLM只能基于自身训练数据回答问题，遇到需要实时数据、复杂计算、外部工具调用的场景就会卡壳。
  • 有了Agent后，LLM就像一个“指挥官”，能思考任务步骤–>选择合适工具–>执行工具调用–>根据结果调整策略，直到完成任务。

核心特点：

  • 目标驱动：围绕用户的具体任务目标展开工作
  • 工具调用能力：能连接外部工具，弥补LLM的局限性
  • 自主决策与迭代：不需要人工干预，能根据工具返回的结果，判断是否需要继续调用工具，或直接生成最终答案

核心区别：

  • 普通Chain：
    • 执行流程固定，按预设步骤运行
    • 工具调用路径写死在代码里
    • 适合简单，标准化任务
  • Agent：
    • 执行流程动态，根据任务和结果自主调整
    • 工具选择由LLM思考决定
    • 适合复杂、多步骤、需要决策的步骤

Agent智能体 = 大语言模型（大脑）+ 工具箱（手脚）+ 决策逻辑（思维），让大模型拥有了感知外部并影响现实的能力

Agent初体验

 

from langchain.agents import create_agent
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.tools import tool

@tool(description="获取天气信息的工具")
def get_weather(location):
    # 这里可以调用天气API获取天气信息，暂时返回一个模拟的结果
    return f"{location}的天气是极端狂风暴雨天气，超级强度雷电以及超大降雨量"

agent = create_agent(
    model=ChatTongyi(model="qwen3-max"),
    tools=[get_weather],
    system_prompt="你是我的人工智能助手，协助我解答问题和提供信息。请根据我的提问，尽力给出准确和有用的回答。如果你不确定答案，可以说不知道。"
)

res = agent.invoke(
    {
        "messages": [
            {"role": "user", "content": "北京天气如何？"},
        ]
    }
)

for msg in res["messages"]:
    print(type(msg).__name__, msg.content)

Agent的stream

通过create_agent方法创建的Agent对象，也是Runnable接口子类实现，也有invoke一次性得到完整结果，stream流式得到结果：

from langchain.agents import create_agent
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.tools import tool

@tool(description="获取股票价格的工具,输入股票名称，输出股票字符串价格")
def get_price(name: str)-> str:
    # 这里可以调用价格API获取价格信息，暂时返回一个模拟的结果
    return f"股票{name}的价格是5元"

@tool(description="获取股票信息的工具，输入股票名称，输出股票字符串信息")
def get_info(name: str) -> str:
    # 这里可以调用信息API获取信息，暂时返回一个模拟的结果
    return f"股票{name}是一个A股上市公司"

agent = create_agent(
    model=ChatTongyi(model="qwen3-max"),
    tools=[get_price, get_info],
    system_prompt="你是我的人工智能助手，协助我解答股票相关问题和提供信息。请告诉我思考过程，为何调用某个工具。如果你不确定答案，可以说不知道。"
)

for chunk in agent.stream(
    {
        "messages": [
            {"role": "user", "content": "请告诉我股票OpenAI的价格和信息？"}
        ]
    },
    stream_mode = "values"
):
    latest_msg = chunk["messages"][-1]
    if latest_msg.content:
        print(type(latest_msg).__name__, latest_msg.content)
    try:
        if latest_msg.tool_calls:
            print("工具调用信息：", [{ "tool_name": tc.tool_name, "arguments": tc.arguments } for tc in latest_msg.tool_calls])
    except Exception as e:
        pass

ReAct

Agent ReAct是大模型智能体的核心思考与行动框架，全称Reasoning + Acting （推理 + 行动），是让Agent像人类一样 思考问题–> 制定策略 –> 执行行动 –>验证结果 的关键逻辑。
典型的ReAct范式的Agent：

思考Reasoning： 分析问题，判断现有信息是否足够，明确下一步，即模型决策是否需要调用外部工具获取更多信息来用回答；
行动Action：执行思考阶段制定的策略，即基于模型决策结果，调用工具获取信息；
观察Observation：获取行动的结果，提取有效信息，即获取工具返回值即判断工具是否正常工作为下一轮思考提供信息。

思考–> 行动 –> 观察 –> 思考 –> … –> … –>结束

LangChain的Agent对象，就说按ReAct模式运行：

from langchain.agents import create_agent
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.tools import tool

@tool(description="获取体重，返回值时整数，单位千克")
def get_weight(input: str) -> int:
    # 这里可以调用体重API获取体重信息，暂时返回一个模拟的结果
    return 57

@tool(description="获取身高，返回值是整数，单位厘米")
def get_height(input: str) -> int:
    # 这里可以调用身高API获取身高信息，暂时返回一个模拟的结果
    return 165

agent = create_agent(
    model=ChatTongyi(model="qwen3-max"),
    tools=[get_weight, get_height],
    system_prompt="你是严格遵循ReAct框架的智能体，必须按 思考-->行动-->观察 --> 再思考 的流程解决问题。且每轮仅能思考并调用1个工具，禁止单词调用多个工具。并告知我你的思考过程，工具的调用原因，按思考、行动、观察三个结构告知我 "
)

for chunk in agent.stream(
    {
        "messages": [
            {"role": "user", "content": "计算我的BMI"}
        ]
    },
    stream_mode = "values"
):

    latest_msg = chunk["messages"][-1]
    if latest_msg.content:
        print(type(latest_msg).__name__, latest_msg.content)
    try:
        if latest_msg.tool_calls:
            print("工具调用信息：", [{ "tool_name": tc.tool_name, "arguments": tc.arguments } for tc in latest_msg.tool_calls])
    except Exception as e:
        pass

middleware中间件

中间件的作用是对智能体的每一步工作进行控制和自定义的执行。作用场景：

  • 日志记录、分析、调试
  • 转换提示词、工具选择
  • 重试、备用、提前终止等逻辑控制
  • 安全防护、个人身份检测等

无中间件：

有中间件：

示例：

from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import before_agent, after_agent, before_model, after_model, wrap_model_call, wrap_tool_call
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.tools import tool
from langgraph.runtime import Runtime

@tool(description="获取天气信息的工具, 传入城市名称字符串，返回该城市的天气信息")
def get_weather(location):
    # 这里可以调用天气API获取天气信息，暂时返回一个模拟的结果
    return f"{location}的天气是极端狂风暴雨天气，超级强度雷电以及超大降雨量"

"""
1. agtent执行前
2. agent执行后
3. 模型执行前
4. 模型执行后
5. 工具执行中
6. 模型执行中
"""

@before_agent
def log_before_agent(state: AgentState, runtime: Runtime)-> None:
    # agent执行前会调用这个函数并传入state和runtime对象
    print(f"[Before Agent] agent启动，并附带{len(state['messages'])}条消息")

@after_agent
def log_after_agent(state: AgentState, runtime: Runtime)-> None:
    # agent执行后会调用这个函数并传入state和runtime对象
    print(f"[After Agent] agent结束，并附带{len(state['messages'])}条消息")

@before_model
def log_before_model(state: AgentState, runtime: Runtime)-> None:
    # 模型执行前会调用这个函数并传入state和runtime对象
    print(f"[Before Model] 模型即将执行，当前消息数{len(state['messages'])}")

@after_model
def log_after_model(state: AgentState, runtime: Runtime)-> None:
    # 模型执行后会调用这个函数并传入state和runtime对象
    print(f"[After Model] 模型执行结束，当前消息数{len(state['messages'])}")

@wrap_model_call  
def model_call_hook(request, handler):
    print("[Wrap Model Call]")
    return handler(request)

@wrap_tool_call
def monitor_tool(request, handler):
    print(f"[Wrap Tool Call]工具执行: {request.tool_call['name']}")
    print(f"[Wrap Tool Call]传入参数: {request.tool_call['args']}")
    return handler(request)

agent = create_agent(
    model=ChatTongyi(model="qwen3-max"),
    tools=[get_weather],
    system_prompt="你是我的人工智能助手，协助我解答问题和提供信息。请根据我的提问，尽力给出准确和有用的回答。如果你不确定答案，可以说不知道。",
    middleware=[log_before_agent, log_after_agent, log_before_model, log_after_model, model_call_hook, monitor_tool]
)

res = agent.invoke(
    {
        "messages": [
            {"role": "user", "content": "上海天气如何？外出散步有什么注意事项"},
        ]
    }
)

for msg in res["messages"]:
    print(type(msg).__name__, msg.content)
```

 

 

 

 

 

Leave a Comment

Your email address will not be published. Required fields are marked *

This site uses Akismet to reduce spam. Learn how your comment data is processed.