26. Function Calling 执行 Loop：模型吐出 tool call 后，后端到底做什么#

1
Function Calling 是机制
2
tool call 是模型输出的一次工具调用请求
3
TOOL_FUNCTIONS 是后端找真实函数的映射表

1
用户输入
2
  -> 模型看到 TOOLS
3
  -> 模型输出 tool_call
4
  -> 后端解析 arguments
5
  -> 后端用 TOOL_FUNCTIONS 找函数
6
  -> 后端执行函数
7
  -> 后端把 tool output 放回 messages
8
  -> 模型生成最终回答

1
TOOLS_FUNCTIONS = {
2
    "search_knowledge_base": search_knowledge_base
3
}

1
TOOL_FUNCTIONS = {
2
    "search_knowledge_base": search_knowledge_base
3
}

1
1. 把用户消息和 TOOLS 发给模型
2
2. 检查模型有没有返回 tool_calls
3
3. 如果有，后端执行对应工具
4
4. 把工具结果作为 tool message 追加进 messages
5
5. 再问模型一次，让模型基于工具结果回答

1
tool call 不是最终答案
2
tool output 也不是最终答案
3
最终答案是模型读完 tool output 之后生成的自然语言回答

1
search_knowledge_base 函数
2
search_knowledge_base 这个函数名字符串
3
tool_call 数据结构
4
Function Calling 机制

1
真实函数：
2
def search_knowledge_base(query: str, limit: int = 3):
3
    ...
4

5
工具说明书：
6
{"name": "search_knowledge_base", "parameters": {...}}
7

8
模型吐出的 tool_call：
9
{"name": "search_knowledge_base", "arguments": "{\"query\":\"RAG\"}"}
10

11
Function Calling：
12
让模型根据工具说明书吐出 tool_call 的机制

1
TOOL_FUNCTIONS = {
2
    "search_knowledge_base": search_knowledge_base,
3
}

1
tool_name = "search_knowledge_base"

1
tool_func = TOOL_FUNCTIONS[tool_name]

1
tool_result = tool_func(**tool_args)

1
{"query": "RAG Evaluation 分几层？", "limit": 3}

1
arguments = '{"query": "RAG Evaluation 分几层？", "limit": 3}'

1
import json
2

3
tool_args = json.loads(arguments)

1
{
2
    "query": "RAG Evaluation 分几层？",
3
    "limit": 3,
4
}

1
import json
2

3
from app.tools.registry import TOOL_FUNCTIONS
4

5

6
def run_tool_call(tool_call):
7
    tool_name = tool_call.function.name
8
    raw_arguments = tool_call.function.arguments
9
    tool_args = json.loads(raw_arguments)
10

11
    if tool_name not in TOOL_FUNCTIONS:
12
        raise ValueError(f"未知工具：{tool_name}")
13

14
    tool_func = TOOL_FUNCTIONS[tool_name]
15
    return tool_func(**tool_args)

1
tool_call -> Python 函数执行结果

1
怎么把结果重新发回模型

1
[
2
  {
3
    "rank": 1,
4
    "title": "RAG 评估与指标",
5
    "content": "RAG 评估分为 Retrieval、Context、Answer 三层。"
6
  }
7
]

1
RAG 评估通常分三层：Retrieval Evaluation、Context Evaluation、Answer Evaluation。

1
后端执行工具
2
  -> 得到 tool output
3
  -> 把 tool output 放回 messages
4
  -> 再问模型
5
  -> 模型组织最终回答

1
这是刚才那个工具调用的结果

1
messages.append({
2
    "role": "tool",
3
    "tool_call_id": tool_call.id,
4
    "content": json.dumps(tool_result, ensure_ascii=False),
5
})

1
刚才模型可能一次请求多个工具
2
tool_call_id 用来说明这个结果属于哪个 tool_call

1
import json
2

3
from app.routers.ai import client, MODEL_NAME
4
from app.tools.registry import TOOLS, TOOL_FUNCTIONS
5

6

7
MAX_STEPS = 3
8

9

10
def run_agent(user_message: str) -> str:
11
    messages = [
12
        {
13
            "role": "system",
14
            "content": "你是一个可以使用工具的助手。需要查项目知识库时使用工具。",
15
        },
16
        {"role": "user", "content": user_message},
17
    ]
18

19
    for _ in range(MAX_STEPS):
20
        response = client.chat.completions.create(
21
            model=MODEL_NAME,
22
            messages=messages,
23
            tools=TOOLS,
24
        )
25

26
        assistant_message = response.choices[0].message
27
        messages.append(assistant_message)
28

29
        if not assistant_message.tool_calls:
30
            return assistant_message.content
31

32
        for tool_call in assistant_message.tool_calls:
33
            tool_name = tool_call.function.name
34
            tool_args = json.loads(tool_call.function.arguments)
35

36
            if tool_name not in TOOL_FUNCTIONS:
37
                raise ValueError(f"未知工具：{tool_name}")
38

39
            tool_func = TOOL_FUNCTIONS[tool_name]
40
            tool_result = tool_func(**tool_args)
41

42
            messages.append({
43
                "role": "tool",
44
                "tool_call_id": tool_call.id,
45
                "content": json.dumps(tool_result, ensure_ascii=False),
46
            })
47

48
    return "工具调用次数过多，已停止。"

1
缺少 query
2
limit 是字符串 "很多"
3
工具名不存在
4
参数里夹带无关内容
5
limit 给 100000

1
def normalize_search_args(args: dict) -> dict:
2
    query = str(args.get("query", "")).strip()
3
    if not query:
4
        raise ValueError("query 不能为空")
5

6
    limit = int(args.get("limit", 3))
7
    limit = max(1, min(limit, 5))
8

9
    return {
10
        "query": query,
11
        "limit": limit,
12
    }

1
def run_tool_call_safely(tool_call, current_user):
2
    tool_name = tool_call.function.name
3
    tool_args = json.loads(tool_call.function.arguments)
4

5
    if tool_name not in TOOL_FUNCTIONS:
6
        raise ValueError("未知工具")
7

8
    if tool_name == "delete_document":
9
        if not current_user.is_admin:
10
            raise PermissionError("没有权限删除文档")
11

12
    tool_func = TOOL_FUNCTIONS[tool_name]
13
    return tool_func(**tool_args)

1
模型负责提出工具调用请求
2
后端负责校验、授权、执行、记录

1
注册工具
2
解析 tool call
3
执行工具
4
把 tool output 放回模型
5
循环控制

1
agent.invoke(...)

1
模型 -> tool call -> 后端工具 -> tool output -> 模型