B4 LLM决策模块：验收讲解 —— 从基础到进阶的五维实践

1
B1 Runtime → 调用 B4.generate_ai_message(messages, tools_schema)
2
           → B4 加载模型 → 注入 messages + schema → LLM 推理
3
           → 解析原始输出 → 返回 AIMessage {content, tool_calls}
4
           → B1 判断是否有 tool_calls
5
             → 有：B3 执行 → ToolMessage → B1 追加消息 → 再次调 B4
6
             → 无：输出 final_answer

1
# _load_model_config() 解析 YAML
2
# _load_model_bundle()  加载 tokenizer + model
3
config = read_yaml(model_config_path)
4
tokenizer = AutoTokenizer.from_pretrained(**config)
5
model = AutoModelForCausalLM.from_pretrained(**config)

1
# _build_prompt_messages() 将 tools_schema JSON 注入 prompt
2
system_prompt = read_prompt("local_tool_agent.txt")
3
system_prompt += "\n\nAvailable tools:\n" + json.dumps(tools_schema)
4
messages = [{"role": "system", "content": system_prompt}] + user_messages

1
{
2
    "role": "assistant",
3
    "content": "",
4
    "tool_calls": [
5
        {
6
            "id": "call_001",
7
            "name": "file_reader",
8
            "args": { "path": "docs/agent_intro.txt", "max_chars": 2000 }
9
        },
10
        {
11
            "id": "call_002",
12
            "name": "calculator",
13
            "args": { "expression": "3.14 * 5" }
14
        }
15
    ]
16
}

1
def generate_ai_message(
2
    model_config: str,       # model.yaml 路径
3
    messages: list[dict],    # 消息序列
4
    tools_schema: list[dict],# 工具说明
5
    mode: str = "prompt_json",
6
    model_name: str = None,
7
    tool_calling: str = None,
8
) -> dict:                   # 标准 AIMessage

1
"Valid schema B:\n"
2
"Valid schema B (call one or more tools, content must be empty):\n"
3
'{"content":"","tool_calls":[{"id":"call_001","name":"file_reader",'
4
'"args":{"path":"docs/agent_intro.txt","max_chars":2000}}]}\n\n'
5
'"args":{"path":"docs/agent_intro.txt","max_chars":2000}},{"id":"call_002",'
6
'"name":"calculator","args":{"expression":"2+2"}}]}\n\n'
7
...
8
"Choose exactly one schema: final content with an empty tool_calls array, or empty content with tool calls. "
9
"You may include zero, one, or multiple tool_calls in the array. "

1
def _mock_generate(messages: list[dict]) -> dict:
2
    tool_messages = [m for m in messages if m.get("role") == "tool"]
3
    if not tool_messages:
4
        # 多个 tool_calls 演示
5
        return make_ai_message("", [
6
            {"id": "call_001", "name": "file_reader",
7
             "args": {"path": "docs/agent_intro.txt", "max_chars": 2000}},
8
            {"id": "call_002", "name": "local_file_search",
9
             "args": {"query": "Agent"}},
10
        ])
11
    latest = tool_messages[-1]
12
    result = _extract_tool_result(latest)
13
    if latest.get("status") != "success" ...
14
        return make_ai_message(f"工具调用失败，无法完成请求：{detail}", [])
15
    output = result.get("output") or {}
16
    content = output.get("content") if isinstance(output, dict) else None
17
    ...
18
    answer = "三条中文要点如下：\n" + ...
19
    return make_ai_message(answer, [])
20
    # 遍历所有 ToolMessage，逐个检查状态
21
    for tm in tool_messages:
22
        if tm.get("status") != "success":
23
            try:
24
                result = _extract_tool_result(tm)
25
                if result.get("status") != "success":
26
                    ...
27
                    return make_ai_message(f"工具调用失败，无法完成请求：{detail}", [])
28
            except ValueError:
29
                return make_ai_message(f"工具调用失败：无法解析工具返回内容", [])
30
    # 全部成功则汇总所有 ToolMessage 的结果
31
    summaries = []
32
    for tm in tool_messages:
33
        try:
34
            result = _extract_tool_result(tm)
35
            output = result.get("output") or {}
36
            content = output.get("content") if isinstance(output, dict) else None
37
            if isinstance(content, str) and content.strip():
38
                summaries.append(content)
39
        except ValueError:
40
            pass
41
    combined = "\n".join(summaries) if summaries else "工具结果未提供可提取内容"
42
    points = _three_points(combined)
43
    answer = "三条中文要点如下：\n" + points
44
    return make_ai_message(answer, [])

1
{
2
    "role": "assistant",
3
    "content": "",
4
    "tool_calls": [
5
        {
6
            "id": "call_001",
7
            "name": "file_reader",
8
            "args": { "path": "docs/agent_intro.txt", "max_chars": 2000 }
9
        },
10
        { "id": "call_002", "name": "local_file_search", "args": { "query": "Agent" } }
11
    ]
12
}

