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

flowchart TB
    subgraph "Agent 系统架构 (B1-B5)"
        direction TB
        B5["B5 记忆模块<br/>读取/保存记忆文档"]
        B1["B1 Runtime<br/>消息编排 + 循环控制"]
        B4["✅ B4 LLM 决策<br/>调用模型生成 AIMessage"]
        B3["B3 工具层<br/>生成 schema + 执行工具"]
        B2["B2 Skill<br/>具体工具函数"]
    end

    User["用户输入"] --> B1
    B1 --> B5
    B5 --> B1
    B1 --> B4
    B4 --> B1
    B1 --> B3
    B3 --> B2
    B2 --> B3
    B3 --> B1
    B1 --> Output["最终回答"]

    style B4 fill:#4CAF50,color:#fff,stroke:#333

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

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

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

flowchart TB
    A["模型原始输出<br/>(文本字符串)"] --> B["json.loads()<br/>标准JSON解析"]
    B --> C{"解析成功？"}
    C -->|是| D["_candidate_to_message()<br/>→ 标准 AIMessage"]
    C -->|否| E["_extract_json_object()<br/>智能提取 JSON"]
    E --> F{"提取成功？"}
    F -->|是| D
    F -->|否| G["_parse_json_with_backtick_tail()<br/>_parse_tool_calls_fragment()<br/>容错回退"]
    G --> D

{
    "role": "assistant",
    "content": "",
    "tool_calls": [
        {
            "id": "call_001",
            "name": "file_reader",
            "args": { "path": "docs/agent_intro.txt", "max_chars": 2000 }
        },
        {
            "id": "call_002",
            "name": "calculator",
            "args": { "expression": "3.14 * 5" }
        }
    ]
}

sequenceDiagram
    participant User as 用户
    participant B1 as B1 Runtime
    participant B5 as B5 记忆
    participant B4 as B4 LLM
    participant B3 as B3 工具层
    participant B2 as B2 Skill

    User->>B1: 提交问题
    B1->>B5: 查找记忆
    B5-->>B1: 返回 global/selected memory
    B1->>B4: generate_ai_message(messages, tools_schema)
    B4->>B4: 加载模型 → LLM 推理 → 解析 AIMessage
    B4-->>B1: 返回 AIMessage {content, tool_calls}

    B1->>B1: 判断是否有 tool_calls?

    alt 有 tool_calls
        B1->>B3: 执行 tool_calls
        B3->>B2: 调用对应 Skill
        B2-->>B3: 返回工具结果
        B3-->>B1: 返回 ToolMessage
        B1->>B1: 追加 AI + Tool 到 messages
        B1->>B4: 再次 generate_ai_message(完整 messages)
        B4-->>B1: 新 AIMessage
        B1->>B1: 再次判断...（循环直到无 tool_calls）
    else 无 tool_calls
        B1-->>User: 输出 final_answer
    end

    B1->>B5: 保存记忆（可选）

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

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

def _mock_generate(messages: list[dict]) -> dict:
    tool_messages = [m for m in messages if m.get("role") == "tool"]
    if not tool_messages:
        # 多个 tool_calls 演示
        return make_ai_message("", [
            {"id": "call_001", "name": "file_reader",
             "args": {"path": "docs/agent_intro.txt", "max_chars": 2000}},
            {"id": "call_002", "name": "local_file_search",
             "args": {"query": "Agent"}},
        ])
    latest = tool_messages[-1]
    result = _extract_tool_result(latest)
    if latest.get("status") != "success" ...
        return make_ai_message(f"工具调用失败，无法完成请求：{detail}", [])
    output = result.get("output") or {}
    content = output.get("content") if isinstance(output, dict) else None
    ...
    answer = "三条中文要点如下：\n" + ...
    return make_ai_message(answer, [])
    # 遍历所有 ToolMessage，逐个检查状态
    for tm in tool_messages:
        if tm.get("status") != "success":
            try:
                result = _extract_tool_result(tm)
                if result.get("status") != "success":
                    ...
                    return make_ai_message(f"工具调用失败，无法完成请求：{detail}", [])
            except ValueError:
                return make_ai_message(f"工具调用失败：无法解析工具返回内容", [])
    # 全部成功则汇总所有 ToolMessage 的结果
    summaries = []
    for tm in tool_messages:
        try:
            result = _extract_tool_result(tm)
            output = result.get("output") or {}
            content = output.get("content") if isinstance(output, dict) else None
            if isinstance(content, str) and content.strip():
                summaries.append(content)
        except ValueError:
            pass
    combined = "\n".join(summaries) if summaries else "工具结果未提供可提取内容"
    points = _three_points(combined)
    answer = "三条中文要点如下：\n" + points
    return make_ai_message(answer, [])

