为什么需要 Agent？

大语言模型（LLM）很强大，但它有一个致命短板：只能根据训练数据回答，无法与外部世界交互。你问它"今天深圳天气怎么样？"，它只能编一个答案——因为它没有调用天气 API 的能力。

Agent（智能体）就是来解决这个问题的。简单来说：

Agent = LLM + 工具（Tools）+ 决策循环

LLM 不再只是"生成文本"，而是变成了一个大脑——它可以根据用户意图，自主决定调用哪个工具、以什么参数调用、如何根据工具返回结果继续推理。

---

核心原理：ReAct 模式

LangChain 的 Agent 底层采用的是 ReAct（Reasoning + Acting） 范式，它的工作流程是一个循环：

用户问题 → 思考(Thought) → 行动(Action) → 观察(Observation) → 思考 → ... → 最终回答

具体来说，每一步 Agent 会输出一个结构：

Thought: 我需要知道深圳今天的天气，所以应该调用天气查询工具
Action: get_weather
Action Input: {"city": "深圳"}
Observation: 深圳今天晴，26°C~32°C
Thought: 我已经拿到天气信息，可以回答用户了
Final Answer: 深圳今天天气晴朗，温度在26°C到32°C之间

这个循环会持续进行，直到 Agent 认为信息足够给出最终答案。

为什么 ReAct 有效？

1. 思考链外化：Thought 让 LLM 把推理过程写出来，大幅提升多步推理的准确率
2. 工具解耦：工具实现与 LLM 完全分离，新增能力不需要重新训练模型
3. 自我纠错：如果某次工具调用出错，Agent 可以在下一轮 Thought 中调整策略

---

LangChain Agent 架构速览

┌──────────────┐     ┌────────────────┐     ┌──────────────┐
│   User Input  │────▶│  AgentExecutor  │────▶│   LLM (大脑)  │
└──────────────┘     └────────────────┘     └──────────────┘
                            │                        │
                            │  ┌─────────────────┐   │
                            └─▶│  Tools (工具箱)   │◀──┘
                               │  - 搜索          │
                               │  - 计算          │
                               │  - API 调用      │
                               └─────────────────┘

核心组件：

组件	作用
LLM	推理引擎，决定下一步做什么
Tools	Agent 可调用的外部能力（函数/API）
Agent	决定"使用哪个工具、什么时候用、怎么用"的策略
AgentExecutor	运行 Agent 循环，管理 Thought→Action→Observation 的迭代

---

实战：构建一个智能运维助手

我们用 LangChain 构建一个智能运维助手，它可以：

• 查询服务器状态（模拟 API）
• 执行 SQL 查询数据库
• 进行数学计算
• 搜索知识库文档

这个场景直接对应日常工作中"排查线上问题"的流程。

1. 环境准备

pip install langchain langchain-openai

2. 定义工具（Tools）

from langchain_core.tools import tool
import json
import random
from datetime import datetime


# ============ 工具 1：查询服务器状态 ============
@tool
def check_server_status(server_name: str) -> str:
    """查询指定服务器的运行状态，返回 CPU、内存、磁盘使用率。
    
    Args:
        server_name: 服务器名称，如 'web-01', 'db-master'
    """
    # 模拟数据（实际项目中替换为真实 API 调用）
    servers = {
        "web-01": {"cpu": 45, "memory": 62, "disk": 38, "status": "healthy"},
        "web-02": {"cpu": 78, "memory": 85, "disk": 42, "status": "warning"},
        "db-master": {"cpu": 23, "memory": 55, "disk": 72, "status": "healthy"},
        "cache-01": {"cpu": 12, "memory": 30, "disk": 25, "status": "healthy"},
    }
    
    if server_name not in servers:
        return f"未找到服务器 {server_name}，可用服务器：{', '.join(servers.keys())}"
    
    info = servers[server_name]
    return json.dumps({
        "server": server_name,
        "cpu_percent": info["cpu"],
        "memory_percent": info["memory"],
        "disk_percent": info["disk"],
        "status": info["status"],
        "check_time": datetime.now().isoformat()
    }, ensure_ascii=False)