1
{
2
    "reasoning": "用户需要总结文档要点，同时了解搜索功能，最后计算示例。",
3
    "plan": [
4
        {
5
            "step": 1,
6
            "description": "读取 Agent 介绍文档",
7
            "tool_call": {
8
                "name": "file_reader",
9
                "args": { "path": "docs/agent_intro.txt", "max_chars": 2000 }
10
            }
11
        },
12
        {
13
            "step": 2,
14
            "description": "执行示例数学计算",
15
            "tool_call": {
16
                "name": "calculator",
17
                "args": { "expression": "3.14 * 5" }
18
            }
19
        }
20
    ]
21
}

1
elif mode == "plan_execute":
2
    tool_messages = [m for m in messages if m.get("role") == "tool"]
3
    backend_plan = config.get("model", {}).get("backend", "transformers")
4
    if backend_plan == "mock" or not _has_torch():
5
        if not tool_messages:
6
            # 阶段 1: 生成计划（Mock 返回 3 步固定计划）
7
            ai_message = _mock_plan_execute(messages)
8
            ...
9
            parsed_candidate = {
10
                "content": "",
11
                "tool_calls": ai_message["tool_calls"],
12
                "plan_step_mode": "plan",
13
                "reasoning": "Mock plan.",
14
                "total_steps": len(ai_message["tool_calls"]),
15
            }
16
            status = "success"
17
        else:
18
            # 阶段 2: 步骤执行（Mock 合并工具结果）
19
            ai_message = _mock_generate(messages)
20
            ...
21
            parsed_candidate = {
22
                "content": ai_message["content"],
23
                "tool_calls": [],
24
                "plan_step_mode": "final",
25
            }
26
            status = "success"
27
    else:
28
    raise ValueError("mode must be mock or prompt_json")
29
        # 真实模型：plan → step 双阶段
30
        if not tool_messages:
31
            plan_messages = _build_plan_prompt_messages(messages, tools_schema)
32
            raw_text = _prompt_json_generate(
33
                config_path, config, plan_messages, tools_schema
34
            )
35
            parsed_candidate, ai_message = _parse_plan_output(raw_text)
36
            status = "success"
37
        else:
38
            step_messages = _build_plan_step_prompt_messages(messages, tools_schema)
39
            raw_text = _prompt_json_generate(
40
                config_path, config, step_messages, tools_schema
41
            )
42
            parsed_candidate, ai_message = _parse_model_output(raw_text)
43
            status = "success"

1
{
2
    "role": "assistant",
3
    "content": "",
4
    "tool_calls": [
5
        { "name": "file_reader", "args": { "path": "docs/agent_intro.txt" } },
6
        { "name": "local_file_search", "args": { "query": "Agent", "root_dir": "docs" } },
7
        { "name": "calculator", "args": { "expression": "2 + 2" } }
8
    ]
9
}

1
{
2
    "role": "assistant",
3

4
    "content": "三条中文要点如下：\n1. Agent 系统通常由模型、工具、记忆和执行循环组成\n2. 工具调用让模型能够读取本地文件、执行计算\n3. Memory 为 Agent 提供全局知识和历史对话上下文",
5

6
    "tool_calls": []
7
}

1
def _load_model_config(model_config: str | Path) -> tuple[Path, dict]:
2
def _load_model_config(
3
    model_config: str | Path, model_name: str | None = None
4
) -> tuple[Path, dict]:
5
    path = Path(model_config).resolve()
6
    config = read_yaml(path)
7
    if not isinstance(config, dict):
8
        raise ValueError("model.yaml must contain an object")
9
    if model_name:
10
        models_section = config.get("models", {})
11
        if not isinstance(models_section, dict):
12
            raise ValueError("model.yaml 'models' section must be an object")
13
        if model_name not in models_section:
14
            available = ", ".join(models_section.keys())
15
            raise ValueError(
16
                f"unknown model_name '{model_name}'. Available: {available}"
17
            )
18
        selected = deepcopy(models_section[model_name])
19
        config["model"] = selected
20
        config["model"]["_selected_model_name"] = model_name
21
        print(
22
            f"model: {selected.get('display_name', model_name)}",
23
            file=sys.stderr, flush=True,
24
        )
25
    return path, config

1
def generate_ai_message(
2
    model_config: str,
3
    messages: list[dict],
4
    tools_schema: list[dict],
5
    mode: str = "prompt_json",
6
    model_name: str | None = None,
7
) -> dict:
8
    config_path, config = _load_model_config(model_config)
9
    config_path, config = _load_model_config(model_config, model_name)
10

11
def build_parser() -> argparse.ArgumentParser:
12
    ...
13
    parser.add_argument(
14
        "--model_name", default=None,
15
        help="Select named model profile from model.yaml (models section)",
16
    )

1
def _prompt_json_generate(
2
    config_path: Path,
3
    config: dict,
4
    messages: list[dict],
5
    tools_schema: list[dict],
6
    tool_calling_mode: str = "prompt_json",
7
) -> str:
8
    ...
9
    # 加载模型...
10
    prompt_messages = _build_prompt_messages(messages, tools_schema)
