如果大模型只能"一问一答",那它本质上就是一个高级搜索引擎。真正的价值在于它能自己决定做什么——调用工具、分析结果、再决定下一步,直到完成任务。这就是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用在不需要它的地方。