跳转至

12.Agent 工具调用解析器

实现一个函数，解析 LLM 输出的文本，提取其中的 JSON 函数调用：输入包含 json {...} 代码块的字符串，返回解析后的 JSON 对象（字典），若解析失败则抛出异常。

原理

LLM 经常以 Markdown 代码块的形式输出 JSON，例如：

Here is the function call:

{"name": "search", "arguments": {"query": "weather"}}

我们需要提取代码块内的 JSON 字符串，并解析为 Python 字典。实现时需考虑以下情况：
- 标准 ````json ... ``` ` 代码块。
- 代码块中可能包含转义字符、多余空白。
- LLM 可能直接输出 JSON 对象而不包含代码块标记。
- JSON 解析可能失败（格式错误），此时应抛出异常并包含明确的错误信息。

**实现**

```python
import json
import re
from typing import Dict, Any

def extract_json_from_text(text: str) -> Dict[str, Any]:
    """
    从 LLM 输出的文本中提取 JSON 函数调用。
    支持 ```json``` 代码块或直接 JSON 对象。
    Args:
        text: LLM 原始输出字符串
    Returns:
        dict: 解析后的 JSON 对象
    Raises:
        ValueError: 未找到 JSON 或解析失败
    """
    # 1. 尝试提取 ```json ... ``` 代码块
    # 使用非贪婪匹配，支持多行
    pattern = r'```json\s*([\s\S]*?)\s*```'
    match = re.search(pattern, text)
    if match:
        json_str = match.group(1).strip()
        try:
            return json.loads(json_str)
        except json.JSONDecodeError as e:
            raise ValueError(f"JSON 解析失败: {e}\n内容: {json_str}")

    # 2. 如果没有代码块，尝试寻找第一个 '{' 和最后一个 '}' 之间的内容
    # 这是一个简单启发式，适用于 LLM 直接输出 JSON 对象的情况
    start = text.find('{')
    end = text.rfind('}')
    if start != -1 and end != -1 and end > start:
        json_str = text[start:end+1]
        try:
            return json.loads(json_str)
        except json.JSONDecodeError as e:
            raise ValueError(f"JSON 解析失败: {e}\n内容: {json_str}")

    # 3. 都没找到，抛出异常
    raise ValueError("未找到有效的 JSON 对象")

# 示例测试
if __name__ == "__main__":
    # 测试1: 代码块
    text1 = '好的，调用如下：\n```json\n{"name": "get_weather", "arguments": {"city": "北京"}}\n```'
    print(extract_json_from_text(text1))

    # 测试2: 直接 JSON
    text2 = '{"name": "search", "arguments": {"query": "AI"}}'
    print(extract_json_from_text(text2))

    # 测试3: 无效 JSON 应抛出异常
    try:
        extract_json_from_text("这不是 JSON {name: test}")
    except ValueError as e:
        print("捕获异常:", e)

关键点

  • 使用正则 re.search 非贪婪匹配代码块，[\s\S] 匹配包括换行在内的所有字符。

  • 如果代码块解析失败，回退到查找首尾花括号，适用于 LLM 可能遗漏代码块标记的场景。

  • 抛出的异常应包含原始字符串片段，便于调试。

实现一个支持流式输出的 JSON 函数调用解析器：模拟逐 token 到达的字符串，当检测到完整的 JSON 对象（括号匹配）时即刻解析并返回，剩余部分留待后续处理。

原理

在流式场景中，LLM 的响应 token 逐步到达。我们需要一个状态机跟踪括号匹配，一旦检测到最外层的大括号配对完成（即深度归零），说明一个完整的 JSON 对象已接收完毕，可以立即提取并解析该 JSON，同时将后续内容保留在缓冲区中供下一轮继续。

实现

import json
from typing import Optional, Tuple

class StreamingJSONParser:
    def __init__(self):
        self.buffer = ""          # 累积的字符串
        self.in_string = False    # 是否在字符串内部
        self.escape = False       # 前一个字符是否为转义符
        self.brace_depth = 0      # 大括号嵌套深度

    def feed(self, chunk: str) -> Optional[dict]:
        """
        输入一个 token 或字符串片段，如果检测到完整 JSON 对象则返回解析结果，
        否则返回 None。
        Args:
            chunk: 新到达的文本片段
        Returns:
            dict 或 None
        """
        self.buffer += chunk
        # 遍历整个缓冲区，但由于我们只需要知道何时第一个完整的 JSON 结束，
        # 可以扫描跟踪深度
        # 注意：这里每次 feed 都会重新扫描整个 buffer，为简单但正确。
        # 实际性能优化可以只扫描新增部分并更新状态。
        depth = 0
        in_str = False
        escape = False
        start_idx = -1
        for i, ch in enumerate(self.buffer):
            if ch == '{' and not in_str:
                if depth == 0:
                    start_idx = i
                depth += 1
            elif ch == '}' and not in_str:
                depth -= 1
                if depth == 0 and start_idx != -1:
                    # 找到完整 JSON
                    json_str = self.buffer[start_idx:i+1]
                    try:
                        obj = json.loads(json_str)
                        # 将剩余部分保留在缓冲区
                        self.buffer = self.buffer[i+1:]
                        return obj
                    except json.JSONDecodeError:
                        # JSON 不合法，可能还未完成，继续等待
                        pass
            elif ch == '"' and not escape:
                in_str = not in_str
            elif ch == '\\' and not escape:
                escape = True
                continue
            escape = False
        return None

# 示例测试
if __name__ == "__main__":
    parser = StreamingJSONParser()
    # 模拟流式输入
    chunks = ['{"na', 'me": "search", "arg', 'uments": {"q', 'uery": "AI"}}', '额外的文本']
    for c in chunks:
        result = parser.feed(c)
        if result:
            print("解析到 JSON:", result)
            print("剩余缓冲区:", parser.buffer)

说明

  • 该解析器维护一个简单的状态机：跟踪是否在字符串内（in_str）以及转义状态，确保不会错误地将字符串内的花括号计入深度。

  • 每当深度从 1 降为 0 时，说明最外层 JSON 对象闭合，尝试解析。若解析失败（例如 JSON 格式尚不完整），则继续等待更多数据。

  • 解析成功后，将已使用的部分从缓冲区移除，剩余文本可继续处理后续 JSON 对象。

  • 此实现假设 JSON 对象只包含简单的嵌套，不处理 JSON 数组等，但对于函数调用场景足够。

编写一个工具调用验证器：给定一个解析出的函数调用字典和预定义的工具 Schema（包含函数名、参数类型和是否必填），验证函数名是否存在、必填参数是否齐全、参数类型是否正确，返回验证结果及错误信息。

原理

工具 Schema 格式通常为：

schemas = {
    "search": {
        "description": "搜索互联网",
        "parameters": {
            "query": {"type": "str", "required": True},
            "num_results": {"type": "int", "required": False}
        }
    }
}

验证器需要检查：

  1. 函数名是否在 Schema 中注册。

  2. 所有必填参数是否都已提供。

  3. 提供的每个参数的类型是否与 Schema 匹配。

  4. 不应出现未定义的参数（可选，可以忽略多余参数或报错）。

实现

from typing import Dict, Any, List, Tuple

def validate_tool_call(
    call: Dict[str, Any],
    schemas: Dict[str, Any]
) -> Tuple[bool, List[str]]:
    """
    验证函数调用是否符合 Schema。
    Args:
        call: 包含 "name" 和 "arguments" 字段的字典。
        schemas: 工具 Schema 字典，键为函数名。
    Returns:
        (is_valid, errors): 是否通过验证以及错误信息列表。
    """
    errors = []
    func_name = call.get("name")
    if not func_name:
        errors.append("函数调用缺少 'name' 字段")
        return False, errors

    schema = schemas.get(func_name)
    if not schema:
        errors.append(f"函数 '{func_name}' 未在 Schema 中定义")
        return False, errors

    params_schema = schema.get("parameters", {})
    arguments = call.get("arguments", {})

    # 检查必填参数
    for param_name, param_info in params_schema.items():
        if param_info.get("required", False):
            if param_name not in arguments:
                errors.append(f"缺少必填参数 '{param_name}'")

    # 检查参数类型
    type_map = {
        "str": str,
        "int": int,
        "float": float,
        "bool": bool,
        "list": list,
        "dict": dict
    }
    for param_name, value in arguments.items():
        if param_name not in params_schema:
            # 可选：忽略未知参数 或 报错
            # errors.append(f"提供了未定义的参数 '{param_name}'")
            continue
        expected_type = params_schema[param_name].get("type")
        if expected_type and expected_type in type_map:
            if not isinstance(value, type_map[expected_type]):
                errors.append(
                    f"参数 '{param_name}' 类型错误：期望 {expected_type}，实际 {type(value).__name__}"
                )

    return len(errors) == 0, errors

# 示例
schemas = {
    "search": {
        "parameters": {
            "query": {"type": "str", "required": True},
            "max_results": {"type": "int", "required": False}
        }
    }
}

call1 = {"name": "search", "arguments": {"query": "天气"}}
call2 = {"name": "search", "arguments": {"max_results": 5}}
call3 = {"name": "unknown", "arguments": {}}
call4 = {"name": "search", "arguments": {"query": 123}}

print(validate_tool_call(call1, schemas))  # (True, [])
print(validate_tool_call(call2, schemas))  # (False, [...])
print(validate_tool_call(call3, schemas))  # (False, [...])
print(validate_tool_call(call4, schemas))  # (False, [...])

扩展

  • 支持 enum 约束、参数范围检查等。

  • 可根据 description 字段生成更友好的错误提示。

设计一个函数，将自然语言描述的工具列表（如“search(query: str) -> str: 搜索互联网”）转换为 JSON Schema 格式，支持提取函数名、参数名、参数类型、描述和返回值类型。

原理

输入的自然语言描述具有一定的模式，例如：

search(query: str, num: int) -> str: 搜索互联网

我们需要使用正则表达式提取函数签名中的函数名、参数列表、返回值类型，以及冒号后的描述。

实现

import re
from typing import Dict, List

def parse_tool_description(description: str) -> Dict[str, Any]:
    """
    解析自然语言工具描述，返回 JSON Schema 格式。
    示例输入: "search(query: str, max_results: int) -> dict: 搜索互联网并返回结果"
    Args:
        description: 工具描述字符串
    Returns:
        schema 字典
    """
    # 正则表达式匹配函数签名
    pattern = r'(\w+)\s*\(([^)]*)\)\s*->\s*(\w+)\s*:\s*(.*)'
    match = re.match(pattern, description.strip())
    if not match:
        raise ValueError(f"无法解析工具描述: {description}")

    func_name = match.group(1)
    params_str = match.group(2)
    return_type = match.group(3)
    desc = match.group(4).strip()

    parameters = {}
    if params_str.strip():
        # 分割参数
        for param in params_str.split(','):
            param = param.strip()
            if not param:
                continue
            # 每个参数格式: name: type
            parts = param.split(':')
            if len(parts) != 2:
                raise ValueError(f"参数格式错误: {param}")
            param_name = parts[0].strip()
            param_type = parts[1].strip()
            parameters[param_name] = {
                "type": param_type,
                "description": ""
            }

    return {
        "name": func_name,
        "description": desc,
        "parameters": parameters,
        "return_type": return_type
    }

# 示例
desc1 = "search(query: str, max_results: int) -> list: 搜索互联网并返回结果列表"
print(parse_tool_description(desc1))

desc2 = "get_weather(city: str) -> str: 获取指定城市的天气信息"
print(parse_tool_description(desc2))

扩展

  • 可以进一步解析参数是否可选（例如 name: str = None）或默认值。

  • 支持复杂的参数类型（如 list[int]）。

实现一个简单的工具注册表类：支持注册工具（函数名映射到可调用对象），根据函数名字符串动态调用对应函数，并捕获执行异常返回错误消息。

实现

from typing import Callable, Dict, Any

class ToolRegistry:
    def __init__(self):
        self._tools: Dict[str, Callable] = {}

    def register(self, name: str, func: Callable):
        """注册一个工具函数"""
        self._tools[name] = func

    def call(self, name: str, *args, **kwargs) -> Any:
        """
        根据名称调用工具函数。
        如果函数不存在，抛出 ValueError。
        如果执行出错，捕获异常并返回错误信息字符串。
        """
        if name not in self._tools:
            raise ValueError(f"工具 '{name}' 未注册")
        try:
            return self._tools[name](*args, **kwargs)
        except Exception as e:
            return f"工具执行错误: {type(e).__name__}: {str(e)}"

    def list_tools(self) -> list:
        """列出所有已注册的工具名称"""
        return list(self._tools.keys())

# 示例
def search(query: str, max_results: int = 10) -> str:
    return f"搜索 '{query}' 的前 {max_results} 条结果"

registry = ToolRegistry()
registry.register("search", search)

print(registry.call("search", query="AI", max_results=5))
# 调用未注册工具
try:
    registry.call("unknown")
except ValueError as e:
    print(e)
# 模拟异常
def faulty():
    raise RuntimeError("网络错误")
registry.register("faulty", faulty)
print(registry.call("faulty"))

要点

  • register 可以接受任意可调用对象，包括 lambda、类方法等。

  • call 方法返回正常结果，或捕获异常并返回描述性错误字符串，便于 LLM 根据错误信息调整后续行为。

模拟多工具选择：给定用户查询字符串和多个工具的 Schema 描述，编写一个基于规则（关键词匹配）的工具选择器，返回最相关的工具名称。

原理

简单的工具选择可以通过关键词匹配实现：预先为每个工具定义一组触发关键词，当用户查询中包含这些关键词时，选择对应的工具。还可以计算匹配关键词的数量或权重来决定最佳工具。

实现

from typing import List, Dict

class KeywordToolSelector:
    def __init__(self, tool_keywords: Dict[str, List[str]]):
        """
        tool_keywords: 工具名 -> 关键词列表
        示例: {"search": ["搜索", "查找", "百度"], "weather": ["天气", "气温", "下雨"]}
        """
        self.tool_keywords = tool_keywords

    def select(self, query: str) -> str:
        """
        根据查询选择最相关的工具。
        返回工具名，若无匹配则返回 None 或默认工具。
        """
        best_tool = None
        best_score = 0
        for tool, keywords in self.tool_keywords.items():
            score = 0
            for kw in keywords:
                if kw in query:
                    # 可以根据关键词长度或重要性加权
                    score += 1
            if score > best_score:
                best_score = score
                best_tool = tool
        return best_tool

# 示例
selector = KeywordToolSelector({
    "search": ["搜索", "百度", "谷歌", "查找", "查询"],
    "weather": ["天气", "气温", "下雨", "晴天", "温度"],
    "translate": ["翻译", "英文", "中文", "转换语言"]
})

print(selector.select("帮我搜索一下最新的 AI 新闻"))    # search
print(selector.select("今天北京天气怎么样"))           # weather
print(selector.select("把这句话翻译成英文"))           # translate
print(selector.select("计算 1+1"))                    # None

扩展

  • 可以结合语义相似度（如使用词向量）提升匹配准确性。

  • 多个工具都匹配时，可以返回排序列表供后续流程（如让 LLM 二次确认）使用。

实现 ReAct 循环中 Action 部分的解析器：输入 LLM 的输出文本，要求能提取出 “Action: tool_name\nAction Input: {...}” 格式的内容，返回工具名和参数字典。

原理 在 ReAct 框架中，LLM 的响应通常包含 Action: 和 Action Input: 字段。我们需要从可能包含多余文本的输出中精确提取这两个字段。解析器需支持：

  • Action: 后跟工具名，直到行尾。

  • Action Input: 后跟 JSON 对象（可能跨行，也可能在同一行）。

  • 健壮处理空白、换行以及可能的 Markdown 格式。

实现

import re
import json
from typing import Tuple, Dict

def parse_react_action(text: str) -> Tuple[str, Dict]:
    """
    从 ReAct 输出中提取 Action 工具名和参数字典。
    格式要求：Action: tool_name\nAction Input: {...}
    Args:
        text: LLM 的原始输出字符串
    Returns:
        (tool_name, arguments_dict)
    Raises:
        ValueError: 解析失败
    """
    # 提取 Action 行
    action_match = re.search(r'Action:\s*(.+?)\s*(?:\n|$)', text)
    if not action_match:
        raise ValueError("未找到 'Action:' 字段")
    tool_name = action_match.group(1).strip()

    # 提取 Action Input：可能是一行 JSON，也可能是多行直到下一个 "Action:" 或文本结束
    # 这里采用从 "Action Input:" 开始提取直到遇到新的 "Action" 或结束
    input_pattern = r'Action Input:\s*([\s\S]*?)(?=\n\s*Action:|\Z)'
    input_match = re.search(input_pattern, text)
    if not input_match:
        raise ValueError("未找到 'Action Input:' 字段")
    input_str = input_match.group(1).strip()

    # 尝试解析 JSON
    # 移除可能的 Markdown 代码块包裹
    if input_str.startswith('```json'):
        input_str = input_str[7:]
        if input_str.endswith('```'):
            input_str = input_str[:-3]
    elif input_str.startswith('```'):
        if input_str.endswith('```'):
            input_str = input_str[3:-3]

    input_str = input_str.strip()
    try:
        arguments = json.loads(input_str)
    except json.JSONDecodeError:
        # 尝试在输入中寻找 JSON 对象
        start = input_str.find('{')
        end = input_str.rfind('}')
        if start != -1 and end != -1:
            try:
                arguments = json.loads(input_str[start:end+1])
            except json.JSONDecodeError as e:
                raise ValueError(f"Action Input JSON 解析失败: {e}")
        else:
            raise ValueError("Action Input 不包含有效的 JSON 对象")
    return tool_name, arguments