# ============ 工具 2：执行 SQL 查询（只读）============
@tool
def query_database(sql: str) -> str:
    """执行只读 SQL 查询，用于排查数据问题。仅支持 SELECT 语句。
    
    Args:
        sql: SQL 查询语句（仅 SELECT）
    """
    # 安全检查
    sql_upper = sql.strip().upper()
    if not sql_upper.startswith("SELECT"):
        return "错误：仅允许执行 SELECT 查询"
    
    # 模拟数据库（实际项目替换为真实数据库连接）
    mock_db = {
        "orders": [
            {"id": 1, "user": "张三", "amount": 299.0, "status": "paid", "created_at": "2026-08-03 10:30"},
            {"id": 2, "user": "李四", "amount": 599.0, "status": "pending", "created_at": "2026-08-03 11:00"},
            {"id": 3, "user": "王五", "amount": 199.0, "status": "failed", "created_at": "2026-08-03 11:15"},
        ]
    }
    
    # 简单解析：如果包含 orders，返回订单数据
    if "orders" in sql.lower():
        return json.dumps(mock_db["orders"], ensure_ascii=False, indent=2)
    
    return json.dumps({"message": "查询执行成功，返回 0 行"}, ensure_ascii=False)


# ============ 工具 3：搜索运维知识库 ============
@tool
def search_knowledge_base(query: str) -> str:
    """在运维知识库中搜索相关文档，用于查找故障处理方案。
    
    Args:
        query: 搜索关键词，如 'CPU过高', 'MySQL连接池耗尽'
    """
    # 模拟知识库（实际项目替换为向量数据库或文档检索）
    knowledge = {
        "cpu过高": {
            "原因": ["死循环进程", "GC 频繁", "并发请求突增"],
            "排查步骤": [
                "1. top 命令找到高 CPU 进程",
                "2. jstack 查看 Java 线程堆栈（Java 应用）",
                "3. 检查是否有慢查询导致数据库 CPU 飙升",
                "4. 查看应用日志的 QPS 是否异常"
            ],
            "常见解决方案": [
                "重启异常服务",
                "限流熔断",
                "扩容机器",
                "优化慢 SQL"
            ]
        },
        "mysql连接池耗尽": {
            "原因": ["慢查询阻塞", "连接未释放", "连接池配置过小"],
            "排查步骤": [
                "1. SHOW PROCESSLIST 查看当前连接",
                "2. SHOW FULL PROCESSLIST 查看具体 SQL",
                "3. 检查应用日志有无连接超时报错"
            ],
            "常见解决方案": [
                "KILL 长时间运行的慢查询",
                "增大连接池 maxActive 配置",
                "检查代码中的连接关闭逻辑"
            ]
        }
    }
    
    # 模糊匹配
    for key, value in knowledge.items():
        if key in query or any(kw in query for kw in query.split()):
            return json.dumps({key: value}, ensure_ascii=False, indent=2)
    
    return f"未找到与「{query}」相关的知识库文档"


# ============ 工具 4：发送告警通知 ============
@tool
def send_alert(level: str, title: str, message: str) -> str:
    """发送告警通知到运维群。
    
    Args:
        level: 告警级别，可选 'info', 'warning', 'critical'
        title: 告警标题
        message: 告警详细内容
    """
    valid_levels = ["info", "warning", "critical"]
    if level not in valid_levels:
        return f"无效告警级别，可选：{', '.join(valid_levels)}"
    
    # 实际项目替换为企业微信/钉钉/飞书 API
    return json.dumps({
        "status": "sent",
        "level": level,
        "title": title,
        "message": message,
        "channel": "企业微信-运维群",
        "timestamp": datetime.now().isoformat()
    }, ensure_ascii=False)


# 注册所有工具
tools = [
    check_server_status,
    query_database,
    search_knowledge_base,
    send_alert
]

3. 创建 Agent

from langchain_openai import ChatOpenAI
from langgraph.preprocessing import ToolNode
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.preprocessing import ToolNode
from langgraph.checkpoint.memory import MemorySaver

import os

# 初始化 LLM（使用 DeepSeek，也可以换成 OpenAI 或其他模型）
llm = ChatOpenAI(
    model="deepseek-chat",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com",
    temperature=0.3  # 运维场景需要稳重，温度调低
)

