Agent Harness 怎么搭:从金融助手看托管式与网关式两条路线

2026-09-25 18 预计阅读时间: 1 分钟
来源: infoq.com AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间:12 分钟

大模型只是 Agent 的推理核心,真正决定它能否上线的,往往是外围那层“壳”:工具注册、会话记忆、模型路由、预算限制、日志追踪和权限控制。通常可以把这层基础设施称为 Agent Harness。

以金融助手为例,原文比较了两种实现路线:使用 AWS AgentCore Harness 构建偏托管式的运行环境,以及使用 LangChain 配合 Envoy AI Gateway 组装更开放的技术栈。两者都能支撑 Agent,但平台责任、可移植性和工程投入明显不同。

Harness 不只是一个 Agent 循环

最简化的 Agent 循环看起来并不复杂:接收问题、调用模型、执行工具,再把工具结果交给模型。然而到了生产环境,团队还必须回答一组更棘手的问题:

  • 哪些工具可以被调用,参数是否经过校验?
  • 会话记忆保存在哪里,保留多久,谁可以读取?
  • 模型调用经过哪个端点,能否切换供应商?
  • 单次请求和单个用户最多消耗多少预算?
  • 一次回答为什么失败,究竟卡在模型、工具还是网关?
  • 含有账户信息的日志是否会进入普通可观测平台?

因此,一个实用的 Harness 至少包含五层能力:

  1. Agent 执行层:负责推理循环、工具选择和终止条件。
  2. 工具层:注册业务能力,并执行鉴权、参数验证与超时控制。
  3. 状态与记忆层:保存会话上下文、用户偏好或任务进度。
  4. 模型访问层:统一模型端点、凭证、重试、限流和路由策略。
  5. 运营层:提供成本统计、Tracing、审计日志、告警与安全策略。

金融场景会放大这些要求。查询余额可以是只读工具,转账却需要幂等键、二次确认和更严格的授权。Harness 如果只解决“让模型调用函数”,仍然不足以承担真实交易工作负载。

两条实现路线的边界在哪里

AWS AgentCore Harness:减少自建组件

托管式方案的主要价值,是把较多运行与运营能力放进云平台边界内。团队可以把精力集中在 Agent 逻辑、工具定义和业务权限上,而不是自行拼装每一个基础设施组件。

这种路线通常更适合以下情况:

  • 工作负载已经主要运行在 AWS 环境中;
  • 团队希望沿用既有的身份、网络、审计和运维体系;
  • 上线速度比底层组件的完全可替换性更重要;
  • 平台团队不希望长期维护独立的 AI 网关和 Agent 运行时。

代价也很明确:运行模型、记忆和可观测性的方式会更多地受平台抽象影响。未来迁移到其他云或自建环境时,业务工具可能容易保留,运行层和运维配置却未必可以原样搬走。

LangChain + Envoy AI Gateway:拆开编排与流量治理

另一条路线是让 LangChain 负责 Agent 编排和工具调用,把模型流量交给 Envoy AI Gateway 一类网关治理。这样,应用只访问统一入口,网关承担认证、路由、限流、重试、指标采集或成本策略。

这条路线的优势在于边界清晰:

  • Agent 框架可以独立演进;
  • 模型供应商可以隐藏在统一网关之后;
  • 多个 Agent 或应用能够复用同一套流量治理;
  • 网关日志可以与现有服务网格和可观测体系衔接。

相应地,团队需要拥有更多运营责任。网关高可用、配置发布、指标基数、密钥轮换、模型协议差异以及 LangChain 版本升级,都需要明确的负责人。开放并不等于免费,它只是把控制权和维护成本同时交给了工程团队。

用一个最小 Harness 看清组件职责

下面是一个可直接运行的概念示例。它不调用真实 AWS、LangChain 或 Envoy API,而是用标准库模拟两种运行模式,重点展示工具、记忆、预算和追踪应该如何分层。接入生产环境时,可分别替换 ModelGateway.complete 和 BalanceTool.run。

将代码保存为 harness_demo.py:

import json
import os
import time
import uuid
from dataclasses import dataclass, field


@dataclass
class RequestContext:
    user_id: str
    session_id: str
    trace_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    spent_units: int = 0


class MemoryStore:
    def __init__(self):
        self._messages = {}

    def append(self, session_id, role, content):
        self._messages.setdefault(session_id, []).append({
            'role': role,
            'content': content,
        })

    def recent(self, session_id, limit=6):
        return self._messages.get(session_id, [])[-limit:]


class BalanceTool:
    name = 'get_balance'

    def run(self, ctx, account_id):
        # 生产环境必须在服务端校验 user_id 是否有权访问 account_id。
        allowed = {'user-123': {'checking-001'}}
        if account_id not in allowed.get(ctx.user_id, set()):
            raise PermissionError('account access denied')
        return {'account_id': account_id, 'currency': 'USD', 'balance': 1250.40}