# 示例
text1 = """
Thought: 我需要搜索天气。
Action: search_weather
Action Input: {"city": "北京", "date": "2025-01-01"}
"""
tool, args = parse_react_action(text1)
print(tool, args)  # search_weather {'city': '北京', 'date': '2025-01-01'}

# 处理多行JSON
text2 = """
Action: query_database
Action Input: {
    "table": "users",
    "columns": ["name", "age"]
}
"""
tool2, args2 = parse_react_action(text2)
print(tool2, args2)

说明

  • 使用正则 [\s\S]*? 进行非贪婪匹配，(?=\n\s*Action:|\Z) 前瞻断言防止跨 Action 提取。

  • 对 Action Input 尝试多种 JSON 提取方式，增强健壮性。

编写一个参数提取器：给定用户自然语言指令和工具的参数定义，利用正则表达式或简单规则从指令中抽取参数值（如“搜索'天气'” → query="天气"）。

原理 参数提取是工具调用的关键前置步骤。我们可以为每个参数预定义一组提取规则（如正则模式、关键词定位等）。例如，工具 search 有一个参数 query，可以设计规则：匹配引号内的内容，或者匹配“搜索xxx”中的 xxx。

实现

import re
from typing import Dict, Any

class ParamExtractor:
    def __init__(self):
        # 为每个工具定义参数提取规则：参数名 -> 正则模式
        self.rules = {
            "search": {
                "query": r'搜索["\u201c](.+?)["\u201d]|搜索(.+?)(?:$|\n)|查找["\u201c](.+?)["\u201d]',
                "max_results": r'(\d+)条|前(\d+)条'
            },
            "weather": {
                "city": r'(?:城市|地点|地方)[：:]\s*(.+?)(?:$|\n|，|。)',
                "date": r'(\d{4}-\d{2}-\d{2})'
            }
        }

    def extract(self, tool_name: str, user_input: str) -> Dict[str, Any]:
        """根据工具名和用户输入提取参数字典"""
        if tool_name not in self.rules:
            return {}
        params = {}
        for param, pattern in self.rules[tool_name].items():
            match = re.search(pattern, user_input)
            if match:
                # 取第一个非None的捕获组
                value = next((g for g in match.groups() if g is not None), None)
                if value:
                    params[param] = value.strip()
        return params

