无审查 LLM API 上的 Agent：函数调用、框架与护栏

函数调用循环的结构

去掉框架后，每个 Agent 都执行相同的四个步骤。你发送对话加上 tools 列表。模型回复文本（表示结束）或 tool_calls（表示需要执行某项操作）。你执行每个调用，并为每个调用追加一条 tool 消息，引用该调用的 id。然后你带着更长的对话再次询问模型。

分支的信号是 finish_reason。当其等于 tool_calls 时，助手消息的 content 设为 null，并包含一个 tool_calls 数组，每个条目包含 id、函数名和 JSON 格式的 arguments string。参考页面 显示了确切的结构。有两个细节容易让人出错：参数是一个必须解析的字符串，且包含调用的助手消息必须按顺序保留在历史记录中，位于你的工具结果之前。

从零开始的完整循环

下面的脚本定义了一个函数，以 JSON 模式描述它，并循环直到模型生成纯文本。它使用指向该 API 的官方 Python SDK。每次迭代都会追加助手消息，然后为每个调用追加一条工具消息，并使用步骤计数器防止失控行为。

Copyimport json, os
from openai import OpenAI

client = OpenAI(base_url="https://api.uncensoredapis.com/v1", api_key=os.environ["API_KEY"])

TICKETS = {"T-101": "open", "T-102": "closed"}

def ticket_status(ticket_id: str) -> str:
    return TICKETS.get(ticket_id, "unknown ticket")

TOOLS = [{
    "type": "function",
    "function": {
        "name": "ticket_status",
        "description": "Look up the status of a support ticket by id.",
        "parameters": {
            "type": "object",
            "properties": {"ticket_id": {"type": "string"}},
            "required": ["ticket_id"],
        },
    },
}]
REGISTRY = {"ticket_status": ticket_status}

messages = [{"role": "user", "content": "Are tickets T-101 and T-102 still open?"}]

for step in range(6):                      # hard cap on loop turns
    r = client.chat.completions.create(
        model="uncensored", messages=messages, tools=TOOLS, max_tokens=400)
    msg = r.choices[0].message
    messages.append(msg.model_dump(exclude_none=True))
    if r.choices[0].finish_reason != "tool_calls":
        print(msg.content)
        break
    for call in msg.tool_calls:
        fn = REGISTRY.get(call.function.name)
        try:
            args = json.loads(call.function.arguments)
            result = fn(**args) if fn else f"error: unknown tool {call.function.name}"
        except Exception as exc:
            result = f"error: {exc}"
        messages.append({"role": "tool", "tool_call_id": call.id, "content": str(result)})
else:
    print("stopped: step limit reached")

注意循环在失败时的行为：错误以字符串形式成为工具结果。这使模型能够读取问题并尝试重试，例如修正错误的工单 ID，而不是你的程序崩溃。for ... else 子句仅在达到上限且没有最终答案时触发，这是检测模型陷入循环的方法。

保持提示词专注。简短的系统消息说明何时调用工具以及何时直接回答，可以减少不必要的调用，每次跳过的调用都能节省一次往返、token 和速率限制的一个槽位。

保护工具执行

模型输出是不可信的输入。由模型选择或参数的工具名称或参数可能是错误的、奇怪的，或者受到模型从网页或用户处读取文本的影响。将每次调用视为来自公共互联网的请求。

• 按名称列入白名单。 在你构建的注册表中查找函数。切勿使用字符串来导入或评估代码。
• 读写分离。 只读工具可以自由运行。任何发送邮件、花钱、删除数据或操作文件的操作都需要明确的人工批准。
• 验证参数 是否符合你声明的模式：必填键、无多余键、类型和长度限制。
• 限制时间和输出。 应用超时，并在将结果反馈之前截断结果，以免巨大的工具输出占用你的 100,000-token 上下文。
• 限制迭代和成本。 限制每个任务的步骤数和每个会话的 token 数，使用 usage 块。
• 记录每次调用 的名称、参数和结果，以便审计行为。

Copyimport json
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutTimeout

ALLOWED = {"ticket_status", "lookup_invoice"}      # read-only tools only
NEEDS_APPROVAL = {"refund_order"}                  # side effects wait for a human
POOL = ThreadPoolExecutor(max_workers=4)

def validate(args: dict, schema: dict) -> None:
    props = schema["properties"]
    for key in schema.get("required", []):
        if key not in args:
            raise ValueError(f"missing argument: {key}")
    for key, val in args.items():
        if key not in props:
            raise ValueError(f"unexpected argument: {key}")
        if props[key]["type"] == "string" and not isinstance(val, str):
            raise ValueError(f"{key} must be a string")
        if isinstance(val, str) and len(val) > 200:
            raise ValueError(f"{key} too long")

def run_tool(name, raw_args, registry, schemas, approve=lambda n, a: False):
    if name in NEEDS_APPROVAL:
        args = json.loads(raw_args)
        if not approve(name, args):
            return "denied: this action needs human approval"
    elif name not in ALLOWED:
        return f"error: tool {name} is not permitted"
    args = json.loads(raw_args)
    validate(args, schemas[name]["parameters"])
    future = POOL.submit(registry[name], **args)
    try:
        return str(future.result(timeout=5))[:2000]    # time and size limits
    except FutTimeout:
        return "error: tool timed out"

也将工具结果视为不可信。如果工具返回包含指令的抓取文本，模型可能会遵循它们。在工具消息中用清晰的分隔符包装此类内容，并在系统提示词中说明工具输出是数据而非命令。

编写模型可用的工具模式

