tags:/n - langchain

• 智能体
• Agent
• Tool Calling
• 面试
  aliases:/n - Agent
• 智能体开发

05 Agents 智能体

一、Agent 介绍

Agent = LLM + Tools + Memory，能够自主决策调用什么工具、如何执行任务。

Agent vs Chain

Agent 工作流程

用户提问 → LLM 决策 → 调用工具 → 获取结果 → LLM 继续决策 → ... → 最终回答

工具定义方式

二、Agent 添加本地工具

"""
Agent + 本地工具演示

运行方式：python 05_Agent本地工具.py
"""
from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool

# 定义工具
@tool
def get_weather(city: str, date: str) -> str:
    """查询指定城市和日期的天气"""
    weather_data = {"北京": "晴朗 25°C", "上海": "多云 28°C"}
    return weather_data.get(city, f"{city}天气数据暂无")

@tool
def calculate(expression: str) -> str:
    """计算数学表达式"""
    try:
        return f"计算结果: {eval(expression)}"
    except Exception as e:
        return f"计算错误: {e}"

# 创建 Agent
llm = ChatOpenAI(model="qwen-plus", ...)
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有用的助手"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

tools = [get_weather, calculate]
agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 运行
result = agent_executor.invoke({"input": "北京今天天气怎么样？"})
print(f"答案: {result['output']}")

三、Agent 添加 MCP 工具

"""
MCP Stdio 方式：本地 MCP Server

运行方式：python 05_MCP工具.py
"""
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate

async def main():
    # 连接 MCP Server
    client = MultiServerMCPClient({
        "weather": {
            "transport": "stdio",
            "command": "python",
            "args": ["./mcp_server.py"],
        }
    })
    
    tools = await client.get_tools()
    
    llm = ChatOpenAI(model="qwen-plus", ...)
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一个有用的助手"),
        ("human", "{input}"),
        ("placeholder", "{agent_scratchpad}"),
    ])
    
    agent = create_tool_calling_agent(llm, tools, prompt)
    agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
    
    result = agent_executor.invoke({"input": "帮我查天气"})
    print(result['output'])

四、Agent 添加记忆

"""
Agent + 记忆演示（多轮对话）

运行方式：python 05_Agent记忆.py
"""
from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage

llm = ChatOpenAI(model="qwen-plus", ...)

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有用的助手"),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_tool_calling_agent(llm, [], prompt)
agent_executor = AgentExecutor(agent=agent, verbose=True)

# 多轮对话
chat_history = []

# 第一轮
result1 = agent_executor.invoke({
    "input": "我叫小明，我是一个程序员",
    "chat_history": chat_history,
})
print(f"AI: {result1['output']}")

# 更新历史
chat_history.append(HumanMessage(content="我叫小明，我是一个程序员"))
chat_history.append(AIMessage(content=result1["output"]))

# 第二轮（Agent 记得用户信息）
result2 = agent_executor.invoke({
    "input": "你还记得我是谁吗？",
    "chat_history": chat_history,
})
print(f"AI: {result2['output']}")

五、Agent 中间件

5.1 摘要中间件

from langchain.agents.middleware import SummarizationMiddleware

# 当消息超过 10 条时自动压缩
summarization = SummarizationMiddleware(llm=llm, max_messages=10)
agent_executor = AgentExecutor(agent=agent, middleware=[summarization])

5.2 人工审核中间件

from langchain.agents.middleware import HumanInTheLoopMiddleware

# 调用 delete_data 前需要人工确认
human_middleware = HumanInTheLoopMiddleware(interrupt_tools=["delete_data"])
agent_executor = AgentExecutor(agent=agent, middleware=[human_middleware])

相关笔记

[[11-LangGraph/01-LangGraph概述与快速入门.md|LangGraph]] · [[01-LangChain概述]] · [[02-Model-IO与模型调用]] · [[03-提示词模板与Chains]] · [[04-RAG检索增强]]