# 示例
extractor = ParamExtractor()
print(extractor.extract("search", "帮我搜索“人工智能最新进展”，显示前5条"))
# {'query': '人工智能最新进展', 'max_results': '5'}
print(extractor.extract("weather", "查询城市：上海 的天气，日期 2025-01-15"))
# {'city': '上海', 'date': '2025-01-15'}

扩展

  • 可以集成 NER 模型进行更精准的实体抽取。

  • 参数类型转换：将提取的字符串转为 int、float 等。

设计一个工具调用重试机制：如果工具执行返回错误，根据错误类型（如超时、参数错误）决定是否重试，并限制最大重试次数，实现该逻辑。

原理

工具调用可能因网络波动、暂时性故障或参数错误而失败。重试机制需区分可重试错误（如超时、服务不可用）和不可重试错误（如参数非法、权限不足），并设置最大重试次数和重试间隔。

实现

import time
import functools
from typing import Callable, Any

class RetryableError(Exception):
    """可重试的异常基类"""
    pass

class TimeoutError(RetryableError):
    pass

class TemporaryError(RetryableError):
    pass

class NonRetryableError(Exception):
    """不可重试的异常"""
    pass

def call_tool_with_retry(
    tool_func: Callable,
    *args,
    max_retries: int = 3,
    delay: float = 1.0,
    backoff: float = 2.0,
    **kwargs
) -> Any:
    """
    带重试逻辑的工具调用包装器。
    Args:
        tool_func: 要调用的工具函数
        max_retries: 最大重试次数（不包括首次调用）
        delay: 初始重试延迟（秒）
        backoff: 延迟倍增因子（指数退避）
    Returns:
        工具函数的返回值
    Raises:
        NonRetryableError: 不可重试的错误直接抛出
        RetryableError: 重试耗尽后仍失败
    """
    last_exception = None
    for attempt in range(max_retries + 1):
        try:
            return tool_func(*args, **kwargs)
        except NonRetryableError as e:
            # 不可重试，直接抛出
            raise e
        except RetryableError as e:
            last_exception = e
            if attempt < max_retries:
                sleep_time = delay * (backoff ** attempt)
                print(f"工具调用失败，{sleep_time:.1f}s 后重试 (尝试 {attempt+1}/{max_retries}): {e}")
                time.sleep(sleep_time)
            else:
                raise RuntimeError(f"重试{max_retries}次后仍失败: {e}")
        except Exception as e:
            # 未知异常，按不可重试处理
            raise NonRetryableError(f"未知错误: {e}")
    raise RuntimeError(f"重试{max_retries}次后仍失败: {last_exception}")