flowchart TB
    subgraph "多 tool_calls 生成"
        A["messages<br/>(用户问题)"] --> B["B4 LLM"]
        B --> C["tool_calls[0]: file_reader"]
        B --> D["tool_calls[1]: calculator"]
        B --> E["tool_calls[2]: local_file_search"]
    end

    subgraph "多 ToolMessage 接收"
        C --> F["ToolMessage[0]: 文件内容"]
        D --> G["ToolMessage[1]: 计算结果"]
        E --> H["ToolMessage[2]: 搜索匹配"]
        F --> I["B4 LLM<br/>(再次调用)"]
        G --> I
        H --> I
        I --> J["最终回答:<br/>合并 3 个工具结果"]
    end

{
    "role": "assistant",
    "content": "",
    "tool_calls": [
        {
            "id": "call_001",
            "name": "file_reader",
            "args": { "path": "docs/agent_intro.txt", "max_chars": 2000 }
        },
        { "id": "call_002", "name": "local_file_search", "args": { "query": "Agent" } }
    ]
}

flowchart TB
    subgraph "ReAct 模式（基础）"
        direction TB
        R1["用户问题"] --> R2["LLM 决策"]
        R2 --> R3["Tool 1"]
        R3 --> R4["LLM 决策"]
        R4 --> R5["Tool 2"]
        R5 --> R6["LLM 决策"]
        R6 --> R7["最终回答"]
    end

    subgraph "Plan-and-Execute（进阶）"
        direction TB
        P1["用户问题"] --> P2["LLM 生成计划"]
        P2 --> P3["Plan: 3 步计划"]
        P3 --> P4["Step 1 → Tool 1"]
        P4 --> P5["Step 2 → Tool 2"]
        P5 --> P6["Step 3 → Tool 3"]
        P6 --> P7["最终回答"]
    end

{
    "reasoning": "用户需要总结文档要点，同时了解搜索功能，最后计算示例。",
    "plan": [
        {
            "step": 1,
            "description": "读取 Agent 介绍文档",
            "tool_call": {
                "name": "file_reader",
                "args": { "path": "docs/agent_intro.txt", "max_chars": 2000 }
            }
        },
        {
            "step": 2,
            "description": "执行示例数学计算",
            "tool_call": {
                "name": "calculator",
                "args": { "expression": "3.14 * 5" }
            }
        }
    ]
}

elif mode == "plan_execute":
    tool_messages = [m for m in messages if m.get("role") == "tool"]
    backend_plan = config.get("model", {}).get("backend", "transformers")
    if backend_plan == "mock" or not _has_torch():
        if not tool_messages:
            # 阶段 1: 生成计划（Mock 返回 3 步固定计划）
            ai_message = _mock_plan_execute(messages)
            ...
            parsed_candidate = {
                "content": "",
                "tool_calls": ai_message["tool_calls"],
                "plan_step_mode": "plan",
                "reasoning": "Mock plan.",
                "total_steps": len(ai_message["tool_calls"]),
            }
            status = "success"
        else:
            # 阶段 2: 步骤执行（Mock 合并工具结果）
            ai_message = _mock_generate(messages)
            ...
            parsed_candidate = {
                "content": ai_message["content"],
                "tool_calls": [],
                "plan_step_mode": "final",
            }
            status = "success"
    else:
    raise ValueError("mode must be mock or prompt_json")
        # 真实模型：plan → step 双阶段
        if not tool_messages:
            plan_messages = _build_plan_prompt_messages(messages, tools_schema)
            raw_text = _prompt_json_generate(
                config_path, config, plan_messages, tools_schema
            )
            parsed_candidate, ai_message = _parse_plan_output(raw_text)
            status = "success"
        else:
            step_messages = _build_plan_step_prompt_messages(messages, tools_schema)
            raw_text = _prompt_json_generate(
                config_path, config, step_messages, tools_schema
            )
            parsed_candidate, ai_message = _parse_model_output(raw_text)
            status = "success"

{
    "role": "assistant",
    "content": "",
    "tool_calls": [
        { "name": "file_reader", "args": { "path": "docs/agent_intro.txt" } },
        { "name": "local_file_search", "args": { "query": "Agent", "root_dir": "docs" } },
        { "name": "calculator", "args": { "expression": "2 + 2" } }
    ]
}

{
    "role": "assistant",

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

    "tool_calls": []
}

