如果大模型只能"一问一答",那它本质上就是一个高级搜索引擎。真正的价值在于它能自己决定做什么——调用工具、分析结果、再决定下一步,直到完成任务。这就是AI Agent的核心思想;

去年给公司搭了一个内部运维Agent,能自动查日志、分析异常、甚至根据情况重启服务。从"提一个问题等一个答案"到"给一个目标它自己搞定",体验是质的飞跃;

Agent的运行机制

Agent的工作循环可以概括为:观察(看当前状态和用户需求)→思考(分析需要做什么)→行动(调用工具)→观察(看工具返回的结果)→思考(分析结果、决定下一步)→行动……直到任务完成;

这个循环由一个"推理引擎"驱动——就是大模型本身。模型根据工具的描述决定调用哪个工具,根据工具的返回值决定下一步。所以工具描述写得好不好,直接决定了Agent的智商。

定义工具

from langchain.tools import tool
import json

@tool
def search_web(query: str) -> str:
    """搜索互联网获取实时信息。当你需要查找最新新闻、技术文档、或任何不在你知识库中的信息时使用。"""
    from duckduckgo_search import DDGS
    with DDGS() as ddgs:
        results = list(ddgs.text(query, max_results=5))
    return json.dumps([{"title": r["title"], "snippet": r["body"]} for r in results],
                      ensure_ascii=False)

@tool
def calculate(expression: str) -> str:
    """计算数学表达式。输入应该是合法的数学表达式,如(2+3)*4。只支持基本运算。"""
    try:
        allowed = set("0123456789+-*/.() ")
        if not all(c in allowed for c in expression):
            return "错误:包含不允许的字符"
        return f"结果:{eval(expression)}"
    except Exception as e:
        return f"计算错误:{e}"

@tool
def query_database(sql: str) -> str:
    """执行只读SQL查询。只支持SELECT语句,禁止INSERT/UPDATE/DELETE。"""
    if not sql.strip().upper().startswith("SELECT"):
        return "错误:只允许SELECT语句"
    import sqlite3
    conn = sqlite3.connect("app.db")
    try:
        rows = conn.execute(sql).fetchall()
        return json.dumps(rows[:50], ensure_ascii=False)  # 限制50行
    finally:
        conn.close()

工具描述是Agent选择工具的依据。描述要清晰说明"什么时候用这个工具"和"输入格式是什么"。描述模糊的工具,Agent要么不用要么乱用。

创建Agent

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder

llm = ChatOpenAI(
    base_url="http://localhost:11434/v1",
    api_key="ollama",
    model="deepseek-r1:7b",
    temperature=0
)

tools = [search_web, calculate, query_database]

prompt = ChatPromptTemplate.from_messages([
    ("system", """你是一个智能助手,可以使用工具来完成任务。

可用工具:{tool_names}

决策原则:
1. 优先尝试用工具获取准确数据,而不是凭记忆回答
2. 如果工具返回的结果不够好,尝试换一种查询方式
3. 完成任务后给出清晰的总结"""),
    ("user", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])

agent = create_openai_tools_agent(llm, tools, prompt)
executor = AgentExecutor(
    agent=agent, tools=tools,
    verbose=True,           # 打印推理过程,调试必备
    max_iterations=10,      # 防止无限循环
    handle_parsing_errors=True
)

verbose=True一定要开,它会打印Agent每一步的思考过程和工具调用,出了问题才知道卡在哪一步。

添加记忆

默认Agent没有上下文记忆。加ConversationBufferMemory让它记住对话历史:

from langchain.memory import ConversationBufferMemory

memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)
executor = AgentExecutor(agent=agent, tools=tools, memory=memory, verbose=True)

executor.invoke({"input": "我负责的项目代号叫Phoenix"})
executor.invoke({"input": "Phoenix项目最近有什么告警?"})  # 记住了Phoenix

安全设计

Agent能调用工具就意味着它能对外部系统产生影响。几个安全要点:数据库工具只给SELECT权限,用只读账号;删除、重启类操作加确认机制——Agent输出"建议重启服务X"但不直接执行,由人确认后执行;日志审计记录所有工具调用,包括输入参数和返回值。

写在最后

Agent不是万能的。对于流程固定的任务(比如ETL流水线),用传统的工作流更可靠。Agent适合"步骤不确定、需要根据中间结果做决策"的场景。分清这两种场景,别把Agent用在不需要它的地方。