# 示例模拟
def unreliable_search(query):
    import random
    if random.random() < 0.7:
        raise TimeoutError("搜索超时")
    return f"搜索结果: {query}"

try:
    result = call_tool_with_retry(unreliable_search, "天气", max_retries=3, delay=0.5)
    print(result)
except Exception as e:
    print("最终失败:", e)

要点

  • 使用指数退避策略缓解服务端压力。

  • 记录每次重试的日志，便于调试。

  • 可集成到工具注册表中，为每个工具单独配置重试策略。

实现一个并行工具调用调度器：接收一个 LLM 返回的多个工具调用请求（列表），并发执行这些工具调用（可用多线程模拟），等待所有完成后收集结果。

原理

当 LLM 返回多个独立的工具调用（如并行搜索多个关键词），我们可以利用多线程并发执行以降低总延迟。调度器需要管理线程池、收集结果（包括异常），并维持调用顺序以便后续处理。

实现

import concurrent.futures
from typing import List, Dict, Any

class ParallelToolScheduler:
    def __init__(self, tool_registry: Dict[str, Callable], max_workers: int = 5):
        """
        tool_registry: 函数名到可调用对象的映射
        """
        self.registry = tool_registry
        self.max_workers = max_workers

    def execute_parallel(self, tool_calls: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """
        tool_calls: 列表，每个元素包含 "name" 和 "arguments"
        返回: 列表，每个元素包含 "name", "arguments", "result" 或 "error"
        """
        results = [None] * len(tool_calls)

        def run_one(idx, call):
            name = call["name"]
            args = call.get("arguments", {})
            try:
                if name not in self.registry:
                    raise ValueError(f"未注册的工具: {name}")
                result = self.registry[name](**args)
                return idx, {"name": name, "arguments": args, "result": result}
            except Exception as e:
                return idx, {"name": name, "arguments": args, "error": str(e)}

        with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            futures = [
                executor.submit(run_one, i, call) for i, call in enumerate(tool_calls)
            ]
            for future in concurrent.futures.as_completed(futures):
                idx, res = future.result()
                results[idx] = res
        return results

# 示例
def search(query):
    import time, random
    time.sleep(random.random())  # 模拟耗时
    return f"搜索 '{query}' 的结果"

def calculator(expr):
    return eval(expr)

scheduler = ParallelToolScheduler({"search": search, "calculator": calculator})
calls = [
    {"name": "search", "arguments": {"query": "天气"}},
    {"name": "search", "arguments": {"query": "新闻"}},
    {"name": "calculator", "arguments": {"expr": "3*15"}}
]
res = scheduler.execute_parallel(calls)
for r in res:
    print(r)

说明

  • 使用 ThreadPoolExecutor 实现并发，适用于 I/O 密集型工具调用。

  • 每个调用分配唯一索引，结果按原始顺序存储。

  • 异常被捕获为 error 字段，不中断其他调用。

编写函数，将工具调用的结果序列化为 LLM 可理解的文本形式，支持截断过长结果，并加上标记（如 “Observation: ...”）。

原理 工具返回的结果可能是长文本、JSON 或列表。需要将其格式化为简洁且信息密集的文本，并加上 ReAct 的 Observation: 标记。对于过长结果，应进行截断并提示，避免撑满 LLM 上下文窗口。

实现

def serialize_tool_result(result: Any, max_length: int = 500) -> str:
    """
    将工具调用结果序列化为 Observation 文本。
    Args:
        result: 工具返回的任意对象
        max_length: 最大字符数，超出则截断
    Returns:
        str: "Observation: ..."
    """
    if isinstance(result, str):
        text = result
    else:
        import json
        text = json.dumps(result, ensure_ascii=False, indent=2)

    if len(text) > max_length:
        truncated = text[:max_length] + f"\n... (结果过长，已截断，原长度 {len(text)} 字符)"
        return f"Observation: {truncated}"
    else:
        return f"Observation: {text}"

# 示例
print(serialize_tool_result({"status": "ok", "data": list(range(100))}, max_length=100))

扩展

  • 针对特定工具可定制序列化逻辑（如表格转 Markdown）。

  • 可加入时间戳或来源标记。

处理工具调用中的参数依赖：给定两个工具调用，后一个调用的参数引用了前一个的输出，实现一个简单的模板替换（如 “{{tool1.result}}”）来填充参数。

原理 在 Agent 流水线中，后一个工具的输入可能依赖前一个工具的输出。我们使用模板字符串（如 {{tool1.result}}、{{tool1.result.key}}）在运行时进行替换。

实现

import re
from typing import Any, Dict

def resolve_parameter_dependencies(
    params: Dict[str, Any],
    context: Dict[str, Any]
) -> Dict[str, Any]:
    """
    解析参数中的模板引用，替换为上下文中的值。
    模板格式：{{tool_name.result}} 或 {{tool_name.result.path.to.key}}
    Args:
        params: 包含模板字符串的参数字典
        context: {"tool_name": {"result": ...}} 前序工具的执行上下文
    Returns:
        解析后的参数字典
    """
    pattern = re.compile(r'\{\{(.+?)\}\}')

    def replace(value):
        if isinstance(value, str):
            def replacer(m):
                path = m.group(1).strip()
                parts = path.split('.')
                obj = context
                for part in parts:
                    if isinstance(obj, dict):
                        obj = obj.get(part)
                    else:
                        obj = getattr(obj, part, None)
                    if obj is None:
                        break
                return str(obj) if obj is not None else m.group(0)
            return pattern.sub(replacer, value)
        elif isinstance(value, dict):
            return {k: replace(v) for k, v in value.items()}
        elif isinstance(value, list):
            return [replace(v) for v in value]
        return value

    return replace(params)

# 示例
context = {
    "search1": {"result": {"title": "AI", "url": "http://example.com"}},
    "search2": {"result": "Beijing"}
}
params = {
    "url": "{{search1.result.url}}",
    "city": "{{search2.result}}"
}
print(resolve_parameter_dependencies(params, context))
# {'url': 'http://example.com', 'city': 'Beijing'}

说明

  • 支持嵌套对象属性访问（如 .result.url）。

  • 若路径无法解析，保留原模板字符串以便调试。

实现一个对话历史中的工具调用记录管理器：能存储每次调用的工具名、参数、结果，并能根据调用顺序生成 Markdown 格式的摘要。

实现

from datetime import datetime
from typing import List, Dict, Any

class ToolCallLogger:
    def __init__(self):
        self.records: List[Dict[str, Any]] = []

    def log(self, tool_name: str, params: Dict, result: Any):
        self.records.append({
            "timestamp": datetime.now().isoformat(),
            "tool": tool_name,
            "params": params,
            "result": result
        })

    def generate_summary(self) -> str:
        """生成 Markdown 格式的调用摘要"""
        lines = ["# 工具调用记录\n"]
        for i, rec in enumerate(self.records, 1):
            lines.append(f"## 调用 {i}: {rec['tool']}")
            lines.append(f"- **时间**: {rec['timestamp']}")
            lines.append(f"- **参数**: {json.dumps(rec['params'], ensure_ascii=False)}")
            result_str = json.dumps(rec['result'], ensure_ascii=False, indent=2)
            # 截断过长结果
            if len(result_str) > 200:
                result_str = result_str[:200] + "..."
            lines.append(f"- **结果**: \n```\n{result_str}\n```")
            lines.append("")
        return "\n".join(lines)

# 示例
logger = ToolCallLogger()
logger.log("search", {"query": "天气"}, {"temp": 25, "city": "北京"})
logger.log("calculator", {"expr": "1+2"}, 3)
print(logger.generate_summary())

模拟 Function Calling 的完整流程：输入用户消息，LLM 返回“需要调用工具”的决策及工具调用 JSON，解析后执行工具，将结果追加回对话，生成最终回复（使用伪代码或模拟函数）。

实现（模拟脚本）

def simulate_function_calling(user_message: str, tools_schema: dict, llm_fn, tool_registry: dict) -> str:
    """
    模拟完整的 Function Calling 流程。
    llm_fn: 模拟的 LLM 函数，接受对话历史，返回决策（"tool_call" 或 "final_answer"）及内容。
    """
    # 初始对话
    conversation = [{"role": "user", "content": user_message}]

    while True:
        # 调用 LLM
        response = llm_fn(conversation, tools_schema)
        if response["decision"] == "final_answer":
            return response["content"]

        elif response["decision"] == "tool_call":
            tool_call = response["tool_call"]
            tool_name = tool_call["name"]
            args = tool_call["arguments"]
            # 执行工具
            try:
                result = tool_registry[tool_name](**args)
                result_str = serialize_tool_result(result)
            except Exception as e:
                result_str = f"Observation: 工具执行错误 - {e}"
            # 追加工具调用和结果到对话历史
            conversation.append({"role": "assistant", "content": f"调用工具: {tool_name}({args})"})
            conversation.append({"role": "tool", "content": result_str})
        else:
            raise ValueError(f"未知决策类型: {response['decision']}")

# 模拟 LLM 函数（固定逻辑演示）
def mock_llm(conversation, tools_schema):
    # 最后一轮如果是用户消息且包含"天气"，则决定调用 get_weather
    last_msg = conversation[-1]["content"] if conversation else ""
    if "天气" in last_msg and not any(turn["role"] == "tool" for turn in conversation):
        return {
            "decision": "tool_call",
            "tool_call": {"name": "get_weather", "arguments": {"city": "北京"}}
        }
    else:
        return {"decision": "final_answer", "content": "北京今天晴天，25°C。"}

# 工具注册
def get_weather(city):
    return f"{city}晴天，25°C"

registry = {"get_weather": get_weather}
schema = {}  # 简化

# 运行模拟
final = simulate_function_calling("北京天气怎么样？", schema, mock_llm, registry)
print(final)

说明

  • 通过循环实现多轮工具调用，直到 LLM 返回最终答案。

  • 每次工具执行结果以 Observation 形式追加，符合 ReAct 风格。

  • 实际系统中，LLM 决策由模型自身完成，这里以 mock 函数模拟。

实现一个工具调用的超时控制装饰器，可设置最大执行时间，超时则抛出指定异常并返回预设错误信息。

实现

import signal
import functools
from typing import Any, Callable

class ToolTimeoutError(Exception):
    pass

def with_timeout(timeout_sec: int):
    """
    装饰器，为函数调用添加超时控制。
    Args:
        timeout_sec: 超时秒数
    Returns:
        装饰后的函数，超时时抛出 ToolTimeoutError
    """
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            # 使用 signal 仅适用于 Unix 主线程，若跨平台可改用 concurrent.futures
            def handler(signum, frame):
                raise ToolTimeoutError(f"工具 '{func.__name__}' 执行超时 ({timeout_sec}s)")

            old_handler = signal.signal(signal.SIGALRM, handler)
            signal.alarm(timeout_sec)
            try:
                result = func(*args, **kwargs)
            finally:
                signal.alarm(0)  # 取消闹钟
                signal.signal(signal.SIGALRM, old_handler)
            return result
        return wrapper
    return decorator

# 跨平台版本（使用 concurrent.futures）
def with_timeout_thread(timeout_sec: int):
    """线程版超时装饰器，适用于所有平台，但开销稍大"""
    import concurrent.futures
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
                future = executor.submit(func, *args, **kwargs)
                try:
                    return future.result(timeout=timeout_sec)
                except concurrent.futures.TimeoutError:
                    raise ToolTimeoutError(f"工具 '{func.__name__}' 执行超时 ({timeout_sec}s)")
        return wrapper
    return decorator

# 示例
@with_timeout(2)
def slow_search(query):
    import time
    time.sleep(3)
    return f"结果: {query}"

try:
    print(slow_search("天气"))
except ToolTimeoutError as e:
    print("超时错误:", e)

说明

  • Unix 下使用 signal.alarm 实现低开销超时，但不适用于多线程。

  • 跨平台方案使用 ThreadPoolExecutor + future.result(timeout)，更通用但线程开销略高。

  • 超时异常可被上层重试机制捕获并处理。

构建一个支持嵌套工具调用的解析器：工具返回的结果中可能包含新的工具调用请求，递归处理直到没有进一步调用为止。

原理

在某些 Agent 场景中，一个工具的执行结果可能是一个全新的工具调用请求（例如，工具返回“需要进一步查询XX”）。我们需要一个递归解析器，能够在每次工具执行后检查返回值，如果返回值本身包含有效的工具调用（符合 Action/Action Input 格式或包含 JSON 代码块），则自动提取并执行，如此递归，直到返回值不再包含工具调用为止。为防止无限递归，必须设置最大递归深度。

实现

import re
import json
from typing import Dict, Any, Callable

def recursive_tool_executor(
    initial_text: str,
    tool_registry: Dict[str, Callable],
    parse_func: Callable[[str], Dict],
    max_depth: int = 5
) -> Any:
    """
    递归执行嵌套工具调用。
    Args:
        initial_text: 包含初始工具调用的文本
        tool_registry: 工具函数映射
        parse_func: 从文本中提取工具调用信息的函数，返回 {"name":..., "arguments":...} 或 None
        max_depth: 最大递归深度
    Returns:
        最终结果（非工具调用时返回原文本）
    """
    depth = 0
    current_text = initial_text
    while depth < max_depth:
        # 尝试从当前文本中解析工具调用
        try:
            call = parse_func(current_text)
        except Exception:
            call = None
        if call is None:
            # 没有工具调用，返回当前文本
            return current_text
        tool_name = call.get("name")
        arguments = call.get("arguments", {})
        if tool_name not in tool_registry:
            # 工具不存在，返回当前文本或抛出异常
            return current_text
        # 执行工具
        try:
            result = tool_registry[tool_name](**arguments)
        except Exception as e:
            result = f"工具执行错误: {e}"
        # 将结果转换为字符串（若为字典等，转为JSON字符串）
        if not isinstance(result, str):
            result = json.dumps(result, ensure_ascii=False)
        current_text = result
        depth += 1
    # 达到最大深度，返回当前文本
    return current_text

# 示例工具
def search(query):
    # 模拟返回一个新的工具调用
    return '需要进一步查询详细信息：\n```json\n{"name": "get_details", "arguments": {"topic": "' + query + '"}}\n```'

def get_details(topic):
    return f"{topic}的详细信息：这是关于{topic}的详细内容。"

# 解析函数：复用前面的 parse_react_action
def parse_action(text):
    try:
        return parse_react_action(text)
    except Exception:
        # 尝试直接查找 JSON
        start = text.find('{')
        end = text.rfind('}')
        if start != -1 and end != -1:
            try:
                obj = json.loads(text[start:end+1])
                if "name" in obj and "arguments" in obj:
                    return obj
            except:
                pass
    return None

registry = {"search": search, "get_details": get_details}
initial = 'Action: search\nAction Input: {"query": "AI"}'
final_result = recursive_tool_executor(initial, registry, parse_action, max_depth=3)
print(final_result)  # 输出详细信息

说明

  • 每次工具执行后，将其结果作为下一轮解析的输入，判断是否再次包含工具调用。

  • 设置最大深度防止无限循环。

  • 解析函数可以支持多种格式，增强健壮性。

为工具参数实现默认值填充：根据工具 Schema 中的 default 字段，在解析出的参数缺失时自动补齐默认值。

原理

在工具定义中，可以为参数指定 default 值。当 LLM 返回的 arguments 缺少某些参数时，系统自动使用 Schema 中的默认值进行填充，确保工具调用不会因参数缺失而失败。Schema 格式如：

{
  "parameters": {
    "query": {"type": "str", "required": true},
    "max_results": {"type": "int", "default": 10}
  }
}

实现

def fill_default_arguments(arguments: Dict[str, Any], schema: Dict[str, Any]) -> Dict[str, Any]:
    """
    根据 Schema 自动填充缺失参数的默认值。
    Args:
        arguments: 解析出的参数字典
        schema: 工具 Schema，包含 parameters 字段
    Returns:
        填充后的参数字典
    """
    filled = arguments.copy()
    params_schema = schema.get("parameters", {})
    for param_name, param_info in params_schema.items():
        if param_name not in filled and "default" in param_info:
            filled[param_name] = param_info["default"]
    return filled

# 示例 Schema
schema = {
    "name": "search",
    "parameters": {
        "query": {"type": "str", "required": True},
        "max_results": {"type": "int", "default": 10},
        "safe_search": {"type": "bool", "default": True}
    }
}

# 用户只提供了 query
parsed_args = {"query": "天气"}
filled_args = fill_default_arguments(parsed_args, schema)
print(filled_args)  # {'query': '天气', 'max_results': 10, 'safe_search': True}

扩展

  • 可以结合参数类型校验（参见第3题），在填充默认值后再次验证。

  • 支持嵌套 Schema 中的默认值填充（递归处理）。

实现一个简单的工具调用日志系统，记录每次调用的时间戳、工具名、参数、结果、耗时，并支持按时间或工具名过滤查询。

实现

import time
from datetime import datetime
from typing import List, Dict, Any, Optional

class ToolLogger:
    def __init__(self):
        self.records: List[Dict[str, Any]] = []

    def log(self, tool_name: str, params: Dict, result: Any, duration: float):
        record = {
            "timestamp": datetime.now().isoformat(),
            "tool": tool_name,
            "params": params,
            "result": result,
            "duration": duration
        }
        self.records.append(record)

    def query(
        self,
        tool_name: Optional[str] = None,
        start_time: Optional[str] = None,
        end_time: Optional[str] = None
    ) -> List[Dict]:
        """按条件过滤记录"""
        results = self.records
        if tool_name:
            results = [r for r in results if r["tool"] == tool_name]
        if start_time:
            results = [r for r in results if r["timestamp"] >= start_time]
        if end_time:
            results = [r for r in results if r["timestamp"] <= end_time]
        return results

    def summary(self) -> str:
        """生成简要统计信息"""
        lines = ["工具调用统计:"]
        tool_counts = {}
        total_duration = 0.0
        for r in self.records:
            tool_counts[r["tool"]] = tool_counts.get(r["tool"], 0) + 1
            total_duration += r["duration"]
        lines.append(f"总调用次数: {len(self.records)}, 总耗时: {total_duration:.2f}s")
        for tool, count in tool_counts.items():
            lines.append(f"  {tool}: {count}次")
        return "\n".join(lines)

# 示例
logger = ToolLogger()
logger.log("search", {"query": "天气"}, "搜索结果...", 0.5)
logger.log("search", {"query": "AI"}, "搜索结果...", 0.3)
logger.log("calculator", {"expr": "1+1"}, 2, 0.01)

print(logger.query(tool_name="search"))
print(logger.summary())

说明

  • log 方法通常在工具调用的包装器（如第15题的超时装饰器）中调用，自动记录耗时。

  • 查询支持多种组合过滤，方便调试和性能分析。

  • 可以扩展为将日志持久化到文件或数据库。

实现工具调用权限检查：定义每个工具的危险等级（如 safe, dangerous），根据当前会话的安全策略决定是否允许执行，如果不允许则返回拒绝理由。

实现

from enum import Enum

class RiskLevel(Enum):
    SAFE = "safe"
    DANGEROUS = "dangerous"

class ToolSecurityPolicy:
    def __init__(self, risk_assignments: Dict[str, RiskLevel], default_allow_dangerous: bool = False):
        self.risk = risk_assignments
        self.allow_dangerous = default_allow_dangerous

    def check(self, tool_name: str) -> Tuple[bool, str]:
        """
        检查工具是否允许执行。
        Returns:
            (allowed, reason)
        """
        level = self.risk.get(tool_name, RiskLevel.DANGEROUS)  # 默认危险
        if level == RiskLevel.SAFE:
            return True, ""
        if self.allow_dangerous:
            return True, "允许执行危险操作"
        return False, f"工具 '{tool_name}' 被标记为危险操作，当前安全策略不允许执行"

# 示例
policy = ToolSecurityPolicy({
    "search": RiskLevel.SAFE,
    "delete_file": RiskLevel.DANGEROUS,
    "send_email": RiskLevel.DANGEROUS
}, default_allow_dangerous=False)

print(policy.check("search"))       # (True, '')
print(policy.check("delete_file")) # (False, "工具 'delete_file' 被标记为危险操作...")

扩展

  • 可结合用户角色/认证信息动态调整策略。

  • 危险操作可以要求用户二次确认（如通过回调函数）。

实现一个可插拔的工具调用解析框架：允许注册不同的输出格式解析器（如 JSON、XML、自定义标记），根据 LLM 的输出格式动态选择解析器并提取函数调用。

实现

from abc import ABC, abstractmethod
from typing import Dict, Any, Optional

class ToolCallParser(ABC):
    @abstractmethod
    def parse(self, text: str) -> Optional[Dict[str, Any]]:
        """从文本中提取工具调用，返回 {"name": ..., "arguments": ...} 或 None"""
        pass

class JSONBlockParser(ToolCallParser):
    """解析 ```json ... ``` 代码块"""
    def parse(self, text: str) -> Optional[Dict]:
        import re, json
        pattern = r'```json\s*([\s\S]*?)\s*```'
        match = re.search(pattern, text)
        if match:
            try:
                obj = json.loads(match.group(1))
                if "name" in obj and "arguments" in obj:
                    return obj
            except:
                pass
        return None

class ReActParser(ToolCallParser):
    """解析 Action/Action Input 格式"""
    def parse(self, text: str) -> Optional[Dict]:
        # 复用前面的 parse_react_action 逻辑，此处简略
        import re, json
        action_match = re.search(r'Action:\s*(.+?)\s*(?:\n|$)', text)
        input_match = re.search(r'Action Input:\s*([\s\S]*?)(?=\n\s*Action:|\Z)', text)
        if action_match and input_match:
            tool_name = action_match.group(1).strip()
            input_str = input_match.group(1).strip()
            # 解析 JSON
            try:
                args = json.loads(input_str)
            except:
                # 尝试提取 JSON
                start = input_str.find('{')
                end = input_str.rfind('}')
                if start != -1 and end != -1:
                    args = json.loads(input_str[start:end+1])
                else:
                    return None
            return {"name": tool_name, "arguments": args}
        return None

class XMLParser(ToolCallParser):
    """解析 XML 格式 <tool_call name=".."><arg key="..">value</arg></tool_call>"""
    def parse(self, text: str) -> Optional[Dict]:
        import re
        match = re.search(r'<tool_call\s+name="(\w+)"\s*>(.*?)</tool_call>', text, re.DOTALL)
        if match:
            name = match.group(1)
            args_str = match.group(2)
            args = {}
            for arg_m in re.finditer(r'<arg\s+key="(\w+)"\s*>(.*?)</arg>', args_str):
                args[arg_m.group(1)] = arg_m.group(2)
            return {"name": name, "arguments": args}
        return None

class ParserRegistry:
    def __init__(self):
        self.parsers: Dict[str, ToolCallParser] = {}

    def register(self, name: str, parser: ToolCallParser):
        self.parsers[name] = parser

    def parse(self, text: str) -> Optional[Dict]:
        # 遍历所有解析器，返回第一个成功的结果
        for parser in self.parsers.values():
            result = parser.parse(text)
            if result is not None:
                return result
        return None

# 示例
registry = ParserRegistry()
registry.register("json", JSONBlockParser())
registry.register("react", ReActParser())
registry.register("xml", XMLParser())

text1 = 'Action: search\nAction Input: {"query": "天气"}'
text2 = '<tool_call name="search"><arg key="query">天气</arg></tool_call>'
text3 = '```json\n{"name": "search", "arguments": {"query": "天气"}}\n```'

print(registry.parse(text1))
print(registry.parse(text2))
print(registry.parse(text3))

说明

  • 每个解析器实现统一接口，可灵活扩展。

  • ParserRegistry 顺序尝试各解析器，第一个成功即返回。

  • 可加入自动检测逻辑（如根据文本特征选择解析器）提高效率。

编写解析器，处理 LLM 输出中可能出现的“幻想工具调用”：检测解析出的工具名是否在注册表中，若不存在，则尝试模糊匹配最相似的工具名并建议更正，否则报错。

原理

LLM 可能生成不存在的工具名（幻觉）。我们可以利用编辑距离（Levenshtein）找出注册表中与解析出的工具名最相似的工具，若相似度高于阈值，则自动更正或提示用户。否则报错。

实现

import Levenshtein  # python-Levenshtein 库，需安装，也可用自实现

def correct_tool_name(parsed_name: str, registry_names: List[str], threshold: float = 0.8) -> str:
    """
    尝试修正幻想工具名。
    Args:
        parsed_name: 解析出的工具名
        registry_names: 注册表中的所有工具名
        threshold: 相似度阈值（0~1），超过则自动更正
    Returns:
        更正后的工具名，若未找到匹配则抛出 ValueError
    """
    best_sim = 0.0
    best_name = None
    for name in registry_names:
        # 计算相似度：1 - 归一化编辑距离
        sim = 1 - Levenshtein.distance(parsed_name.lower(), name.lower()) / max(len(parsed_name), len(name))
        if sim > best_sim:
            best_sim = sim
            best_name = name
    if best_sim >= threshold:
        print(f"工具名 '{parsed_name}' 不存在，自动更正为 '{best_name}' (相似度 {best_sim:.2f})")
        return best_name
    else:
        # 列出可能候选
        candidates = sorted(registry_names, key=lambda n: Levenshtein.distance(parsed_name, n))[:3]
        raise ValueError(f"工具 '{parsed_name}' 未注册。您是否想使用: {candidates}?")

# 示例
registry = ["search", "get_weather", "send_email", "delete_file"]
print(correct_tool_name("searh", registry))   # 自动更正为 "search"
try:
    correct_tool_name("fly_to_moon", registry)
except ValueError as e:
    print(e)

说明

  • 使用 Levenshtein.distance 计算编辑距离，相似度阈值通常设为 0.8。

  • 若相似度不足，则提供最相似的候选项供用户或 LLM 确认。

  • 可集成到工具调用链中，在调用前自动修正，提高容错性。

实现一个支持多模态输入的工具调用：工具 Schema 中参数标注为 image，解析时检测输入是否为 Base64 或 URL，并转换为对应格式供工具使用。

原理

某些工具可能需要图像输入（如视觉问答、图像生成）。工具 Schema 中可定义参数类型为 image，并指定期望格式（如 base64 或 url）。在调用前，系统自动检测参数是 Base64 字符串还是 HTTP URL，并进行相应转换（如将 URL 下载为 Base64，或将 Base64 上传并获得 URL），使工具能正确接收。

实现

import base64
import re
from typing import Dict, Any

def preprocess_image_arguments(args: Dict[str, Any], schema: Dict[str, Any]) -> Dict[str, Any]:
    """
    处理工具调用中的图像参数：根据期望格式进行转换。
    schema 中的参数可指定 "format": "base64" 或 "url"。
    支持 Base64 自动检测（data:image/...;base64,... 或直接 Base64）和 URL 检测。
    """
    processed = args.copy()
    params_schema = schema.get("parameters", {})
    for param_name, param_info in params_schema.items():
        if param_info.get("type") != "image":
            continue
        value = processed.get(param_name)
        if not value:
            continue
        target_format = param_info.get("format", "base64")
        current_format = detect_image_format(value)
        if current_format == target_format:
            continue
        # 转换
        if target_format == "base64" and current_format == "url":
            processed[param_name] = url_to_base64(value)
        elif target_format == "url" and current_format == "base64":
            processed[param_name] = base64_to_url(value)  # 实际需上传到存储，这里模拟
        # 其他情况保持原样
    return processed

def detect_image_format(value: str) -> str:
    """检测字符串是 base64 还是 url"""
    if value.startswith("http://") or value.startswith("https://"):
        return "url"
    if re.match(r'^[A-Za-z0-9+/=]+$', value) or value.startswith("data:image"):
        return "base64"
    return "unknown"

def url_to_base64(url: str) -> str:
    # 模拟下载图片并转为 base64（实际需使用 requests）
    return f"base64_of({url})"

def base64_to_url(b64: str) -> str:
    # 模拟上传 base64 到临时存储并返回 URL
    return f"https://tmp.store/{hash(b64)}.png"

# 示例
schema = {
    "name": "analyze_image",
    "parameters": {
        "image": {"type": "image", "format": "base64"},
        "prompt": {"type": "str"}
    }
}
args = {"image": "https://example.com/photo.jpg", "prompt": "描述图片"}
processed = preprocess_image_arguments(args, schema)
print(processed)  # image 已转换为 base64

说明

  • 自动检测格式，根据期望格式进行转换，简化工具实现。

  • 实际转换需引入网络请求（下载/上传），此处用模拟函数示意。

构建一个完整的 Agent 工具调用循环：包含接收用户输入、调用 LLM 获取响应、解析工具调用（或结束标记）、执行工具、将结果反馈给 LLM，直到 LLM 输出最终答案，要求处理解析失败和工具执行异常。

原理

Agent 循环是 LLM Agent 的核心。其流程为：

  1. 接收用户输入，添加到对话历史。

  2. 调用 LLM，获取响应。

  3. 尝试从响应中解析工具调用。如果解析成功且工具调用有效，则执行工具，将执行结果作为 Observation 追加到对话历史，回到步骤 2。

  4. 如果解析失败（即响应中不包含工具调用，或解析出错），判断是否达到最大步数或 LLM 输出了最终答案（如包含 "Final Answer:" 标记）。若是，则将响应作为最终结果返回；否则将解析错误信息反馈给 LLM 并回到步骤 2。

  5. 若执行工具过程中出现异常，则将异常信息作为 Observation 反馈给 LLM，继续循环。

实现

import json
from typing import List, Dict, Any, Tuple

class AgentLoop:
    def __init__(self, llm_fn, tool_registry, parse_fn, max_steps=10):
        self.llm = llm_fn                     # 函数：输入对话历史，返回 LLM 响应文本
        self.registry = tool_registry         # 工具名 -> 可调用对象
        self.parse_fn = parse_fn              # 从文本解析工具调用的函数，返回 (tool_name, args) 或 None
        self.max_steps = max_steps
        self.history: List[Dict] = []

    def run(self, user_input: str) -> str:
        self.history.append({"role": "user", "content": user_input})
        steps = 0
        while steps < self.max_steps:
            # 1. 调用 LLM
            llm_response = self.llm(self.history)
            # 2. 尝试解析工具调用
            tool_call = self._parse_safe(llm_response)
            if tool_call is None:
                # 无工具调用，检查是否是最终答案
                if self._is_final(llm_response):
                    return llm_response
                # 否则将 LLM 响应作为普通回复添加到历史
                self.history.append({"role": "assistant", "content": llm_response})
                # 如果没有工具调用且不是最终答案，可能需要用户进一步输入，此处跳出循环
                break

            tool_name, args = tool_call
            # 3. 检查工具是否存在
            if tool_name not in self.registry:
                obs = f"错误：工具 '{tool_name}' 未注册。"
            else:
                # 4. 执行工具
                try:
                    result = self.registry[tool_name](**args)
                    obs = f"Observation: {json.dumps(result, ensure_ascii=False)}"
                except Exception as e:
                    obs = f"工具执行错误: {type(e).__name__}: {e}"
            # 5. 将工具调用和结果追加到历史
            self.history.append({"role": "assistant", "content": llm_response})
            self.history.append({"role": "tool", "content": obs})
            steps += 1
        # 超过最大步数，返回最后一次 LLM 响应（或提示）
        if self.history:
            last_msg = self.history[-1]["content"]
            if self._is_final(last_msg):
                return last_msg
            return "Agent 达到最大步数限制，未完成任务。"
        return "无法完成任务。"

    def _parse_safe(self, text: str) -> Tuple[str, Dict] | None:
        try:
            return self.parse_fn(text)
        except Exception:
            return None

    def _is_final(self, text: str) -> bool:
        return "Final Answer:" in text

# 示例 LLM 函数（模拟）
def mock_llm(history):
    last = history[-1]["content"] if history else ""
    if "天气" in last:
        return 'Action: get_weather\nAction Input: {"city": "北京"}'
    if "Observation" in last and "25" in last:
        return 'Final Answer: 北京今天25°C，晴朗。'
    if "错误" in last:
        return 'Final Answer: 抱歉，我无法完成该操作。'
    return 'Final Answer: 你好，有什么可以帮助的？'

def get_weather(city):
    return {"city": city, "temp": 25, "condition": "晴"}

def parse_react(text):
    # 简单实现
    if "Action:" in text and "Action Input:" in text:
        # 省略详细解析，返回模拟
        return ("get_weather", {"city": "北京"})
    return None

agent = AgentLoop(llm_fn=mock_llm, tool_registry={"get_weather": get_weather}, parse_fn=parse_react)
print(agent.run("北京天气怎么样？"))

说明

  • 循环中每次都将 LLM 响应和工具结果追加到历史，LLM 能逐步获得信息。

  • _parse_safe 捕获解析异常，避免程序中断。

  • 工具执行异常同样转为文本反馈，让 LLM 有机会调整（例如选择其他工具或请求用户帮助）。

  • 最终答案通过特定标记判断，也可以让 LLM 输出一个 Finish 动作。