# 将工具注入 LLM
llm_with_tools = llm.bind_tools(tools)


# ============ 构建 LangGraph Agent（现代写法）============
from langgraph.graph import StateGraph, MessagesState, START, END

def call_model(state: MessagesState):
    """调用 LLM，由 LLM 决定是否调用工具"""
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}


def should_continue(state: MessagesState):
    """判断 Agent 循环是否应该继续（还有工具调用）还是结束"""
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"  # 还有工具要调用，继续
    return "end"  # 推理完成，输出最终答案


# 构建图
workflow = StateGraph(MessagesState)

# 添加节点
workflow.add_node("agent", call_model)
workflow.add_node("tools", ToolNode(tools))

# 添加边：控制流
workflow.add_edge(START, "agent")
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {"tools": "tools", "end": END}
)
workflow.add_edge("tools", "agent")  # 工具执行后回到 agent 继续推理

# 编译 Agent
agent = workflow.compile()

4. Agent 循环可视化

上面这个图的执行流程：

        START
          │
          ▼
     ┌─────────┐     tool_calls 存在?     ┌────────┐
     │  agent  │──────────────────────────▶│ tools  │
     │ (LLM)   │◀──────────────────────────│(执行)   │
     └─────────┘     回到 agent 继续推理    └────────┘
          │
          │ tool_calls 为空 (推理完成)
          ▼
         END

这就是 Agent 循环的本质：LLM 不断思考-行动-观察，直到能给出最终答案。

5. 运行测试

def run_agent(user_query: str):
    """运行 Agent 并打印详细的思考过程"""
    print(f"=" * 60)
    print(f"👤 用户问题：{user_query}")
    print(f"=" * 60)
    
    result = agent.invoke({"messages": [user_query]})
    messages = result["messages"]
    
    for i, msg in enumerate(messages):
        if hasattr(msg, "tool_calls") and msg.tool_calls:
            # LLM 决定调用工具
            for tc in msg.tool_calls:
                print(f"\n🧠 Thought: 需要调用工具")
                print(f"🔧 Action: {tc['name']}")
                print(f"📥 Action Input: {json.dumps(tc['args'], ensure_ascii=False)}")
        elif hasattr(msg, "content") and msg.content:
            # LLM 的思考或最终回答
            role = "👤" if hasattr(msg, "type") and msg.type == "human" else "🤖"
            content = msg.content
            if len(content) > 200:
                content = content[:200] + "..."
            if role == "🤖" and i > 0:  # 跳过 system message
                print(f"\n💬 Observation/Answer: {content}")
    
    # 最终答案
    last_msg = messages[-1]
    final_answer = last_msg.content if hasattr(last_msg, "content") else str(last_msg)
    print(f"\n{'=' * 60}")
    print(f"✅ 最终回答：\n{final_answer}")
    print(f"{'=' * 60}")
    return final_answer


# 场景 1：排查服务器问题
run_agent("web-02 服务器 CPU 飙到 78% 了，帮我看看什么情况，并给出处理建议")

# 场景 2：排查订单问题
run_agent("今天有没有失败的订单？帮我查一下订单表")

# 场景 3：故障处理后再发告警
run_agent("db-master 磁盘 72%，查一下 MySQL 连接情况，如果正常就发个 info 通知")

6. 执行效果示例

运行场景 3 的输出大概是这样：

============================================================
👤 用户问题：db-master 磁盘 72%，查一下 MySQL 连接情况，
   如果正常就发个 info 通知
============================================================

🧠 Thought: 需要先检查 db-master 的状态
🔧 Action: check_server_status
📥 Action Input: {"server_name": "db-master"}

💬 Observation: {"server":"db-master","cpu_percent":23,...}

🧠 Thought: 磁盘 72% 需要关注，再查一下 MySQL 连接池
🔧 Action: search_knowledge_base
📥 Action Input: {"query": "mysql连接池耗尽"}

💬 Observation: {"mysql连接池耗尽":{"原因":[...],...}}

🧠 Thought: 服务器状态正常，没有连接池问题，
       可以发一个 info 级别的通知
