上周在做一个内部运维Agent的时候，遇到一个典型问题：用户问"服务器怎么又挂了"，Agent需要查监控、查日志、查变更记录，最后给出结论。如果让LLM自己编造答案，那就是胡说八道；如果每次都人工介入，那Agent就失去了意义。最终靠Function Calling + 多步推理解决了这个问题。
这篇文章把Function Calling的实现细节和多步推理的工程实践整理出来，包括如何定义工具、如何让LLM自主决定调用顺序、以及踩过的几个坑。
Function Calling是什么
简单说，就是让LLM能"调用函数"。你告诉它有哪些工具可用、每个工具接受什么参数，LLM自己决定什么时候调用哪个工具、传什么参数。你拿到调用请求后去执行，把结果喂回LLM，它再继续推理。
整个流程可以用这个图来概括：
定义工具
以查服务器状态为例，定义一个获取监控指标的工具：
tools = [
{
"type": "function",
"function": {
"name": "get_server_metrics",
"description": "获取服务器的实时监控指标，包括CPU、内存、磁盘使用率",
"parameters": {
"type": "object",
"properties": {
"server_id": {
"type": "string",
"description": "服务器ID，如srv-001"
},
"metric": {
"type": "string",
"enum": ["cpu", "memory", "disk", "all"],
"description": "要查询的指标类型"
},
"duration": {
"type": "string",
"description": "查询时间范围，如1h、24h",
"default": "1h"
}
},
"required": ["server_id", "metric"]
}
}
},
{
"type": "function",
"function": {
"name": "get_recent_logs",
"description": "获取服务器最近的错误日志",
"parameters": {
"type": "object",
"properties": {
"server_id": {
"type": "string",
"description": "服务器ID"
},
"level": {
"type": "string",
"enum": ["error", "warn", "info"],
"description": "日志级别"
},
"limit": {
"type": "integer",
"description": "返回条数，默认20"
}
},
"required": ["server_id"]
}
}
}
]
工具定义的关键是description要写得清楚。LLM是靠自然语言理解来决定调哪个工具的，description写得模糊它就会选错。
调用LLM并处理工具请求
import json
from openai import OpenAI
client = OpenAI()
def run_agent(user_message, max_rounds=5):
messages = [
{"role": "system", "content": "你是一个运维助手，可以查询服务器监控和日志。"},
{"role": "user", "content": user_message}
]
for round in range(max_rounds):
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools,
tool_choice="auto"
)
msg = response.choices[0].message
# 没有工具调用，直接返回回答
if not msg.tool_calls:
return msg.content
# 处理每个工具调用
messages.append(msg)
for tc in msg.tool_calls:
result = execute_tool(tc.function.name, json.loads(tc.function.arguments))
messages.append({
"role": "tool",
"tool_call_id": tc.id,
"content": json.dumps(result, ensure_ascii=False)
})
return "推理轮数超过限制，请简化问题。"
这里max_rounds=5是防止LLM陷入无限循环。实际跑下来，简单问题1轮就够，复杂排查通常3-4轮。
工具执行层
def execute_tool(name, args):
if name == "get_server_metrics":
# 实际会调监控系统API
return query_prometheus(args["server_id"], args.get("metric", "all"), args.get("duration", "1h"))
elif name == "get_recent_logs":
return query_elasticsearch(args["server_id"], args.get("level", "error"), args.get("limit", 20))
else:
return {"error": f"未知工具: {name}"}
这个层要把LLM的参数翻译成实际的API调用。注意要做参数校验，LLM有时候会传一些奇怪的值。
多步推理的关键
多步推理是Function Calling最强大的地方。LLM不是只调一次工具就完事，它会根据第一次的结果决定要不要再调别的工具。
举个实际场景：
这个过程中LLM自己决定了调用顺序：先看指标、再看日志、再查变更。你不需要写死这个流程，LLM会根据上下文自主判断。
踩过的坑
工具描述不准确导致调错
一开始我把get_server_metrics的description写成了"获取服务器信息"，结果用户问"服务器在哪"的时候LLM也去调这个工具。改成"获取服务器实时监控指标"后就对了。
教训：description要精确描述工具的功能边界，不要写得太笼统。
参数类型不匹配
LLM有时候会把整数传成字符串，或者把枚举值拼写错误。建议在工具执行层加一层参数校验：
def validate_params(schema, params):
"""简易参数校验"""
for field, rules in schema.get("properties", {}).items():
if field in params:
if rules.get("type") == "integer":
params[field] = int(params[field])
if "enum" in rules and params[field] not in rules["enum"]:
return None, f"{field} must be one of {rules['enum']}"
return params, None
循环调用停不下来
有些场景LLM会反复调同一个工具。比如查日志没找到想要的信息，它会换个参数再查一次，还是没找到，又换个参数。加了max_rounds限制后解决了，但更好的做法是让LLM在工具返回结果不理想时直接告诉用户"没找到相关信息"。
# 在system prompt里加一句约束
"如果工具返回的信息不足以回答问题，直接告诉用户现有信息不足，不要反复调用相同工具。"
流式输出支持
Function Calling也支持流式输出，但实现起来更复杂：
def run_agent_stream(user_message, max_rounds=5):
messages = [
{"role": "system", "content": "你是一个运维助手。"},
{"role": "user", "content": user_message}
]
for round in range(max_rounds):
stream = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools,
stream=True
)
tool_calls_acc = []
final_content = ""
for chunk in stream:
delta = chunk.choices[0].delta
if delta.content:
final_content += delta.content
yield delta.content
if delta.tool_calls:
# 累积工具调用
for tc in delta.tool_calls:
while len(tool_calls_acc) <= tc.index:
tool_calls_acc.append({"id": "", "function": {"name": "", "arguments": ""}})
if tc.id:
tool_calls_acc[tc.index]["id"] = tc.id
if tc.function.name:
tool_calls_acc[tc.index]["function"]["name"] = tc.function.name
if tc.function.arguments:
tool_calls_acc[tc.index]["function"]["arguments"] += tc.function.arguments
if not tool_calls_acc:
return
# 执行工具并继续
msg = {"role": "assistant", "tool_calls": [
{"id": tc["id"], "type": "function", "function": tc["function"]}
for tc in tool_calls_acc
]}
messages.append(msg)
for tc in tool_calls_acc:
result = execute_tool(tc["function"]["name"], json.loads(tc["function"]["arguments"]))
messages.append({
"role": "tool",
"tool_call_id": tc["id"],
"content": json.dumps(result, ensure_ascii=False)
})
流式版本代码量大了不少，但用户体验好很多——用户能实时看到LLM的思考过程和工具调用结果。
性能优化
并行工具调用
如果LLM一次返回多个工具调用请求，可以并行执行：
import asyncio
async def execute_tools_parallel(tool_calls):
tasks = []
for tc in tool_calls:
name = tc.function.name
args = json.loads(tc.function.arguments)
tasks.append(asyncio.to_thread(execute_tool, name, args))
return await asyncio.gather(*tasks)
实测并行执行可以节省30-50%的响应时间，尤其是需要同时查多个数据源的时候。
工具结果压缩
有时候工具返回的数据量很大，比如日志查询可能返回几百行。直接全部喂给LLM会浪费token，建议做个摘要：
def compress_tool_result(result, max_chars=2000):
text = json.dumps(result, ensure_ascii=False)
if len(text) <= max_chars:
return text
# 截断并加提示
return text[:max_chars] + f"\n... (结果已截断，原始长度{len(text)}字符)"
写在最后
Function Calling是构建AI Agent的基础能力。做好了，LLM就是一个能自主决策、按需调用工具的智能体；做不好，就是一个会调错函数的麻烦制造者。
关键要点：工具描述要精确、参数校验要严格、循环调用要限制、结果要压缩。把这几个点做好，你的Agent就稳定了。