def _load_model_config(model_config: str | Path) -> tuple[Path, dict]:
def _load_model_config(
    model_config: str | Path, model_name: str | None = None
) -> tuple[Path, dict]:
    path = Path(model_config).resolve()
    config = read_yaml(path)
    if not isinstance(config, dict):
        raise ValueError("model.yaml must contain an object")
    if model_name:
        models_section = config.get("models", {})
        if not isinstance(models_section, dict):
            raise ValueError("model.yaml 'models' section must be an object")
        if model_name not in models_section:
            available = ", ".join(models_section.keys())
            raise ValueError(
                f"unknown model_name '{model_name}'. Available: {available}"
            )
        selected = deepcopy(models_section[model_name])
        config["model"] = selected
        config["model"]["_selected_model_name"] = model_name
        print(
            f"model: {selected.get('display_name', model_name)}",
            file=sys.stderr, flush=True,
        )
    return path, config

def generate_ai_message(
    model_config: str,
    messages: list[dict],
    tools_schema: list[dict],
    mode: str = "prompt_json",
    model_name: str | None = None,
) -> dict:
    config_path, config = _load_model_config(model_config)
    config_path, config = _load_model_config(model_config, model_name)

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

def _prompt_json_generate(
    config_path: Path,
    config: dict,
    messages: list[dict],
    tools_schema: list[dict],
    tool_calling_mode: str = "prompt_json",
) -> str:
    ...
    # 加载模型...
    prompt_messages = _build_prompt_messages(messages, tools_schema)
    inputs = tokenizer.apply_chat_template(
        prompt_messages,
        tokenize=True,
        add_generation_prompt=True,
        return_tensors="pt", return_dict=True,
    )
    tool_calling_mode = config.get("tool_calling", {}).get("mode", "prompt_json")
    if tool_calling_mode == "builtin":
        # 内置传参：不注入 prompt，通过 chat template 的 tools= 参数传入
        prompt_messages = deepcopy(messages)
        inputs = tokenizer.apply_chat_template(
            prompt_messages,
            tools=tools_schema,  # ← 关键区别
            tokenize=True,
            add_generation_prompt=True,
            return_tensors="pt", return_dict=True,
        )
    else:
        # prompt 注入：tools_schema 拼接到 system message 文本中
        prompt_messages = _build_prompt_messages(messages, tools_schema)
        inputs = tokenizer.apply_chat_template(
            prompt_messages,
            tokenize=True,
            add_generation_prompt=True,
            return_tensors="pt", return_dict=True,
        )

from b4_local_agent_llm import generate_ai_message

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

    for idx, case_file in enumerate(case_files):
        messages = read_json(case_file)
        case_name = case_file.stem
        start = time.perf_counter()
        result = generate_ai_message(
            model_config, messages, tools, mode,
            artifact_dir=outdir,
            artifact_stem=f"bench_{case_name}" if outdir else None,
            model_name=model_name,
            tool_calling=tool_calling,
        )
        elapsed_ms = (time.perf_counter() - start) * 1000

        record = {
            "case": case_name,
            "status": result["status"],
            "latency_ms": round(elapsed_ms, 1),
            "tool_calls_count": len(
                result["ai_message"].get("tool_calls", [])
            ),
            "content_len": len(result["ai_message"].get("content", "")),
        }
        if result["status"] == "success":
            total_success += 1
        else:
            total_error += 1
            record["error"] = result.get("error", {})
        results.append(record)

    return {
        "total_cases": len(case_files),
        "success": total_success,
        "error": total_error,
        "success_rate": f"{total_success}/{total} ({...}%)",
        "avg_latency_ms": round(avg_latency, 1),
        "cases": results,
    }

xychart-beta
    title "各测试场景延迟对比（ms）"
    x-axis ["file_read", "calculator", "file_search", "multi_tool", "direct_answer", "table_analyzer"]
    y-axis "延迟 (ms)" 0 --> 10000
    bar [1574.5, 9222.5, 3592.3, 7361.7, 769.7, 1724.8]

xychart-beta
    title "各场景工具调用次数"
    x-axis ["file_read", "calculator", "file_search", "multi_tool", "direct_answer", "table_analyzer"]
    y-axis "tool_calls 数" 0 --> 4
    bar [1, 0, 1, 3, 0, 1]

xychart-beta
    title "两种传参方式成功率对比"
    x-axis ["prompt_json", "builtin"]
    y-axis "成功率 (%)" 0 --> 100
    bar [83.3, 0]

def _candidate_to_message(candidate: dict) -> tuple[dict, dict]:
    ...
    content = candidate.get("content", "")
    tool_calls = candidate.get("tool_calls", [])

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

    message = {
        "role": "assistant",
        "content": candidate.get("content", ""),
        "tool_calls": candidate.get("tool_calls", []),
        "content": content,
        "tool_calls": tool_calls,
    }
    validate_ai_message(message)
    has_content = bool(message["content"].strip())
    has_tool_calls = bool(message["tool_calls"])
    if has_content == has_tool_calls:
        raise ValueError(
            "model output must contain either final content or tool calls, but not both"
        )
    # 移除互斥检查（因为已经规范化）

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

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