模式是模型看到的唯一文档，因此值得像公共 API 一样精心维护。使用匹配任务的动词和名词命名函数，例如 ticket_status 或 lookup_invoice，而不是像 run 或 helper 这样的通用标签。将 description 写成指令：函数做什么、返回什么以及何时不使用它。

• 保持参数少且扁平。 两三个简单字段的调用正确率远高于包含十个字段的嵌套对象。
• 对封闭集使用枚举，例如 "enum": ["open", "closed"]，这样模型就无法发明值。
• 用单位或示例描述每个属性："Ticket id such as T-101" 优于裸字符串。
• 返回紧凑的结果。 字符串或小 JSON 对象效果最好；冗长的转储占用上下文但收益甚微。
• 优先使用多个专用工具，而不是带有模式标志的一个全能工具。

当调用验证失败时，将验证消息作为工具结果返回，而不是抛出异常。模型通常会在下一轮修复调用，因为它可以读取哪里出错了。如果它以相同的方式失败两次，重写描述，而不是循环。

LangChain 配置

LangChain 的 ChatOpenAI 类接受 base_url 参数，因此无需自定义提供商包。将 model 设置为 uncensored，从你的环境中传递密钥，并使用 bind_tools 绑定工具。安装 langchain-openai 和 langchain-core。

Copyimport os
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

@tool
def ticket_status(ticket_id: str) -> str:
    """Return the status of a support ticket."""
    return {"T-101": "open", "T-102": "closed"}.get(ticket_id, "unknown ticket")

llm = ChatOpenAI(
    model="uncensored",
    base_url="https://api.uncensoredapis.com/v1",
    api_key=os.environ["API_KEY"],
    temperature=0.3,
    max_tokens=400,
)
llm_with_tools = llm.bind_tools([ticket_status])

ai = llm_with_tools.invoke("What is the status of ticket T-101?")
for call in ai.tool_calls:
    print(call["name"], call["args"], "->", ticket_status.invoke(call["args"]))

invoke 的结果是一个 AIMessage，其 tool_calls 列表包含已解析的名称和参数，因此你可以跳过手动 JSON 步骤。要获得完整的自主循环，将相同的 llm 对象交给 LangChain 的 Agent 构造函数；该接口只需支持工具调用即可，它确实支持。如果 Agent 似乎忽略了其工具，请检查是否存在 docstring 和类型提示，因为它们会成为模型看到的模式。

LlamaIndex 配置

LlamaIndex 为遵循聊天完成格式的服务器提供 OpenAILike。三个设置很重要。api_base 指向 API，is_chat_model=True 选择聊天路由，is_function_calling_model=True 告诉框架它可以发送工具。将 context_window 设置为 100000，以便其文本拆分器和内存缓冲区正确调整大小。安装 llama-index-llms-openai-like。

Copyimport os
from llama_index.llms.openai_like import OpenAILike
from llama_index.core.tools import FunctionTool

def ticket_status(ticket_id: str) -> str:
    """Return the status of a support ticket."""
    return {"T-101": "open", "T-102": "closed"}.get(ticket_id, "unknown ticket")

llm = OpenAILike(
    model="uncensored",
    api_base="https://api.uncensoredapis.com/v1",
    api_key=os.environ["API_KEY"],
    is_chat_model=True,
    is_function_calling_model=True,
    context_window=100000,
    max_tokens=400,
)

tool = FunctionTool.from_defaults(fn=ticket_status)
response = llm.predict_and_call([tool], "Check ticket T-102 for me.")
print(response)

从这里你可以将相同的 llm 传递给 Agent 工作器或查询引擎。请记住，检索管道会将大量文本块塞入一个提示词；在提示词和补全共享 100,000 token 上下文窗口的情况下，请限制检索到的文本块数量，并将 max_tokens 保持在较低水平。

在预算内运行 Agent

一个 Agent 会将一条用户消息转化为多次请求，因此需要规划容量。一个包含五次工具轮次的任务需要六次模型调用。因此，十个并发任务会在短时间内产生六十次请求，300 次/分钟的速率限制会比聊天应用更快达到。请在客户端前方部署 errors and rate limits playbook 中的令牌桶限流机制。

上下文随着每一轮而增长，因为每个工具结果都会被附加。一旦工具输出已被使用，请总结或丢弃旧的工具输出，并关注 usage.prompt_tokens；当它接近 50,000 时，压缩历史记录。对于成本，假设一个示例任务包含 6 次调用，平均 3,000 个提示词 token 和 200 个完成 token：即 18,000 个输入 token（约 $0.0045，按每百万 $0.25 计算）和 1,200 个输出 token（约 $0.0012，按每百万 $1.00 计算）。这些 token 计数是用于求和的假设，而非测量值。如果你是该接口的新手，请从 文档 和免费试用额度开始。

测试和调试 Agent

Agent 的失败方式是单个函数的单元测试无法发现的，因此测试对话。记录每次运行的完整消息列表，包括工具参数和结果，并从日志中重放失败。三个习惯能迅速带来回报。

1. 模拟模型。 存根客户端，使其返回脚本化的 tool_calls 消息。这可以在没有网络和花费的情况下测试你的注册表、验证和循环终止。
2. 固定采样。 对于工具驱动的任务，使用较低的 temperature，例如 0.2。降低随机性可以提供更可重复的参数 JSON，你可以在后续的写作密集型步骤中提高它。
3. 断言轨迹。 对于少量固定提示词，检查调用了哪些工具以及顺序，而不仅仅是最终文本。

同时观察 finish_reason。在工具调用中间出现 length 值意味着参数被截断；提高 max_tokens（最高 32,000），JSON 将完成。流式输出也适用于工具，但请按调用索引累积参数片段，并在流关闭后仅进行解析。有关这些块的复习，请参阅 参考中的流式输出部分。