class ModelGateway:
    def __init__(self, mode):
        self.mode = mode

    def complete(self, messages, tool_result):
        # 实际接入点:
        # mode=agentcore 时替换为相应托管运行时或模型访问 SDK。
        # mode=envoy-langchain 时改为调用网关地址,由 LangChain 执行编排。
        return (
            f'[{self.mode}] Your checking account balance is '
            f'{tool_result["currency"]} {tool_result["balance"]:.2f}.'
        )


class FinanceHarness:
    def __init__(self, gateway, memory, max_units=10):
        self.gateway = gateway
        self.memory = memory
        self.max_units = max_units
        self.balance_tool = BalanceTool()

    def ask_balance(self, ctx, account_id):
        started = time.time()
        self._charge(ctx, 2)
        result = self.balance_tool.run(ctx, account_id)

        self._charge(ctx, 3)
        answer = self.gateway.complete(
            self.memory.recent(ctx.session_id), result
        )
        self.memory.append(ctx.session_id, 'assistant', answer)

        print(json.dumps({
            'event': 'agent_request_finished',
            'trace_id': ctx.trace_id,
            'session_id': ctx.session_id,
            'tool': self.balance_tool.name,
            'spent_units': ctx.spent_units,
            'duration_ms': round((time.time() - started) * 1000, 2),
            'status': 'ok',
        }))
        return answer

    def _charge(self, ctx, units):
        if ctx.spent_units + units > self.max_units:
            raise RuntimeError('request budget exceeded')
        ctx.spent_units += units


if __name__ == '__main__':
    mode = os.getenv('HARNESS_MODE', 'envoy-langchain')
    harness = FinanceHarness(
        gateway=ModelGateway(mode),
        memory=MemoryStore(),
        max_units=10,
    )
    context = RequestContext(user_id='user-123', session_id='session-001')
    print(harness.ask_balance(context, 'checking-001'))

运行两种模拟模式:

HARNESS_MODE=agentcore python harness_demo.py
HARNESS_MODE=envoy-langchain python harness_demo.py

这个示例刻意把四类职责拆开:

  • BalanceTool 在工具内部执行资源级授权,而不是相信模型传来的账户编号;
  • MemoryStore 只暴露有限历史,避免上下文无限增长;
  • ModelGateway 隔离模型访问方式,使业务代码不直接依赖供应商端点;
  • FinanceHarness 在调用前扣减预算,并输出结构化 Trace 事件。

真实系统还应把内存存储替换为持久化服务,并加入超时、重试、熔断、敏感字段脱敏和分布式 Trace。预算也不应只使用示例中的抽象单位,而要结合输入 Token、输出 Token、工具成本和模型单价计算。

比较时不要只看“能不能调用工具”

选择方案时,可以逐项评估以下维度:

维度 托管式 AgentCore 路线 LangChain + Envoy 路线
工具编排 更贴近平台提供的运行抽象 由应用框架和团队代码控制
记忆 倾向复用平台能力与云内服务 可自由选择数据库、缓存或向量存储
模型访问 便于融入云平台的身份和模型体系 网关可统一多个模型端点
成本控制 可借助平台级计量与治理能力 可在网关和应用层实施自定义策略
可观测性 更容易接入同一云环境的运营体系 能与已有 Envoy、服务网格和监控栈结合
可移植性 平台集成更深,迁移成本可能更高 组件可替换,但配置和兼容工作更多
运维责任 更多责任交给平台 团队负责网关、框架和集成层生命周期

这不是简单的“托管优于开源”或“开放方案更灵活”。真正的问题是:团队希望在哪一层保留控制权,又愿意为这份控制权承担多少值班和升级成本。

上线前的决策清单

如果应用已经深度运行在 AWS 上,又需要尽快形成统一的安全与运营边界,可以优先验证 AgentCore Harness。验证重点应放在工具权限、记忆生命周期、成本归因以及现有监控系统的衔接上。

如果组织已经维护 Envoy 或服务网格,希望跨模型供应商路由流量,并且具备平台工程能力,那么 LangChain 与 Envoy AI Gateway 的组合更值得探索。此时应先确定网关与应用各自负责什么,避免限流、重试和日志在两层重复实现。

无论选择哪条路线,上线前都应确认:

  • 工具执行使用用户身份,而不是共享的超级权限;
  • 写操作具有显式确认、幂等键和审计记录;
  • 记忆数据设置过期策略,并区分会话记忆与长期画像;
  • 成本限制同时覆盖单次请求、用户和租户;
  • Trace 能串联 Agent、模型网关和工具服务;
  • 日志不会泄露账户编号、交易信息、Prompt 或凭证;
  • 模型或网关不可用时,系统能够安全失败,而不是绕过策略直接调用。

一个成熟的 Agent Harness,本质上不是让模型拥有更多自由,而是把模型的不确定性限制在可审计、可计费、可恢复的工程边界内。


相关推荐