11
    inputs = tokenizer.apply_chat_template(
12
        prompt_messages,
13
        tokenize=True,
14
        add_generation_prompt=True,
15
        return_tensors="pt", return_dict=True,
16
    )
17
    tool_calling_mode = config.get("tool_calling", {}).get("mode", "prompt_json")
18
    if tool_calling_mode == "builtin":
19
        # 内置传参：不注入 prompt，通过 chat template 的 tools= 参数传入
20
        prompt_messages = deepcopy(messages)
21
        inputs = tokenizer.apply_chat_template(
22
            prompt_messages,
23
            tools=tools_schema,  # ← 关键区别
24
            tokenize=True,
25
            add_generation_prompt=True,
26
            return_tensors="pt", return_dict=True,
27
        )
28
    else:
29
        # prompt 注入：tools_schema 拼接到 system message 文本中
30
        prompt_messages = _build_prompt_messages(messages, tools_schema)
31
        inputs = tokenizer.apply_chat_template(
32
            prompt_messages,
33
            tokenize=True,
34
            add_generation_prompt=True,
35
            return_tensors="pt", return_dict=True,
36
        )

1
from b4_local_agent_llm import generate_ai_message
2

3
def run_benchmark(model_config, cases_dir, tools_schema, mode, ...) -> dict:
4
    cases_path = Path(cases_dir).resolve()
5
    case_files = sorted(cases_path.glob("*.json"))
6
    tools = read_json(resolve_cli_path(tools_schema))
7
    results = []
8
    total_success = 0
9
    total_error = 0
10

11
    for idx, case_file in enumerate(case_files):
12
        messages = read_json(case_file)
13
        case_name = case_file.stem
14
        start = time.perf_counter()
15
        result = generate_ai_message(
16
            model_config, messages, tools, mode,
17
            artifact_dir=outdir,
18
            artifact_stem=f"bench_{case_name}" if outdir else None,
19
            model_name=model_name,
20
            tool_calling=tool_calling,
21
        )
22
        elapsed_ms = (time.perf_counter() - start) * 1000
23

24
        record = {
25
            "case": case_name,
26
            "status": result["status"],
27
            "latency_ms": round(elapsed_ms, 1),
28
            "tool_calls_count": len(
29
                result["ai_message"].get("tool_calls", [])
30
            ),
31
            "content_len": len(result["ai_message"].get("content", "")),
32
        }
33
        if result["status"] == "success":
34
            total_success += 1
35
        else:
36
            total_error += 1
37
            record["error"] = result.get("error", {})
38
        results.append(record)
39

40
    return {
41
        "total_cases": len(case_files),
42
        "success": total_success,
43
        "error": total_error,
44
        "success_rate": f"{total_success}/{total} ({...}%)",
45
        "avg_latency_ms": round(avg_latency, 1),
46
        "cases": results,
47
    }

1
def _candidate_to_message(candidate: dict) -> tuple[dict, dict]:
2
    ...
3
    content = candidate.get("content", "")
4
    tool_calls = candidate.get("tool_calls", [])
5

6
    # 规范化：如果两者都非空，优先使用 content，清空 tool_calls
7
    if content and tool_calls:
8
        print(
9
            "⚠️ 警告：模型同时提供了 content 和 tool_calls，已忽略 tool_calls",
10
            file=sys.stderr, flush=True
11
        )
12
        tool_calls = []   # 清空，使最终回答优先
13

14
    message = {
15
        "role": "assistant",
16
        "content": candidate.get("content", ""),
17
        "tool_calls": candidate.get("tool_calls", []),
18
        "content": content,
19
        "tool_calls": tool_calls,
20
    }
21
    validate_ai_message(message)
22
    has_content = bool(message["content"].strip())
23
    has_tool_calls = bool(message["tool_calls"])
24
    if has_content == has_tool_calls:
25
        raise ValueError(
26
            "model output must contain either final content or tool calls, but not both"
27
        )
28
    # 移除互斥检查（因为已经规范化）

1
def _extract_json_object(text: str) -> dict | None:
2
    """从文本中依次尝试每个 '{' 位置，直到成功解析出一个 JSON 对象。"""
3
    decoder = json.JSONDecoder()
4
    pos = 0
5
    while True:
6
        start = text.find('{', pos)
7
        if start == -1:
8
            return None
9
        try:
10
            obj, end = decoder.raw_decode(text[start:])
11
            return obj
12
        except json.JSONDecodeError:
13
            pos = start + 1
14
            continue
15

16
def _parse_model_output(raw_text: str) -> tuple[dict, dict]:
17
    try:
18
        candidate = json.loads(raw_text.strip())
19
    except json.JSONDecodeError as exc:
20
        # 尝试提取 JSON 对象
21
        extracted = _extract_json_object(raw_text)
22
        if extracted is not None:
23
            try:
24
                return _candidate_to_message(extracted)
25
            except Exception:
26
                pass  # 不符合要求则继续其他容错
27
        # 原有容错逻辑
28
        try:
29
            candidate = _parse_json_with_backtick_tail(raw_text, exc)