🔧 Action: send_alert
📥 Action Input: {"level":"info","title":"db-master 巡检通知",...}

============================================================
✅ 最终回答：
db-master 当前运行正常：CPU 23%，内存 55%，磁盘 72%。
磁盘使用率 72% 虽未触发告警阈值（80%），但建议关注。
已通过企业微信发送巡检通知。
============================================================

---

日常工作结合场景

这个 Agent 框架可以直接应用在以下真实场景：

场景 1：故障自动排查

用户输入："线上订单支付失败率突然升高，帮我排查"
Agent 动作序列：
  1. 查数据库 → 最近 5 分钟失败订单的 error_code
  2. 查知识库 → error_code 对应的处理方案
  3. 查服务器状态 → 支付网关服务的 CPU/内存
  4. 综合信息 → 给出根因分析和处理建议
  5. 发送告警 → 通知相关值班人员

场景 2：日报自动生成

用户输入："生成本周运维周报"
Agent 动作序列：
  1. 查数据库 → 本周故障工单统计
  2. 查服务器状态 → 各服务 SLA 数据
  3. 查知识库 → 本周更新的运维文档
  4. 组织成周报格式 → 输出 Markdown

场景 3：权限/配置自助查询

把 Agent 接入企业 IM（企微/飞书/钉钉），团队成员可以自然语言查询：

• "我有哪些数据库权限？"
• "xxx 服务的配置文件在哪里？"
• "帮我重启 dev 环境的 web 服务"

---

进阶扩展

1. 添加记忆能力

from langgraph.checkpoint.memory import MemorySaver

# 编译时加入 checkpoint，Agent 会记住之前的对话
memory = MemorySaver()
agent = workflow.compile(checkpointer=memory)

# 使用时指定 thread_id 保持会话
config = {"configurable": {"thread_id": "user-session-1"}}
agent.invoke({"messages": ["上次那个 web-02 的问题解决了吗？"]}, config)

2. 接入真实数据源

@tool
def query_real_database(sql: str) -> str:
    """执行真实的 MySQL 只读查询"""
    import pymysql
    conn = pymysql.connect(
        host=os.getenv("DB_HOST"),
        user=os.getenv("DB_USER"),
        password=os.getenv("DB_PASS"),
        database="monitor",
        cursorclass=pymysql.cursors.DictCursor
    )
    try:
        with conn.cursor() as cursor:
            cursor.execute(sql)
            results = cursor.fetchall()
            return json.dumps(results, ensure_ascii=False, default=str)
    finally:
        conn.close()

3. 人工确认环节

对于危险操作，加上人工审批：

@tool
def restart_service(service_name: str) -> str:
    """重启服务（需审批）"""
    # 这里接入审批系统，审批通过后才执行
    approval_id = create_approval(f"重启服务: {service_name}")
    result = wait_for_approval(approval_id, timeout=300)
    if result == "approved":
        # 执行实际重启
        return f"服务 {service_name} 重启成功"
    return "操作被拒绝"

---

关键注意事项

要点	说明
工具描述要精确	LLM 靠工具函数的 docstring 决定是否调用。写清楚参数含义和返回值
做好输入校验	Agent 可能传入奇怪的参数，工具内部必须做好边界处理
控制最大步数	max_iterations 默认通常 15 步，避免 Agent 陷入死循环
temperature 调低	运维/数据库场景建议 0.1~0.3，避免 Agent "乱来"
错误信息要友好	工具返回的错误信息会被 LLM 阅读，写清楚"哪里错了、怎么修正"
数据库操作限制	生产环境 Agent 的 SQL 工具必须限制为只读，绝不能执行 DROP/DELETE

---

总结

LangChain Agent 本质上是 LLM + 工具 + 决策循环。它不是一个黑盒，而是一个可以被完全理解和控制的流程：

1. LLM 是大脑 —— 负责推理和决策
2. Tools 是手脚 —— 负责执行具体操作
3. AgentExecutor 是神经系统 —— 协调循环，直到任务完成

对于日常运维、数据查询、故障排查等需要"理解意图 + 多步操作"的场景，Agent 比写死流程的脚本灵活得多。花半天搭一个 Agent 原型，可能帮你省掉未来几个月的手工运维时间。