1. 从“失控的工具调用”到“结构化状态管理”的必然演进最近在设计和实现一些需要调用外部工具Tool-Calling的智能体Agent时我反复被一个问题困扰如何确保这些智能体在复杂的、多步骤的任务执行过程中始终如一地遵守预设的策略和约束比如一个负责处理财务数据的智能体必须确保它调用的每一个API、生成的每一条查询都符合数据访问权限和审计规则一个自动化客服智能体在与用户交互时其每一步操作都必须遵循服务协议和隐私政策。这不仅仅是简单的“调用前检查”而是在一个可能包含分支、循环、条件判断的漫长执行轨迹中对智能体的“行为记忆”和“决策上下文”进行持续、可信的追踪与验证。传统的智能体架构其内部状态往往是“黑盒”或“非结构化”的。状态可能散落在对话历史、临时的内存变量或未经验证的执行日志中。当需要判断“智能体到目前为止的行为是否合规”时我们缺乏一个权威的、不可篡改的“账本”来作为依据。这就好比让一个会计手工记账每一笔收支都记在不同的便签纸上事后审计几乎无法进行。LedgerAgent这个概念正是为了解决这一核心痛点而提出的。它本质上是一种为工具调用型智能体引入“结构化状态”Structured State的设计范式其核心思想是将智能体的每一次工具调用、每一次状态变更都像会计记账一样记录在一个结构化的、可验证的“账本”Ledger中并基于此账本实时执行策略Policy adherence 检查。简单来说LedgerAgent 不是一个具体的库或框架虽然我们可以基于此理念构建而是一种架构模式。它旨在将智能体从“凭感觉行事”的松散状态转变为“每一步都有据可查、有规可循”的严谨系统。这对于金融、医疗、法律、企业级自动化等对合规性、安全性和可审计性要求极高的场景具有至关重要的意义。接下来我将深入拆解 LedgerAgent 的核心组件、工作原理并分享一个从零开始构建简易 LedgerAgent 原型的设计思路与实战代码。2. 解构 LedgerAgent账本、状态与策略的三位一体要理解 LedgerAgent必须厘清三个核心概念结构化状态Structured State、策略Policy和账本Ledger。这三者构成了一个闭环的控制系统。2.1 结构化状态从混沌到秩序智能体在运行过程中会产生大量信息用户输入、自身思考Chain-of-Thought、工具调用请求、工具调用结果、环境反馈等。在普通智能体中这些信息可能以非结构化的文本序列或简单的字典形式存在。LedgerAgent 要求我们将这些信息结构化。一个典型的结构化状态条目State Entry可能包含以下字段entry_id: 条目的唯一标识符UUID或自增ID。timestamp: 条目创建的时间戳。agent_phase: 智能体所处的阶段如“THINKING”、“TOOL_CALL_REQUEST”、“TOOL_CALL_RESULT”、“ACTION”、“OBSERVATION”。content: 该阶段的具体内容。例如在TOOL_CALL_REQUEST阶段content可能是一个结构化的 JSON包含tool_name、tool_parameters在THINKING阶段可能是推理链文本。parent_entry_id: 指向引发当前条目的父条目ID用于构建状态树State Tree追溯决策链。metadata: 其他元数据如会话ID、用户ID、环境变量等。这种结构化的好处是显而易见的状态变得可查询、可分析、可验证。我们可以轻易地回答“智能体在过去的5分钟里调用了哪些工具”、“它是基于哪一步的思考结果做出了这个工具调用决策”2.2 策略定义行为的“交通规则”策略Policy是一组规则或约束条件用于规定智能体什么能做、什么不能做、以及应该以何种方式做。策略可以非常具体例如工具调用策略“禁止调用delete_database工具”“调用query_sensitive_data工具前必须已成功调用user_authentication工具”。数据流策略“工具A的输出结果若包含‘PII’个人身份信息标签则只能作为工具B的输入且工具B必须具有‘PII处理权限’”。会话策略“单次会话中调用付费API的次数不得超过10次”。业务逻辑策略“生成报告前必须已收集‘数据源A’和‘数据源B’的结果”。策略通常以声明式的语言或领域特定语言DSL来定义也可以通过在代码中嵌入检查函数来实现。策略引擎是 LedgerAgent 的“大脑”它负责解读并执行这些规则。2.3 账本不可篡改的执行轨迹记录账本Ledger是结构化状态的持久化存储和序列。它按时间顺序记录了智能体生命周期内的所有状态条目。账本的核心特性是“只追加”Append-Only。一旦一个状态条目被写入账本就不应被修改或删除。这保证了执行轨迹的完整性和可审计性就像区块链的分布式账本一样为事后追溯和问责提供了铁证。账本可以是内存中的数据结构如列表、数据库中的一张表、一个文件如JSON Lines格式甚至是分布式的日志系统如Apache Kafka。选择哪种实现取决于对持久性、性能和查询能力的需求。三者的工作流关系智能体产生一个新的意图或行动例如决定调用一个工具。该意图被格式化为一个结构化的状态条目TOOL_CALL_REQUEST。在真正执行调用之前这个待定的状态条目连同当前的完整账本历史被提交给策略引擎进行验证。策略引擎遍历所有相关策略检查如果将此条目加入账本是否会违反任何规则。例如检查该工具是否在允许列表内参数是否合规调用频率是否超限等。验证通过该条目被正式“记账”写入账本智能体随后执行真正的工具调用并将调用结果作为新的状态条目TOOL_CALL_RESULT再次记入账本。验证拒绝条目被拒绝写入账本智能体收到一个策略违规错误必须调整其决策例如选择其他工具或参数。这个“提议 - 验证 - 记账 - 执行”的循环是 LedgerAgent 确保策略遵从性的关键机制。3. 实战构建一个简易 LedgerAgent 原型设计与实现理论讲完了我们来点实际的。我将设计一个基于 Python、使用 LangChain 框架因其在工具调用方面的生态成熟的简易 LedgerAgent 原型。这个原型将清晰地展示上述核心概念。3.1 系统架构与核心类设计我们将创建几个核心类StructuredStateEntry: 表示一个状态条目。Ledger: 负责存储和管理状态条目序列。Policy与PolicyEngine: 定义策略规则和验证引擎。LedgerAgent: 继承或包装一个基础智能体如 LangChain 的AgentExecutor在其决策循环中插入账本记录和策略检查。from datetime import datetime from enum import Enum from typing import Any, Dict, List, Optional, Tuple from uuid import uuid4, UUID from pydantic import BaseModel, Field from langchain.agents import AgentExecutor, BaseSingleActionAgent from langchain.schema import AgentAction, AgentFinish class AgentPhase(str, Enum): 智能体执行阶段枚举 THINKING THINKING TOOL_CALL_REQUEST TOOL_CALL_REQUEST TOOL_CALL_RESULT TOOL_CALL_RESULT ACTION ACTION OBSERVATION OBSERVATION ERROR ERROR POLICY_VIOLATION POLICY_VIOLATION class StructuredStateEntry(BaseModel): 结构化状态条目数据模型 entry_id: UUID Field(default_factoryuuid4) timestamp: datetime Field(default_factorydatetime.utcnow) phase: AgentPhase content: Dict[str, Any] # 结构化内容如 {tool_name: search, args: {...}} parent_entry_id: Optional[UUID] None # 形成状态树 metadata: Dict[str, Any] Field(default_factorydict) class Config: use_enum_values True class Ledger: 账本管理结构化状态条目的只追加存储 def __init__(self): self._entries: List[StructuredStateEntry] [] def append(self, entry: StructuredStateEntry) - None: 只追加写入条目 self._entries.append(entry) def get_all(self) - List[StructuredStateEntry]: 获取所有条目按时间顺序 return self._entries.copy() def get_since(self, entry_id: Optional[UUID] None) - List[StructuredStateEntry]: 获取自某个条目之后的所有条目用于增量策略检查 if entry_id is None: return self.get_all() for i, entry in enumerate(self._entries): if entry.entry_id entry_id: return self._entries[i1:] return [] class PolicyViolation(Exception): 策略违反异常 def __init__(self, message: str, entry: StructuredStateEntry): super().__init__(message) self.violating_entry entry class Policy: 策略基类所有具体策略需继承此类 def check(self, proposed_entry: StructuredStateEntry, ledger: Ledger) - Optional[str]: 检查待添加的条目是否违反本策略。 返回 None 表示通过返回 str 表示违反原因。 此方法接收待添加条目和当前完整账本。 raise NotImplementedError class ToolWhitelistPolicy(Policy): 工具白名单策略只允许调用指定工具 def __init__(self, allowed_tools: List[str]): self.allowed_tools set(allowed_tools) def check(self, proposed_entry: StructuredStateEntry, ledger: Ledger) - Optional[str]: if proposed_entry.phase AgentPhase.TOOL_CALL_REQUEST: tool_name proposed_entry.content.get(tool_name) if tool_name and tool_name not in self.allowed_tools: return fTool {tool_name} is not in the whitelist. Allowed: {list(self.allowed_tools)} return None class SequentialDependencyPolicy(Policy): 顺序依赖策略调用工具B之前必须先成功调用工具A def __init__(self, prerequisite_tool: str, dependent_tool: str): self.prerequisite prerequisite_tool self.dependent dependent_tool def check(self, proposed_entry: StructuredStateEntry, ledger: Ledger) - Optional[str]: if proposed_entry.phase AgentPhase.TOOL_CALL_REQUEST: tool_name proposed_entry.content.get(tool_name) if tool_name self.dependent: # 检查历史账本中是否有成功的 prerequisite 工具调用 has_prerequisite any( e.phase AgentPhase.TOOL_CALL_RESULT and e.content.get(tool_name) self.prerequisite and e.content.get(success) is True for e in ledger.get_all() ) if not has_prerequisite: return fCannot call {self.dependent} before {self.prerequisite} has been successfully called. return None class PolicyEngine: 策略引擎组合并执行所有策略 def __init__(self, policies: List[Policy]): self.policies policies def validate(self, proposed_entry: StructuredStateEntry, ledger: Ledger) - None: 验证待添加条目违反任何策略则抛出 PolicyViolation 异常 for policy in self.policies: violation_reason policy.check(proposed_entry, ledger) if violation_reason: # 记录策略违反条目 violation_entry StructuredStateEntry( phaseAgentPhase.POLICY_VIOLATION, content{ policy_type: policy.__class__.__name__, reason: violation_reason, blocked_entry: proposed_entry.dict() }, parent_entry_idproposed_entry.entry_id ) ledger.append(violation_entry) # 将违规记录也记入账本 raise PolicyViolation(violation_reason, proposed_entry)3.2 集成 LangChain Agent打造 LedgerAgentExecutor现在我们将上述组件与 LangChain 的AgentExecutor集成。关键点在于重写或 Hook_take_next_step这类核心方法在智能体决定调用工具AgentAction和得到工具结果后插入账本记录和策略检查。class LedgerAgentExecutor(AgentExecutor): 带账本和策略检查的 AgentExecutor。 继承自 LangChain 的 AgentExecutor重写关键步骤以集成结构化状态管理。 def __init__(self, agent: BaseSingleActionAgent, tools: List, policy_engine: PolicyEngine, **kwargs): super().__init__(agentagent, toolstools, **kwargs) self.ledger Ledger() self.policy_engine policy_engine self._current_parent_id: Optional[UUID] None def _record_entry(self, phase: AgentPhase, content: Dict, parent_id: Optional[UUID] None) - StructuredStateEntry: 创建并记录一个状态条目 entry StructuredStateEntry(phasephase, contentcontent, parent_entry_idparent_id or self._current_parent_id) # 如果是 TOOL_CALL_REQUEST需通过策略引擎验证 if phase AgentPhase.TOOL_CALL_REQUEST: try: self.policy_engine.validate(entry, self.ledger) except PolicyViolation as e: # 验证失败记录异常并直接返回阻止后续工具调用 self.ledger.append(e.violating_entry) raise e # 验证通过或无需验证的阶段正式记账 self.ledger.append(entry) # 更新当前父ID用于链式记录 if phase in [AgentPhase.TOOL_CALL_REQUEST, AgentPhase.THINKING]: self._current_parent_id entry.entry_id elif phase AgentPhase.TOOL_CALL_RESULT: self._current_parent_id None # 结果条目后重置父ID return entry async def _atake_next_step(self, name_to_tool_map: Dict[str, Any], inputs: Dict[str, str]) - List[Tuple[AgentAction, str]]: 重写异步的下一步执行方法。 这是核心拦截点在智能体决定行动和得到观察结果时插入记账逻辑。 # 1. 智能体“思考”并决定下一步行动 # 这里调用父类方法获取原始的 AgentAction 或 AgentFinish next_step_output await super()._atake_next_step(name_to_tool_map, inputs) # next_step_output 可能是一个 AgentAction调用工具或 AgentFinish结束 for agent_action, output in next_step_output: if isinstance(agent_action, AgentAction): # 2. 记录工具调用请求TOOL_CALL_REQUEST tool_call_entry self._record_entry( phaseAgentPhase.TOOL_CALL_REQUEST, content{ tool_name: agent_action.tool, tool_input: agent_action.tool_input, log: agent_action.log } ) # 3. 执行工具调用父类逻辑 # 注意实际的工具调用在父类的 _execute_agent_action 中我们通过重写 _record_entry 已经做了策略检查。 # 如果策略检查失败_record_entry 会抛出 PolicyViolation此处不会执行到。 observation await self._arun_tool(agent_action, name_to_tool_map) # 4. 记录工具调用结果TOOL_CALL_RESULT self._record_entry( phaseAgentPhase.TOOL_CALL_RESULT, content{ tool_name: agent_action.tool, input: agent_action.tool_input, observation: observation, success: True # 简化处理实际应根据工具执行是否异常判断 }, parent_idtool_call_entry.entry_id ) # 5. 将观察结果返回给智能体进行下一轮思考 return [(agent_action, observation)] elif isinstance(agent_action, AgentFinish): # 任务结束记录最终动作 self._record_entry( phaseAgentPhase.ACTION, content{return_values: agent_action.return_values, log: agent_action.log} ) return [(agent_action, )] return [] # 同步方法 _take_next_step 也需要类似重写此处省略原理相同。3.3 运行示例与账本分析假设我们有两个工具get_user_profile获取用户资料和query_payment_history查询支付记录。我们制定策略必须成功调用get_user_profile后才能调用query_payment_history。from langchain.agents import initialize_agent, Tool from langchain.llms import OpenAI # 1. 定义工具模拟 def get_user_profile(user_id: str) - str: return fProfile of user {user_id}: Active, VIP member. def query_payment_history(user_id: str) - str: return fPayment history for {user_id}: 3 transactions in 2024. tools [ Tool(nameGetUserProfile, funcget_user_profile, descriptionGet user profile by ID), Tool(nameQueryPaymentHistory, funcquery_payment_history, descriptionQuery users payment history), ] # 2. 初始化基础LLM和智能体 llm OpenAI(temperature0) # 假设使用OpenAI模型 base_agent initialize_agent(tools, llm, agentzero-shot-react-description, verboseTrue) # 3. 定义策略 policies [ SequentialDependencyPolicy(prerequisite_toolGetUserProfile, dependent_toolQueryPaymentHistory), ] policy_engine PolicyEngine(policies) # 4. 创建 LedgerAgentExecutor ledger_agent LedgerAgentExecutor(agentbase_agent.agent, toolstools, policy_enginepolicy_engine, max_iterations5, verboseTrue) # 5. 运行一个任务 try: result ledger_agent.run(先获取用户123的资料然后查询他的支付记录。) print(任务结果:, result) except PolicyViolation as e: print(f策略违规原因{e}) # 6. 查看完整的执行账本 print(\n 完整账本记录 ) for entry in ledger_agent.ledger.get_all(): print(f[{entry.timestamp.isoformat()}] {entry.phase}: {entry.content})预期输出与账本分析 如果智能体首先尝试调用QueryPaymentHistory策略引擎会在TOOL_CALL_REQUEST阶段拦截抛出PolicyViolation异常并在账本中记录一条POLICY_VIOLATION条目。智能体如果设计良好应该能根据这个错误调整其计划转而先调用GetUserProfile。一个成功的账本可能如下所示[2024-05-27T10:00:00] THINKING: {reasoning: 用户要求先获取资料再查记录我应该先调用GetUserProfile。} [2024-05-27T10:00:01] TOOL_CALL_REQUEST: {tool_name: GetUserProfile, tool_input: {user_id: 123}} [2024-05-27T10:00:02] TOOL_CALL_RESULT: {tool_name: GetUserProfile, observation: Profile of user 123..., success: true} [2024-05-27T10:00:03] THINKING: {reasoning: 已获取资料现在可以安全地查询支付记录了。} [2024-05-27T10:00:04] TOOL_CALL_REQUEST: {tool_name: QueryPaymentHistory, tool_input: {user_id: 123}} [2024-05-27T10:00:05] TOOL_CALL_RESULT: {tool_name: QueryPaymentHistory, observation: Payment history for 123..., success: true} [2024-05-27T10:00:06] ACTION: {return_values: {output: 已完成查询。}}这个账本清晰地展示了智能体的完整思维和行动轨迹并且每一步都隐含了策略合规的证明因为违规的请求根本无法被记录。4. 深入策略引擎从简单规则到复杂逻辑上面的例子展示了基于当前待记条目和完整历史账本的策略检查。但在实际生产中策略可能复杂得多。4.1 基于状态的策略Stateful Policies有些策略需要维护自己的内部状态。例如一个“限流策略”需要记录单位时间内的调用次数。class RateLimitPolicy(Policy): 限流策略限制特定工具在时间窗口内的调用次数 def __init__(self, tool_name: str, max_calls: int, window_seconds: int): self.tool_name tool_name self.max_calls max_calls self.window timedelta(secondswindow_seconds) def check(self, proposed_entry: StructuredStateEntry, ledger: Ledger) - Optional[str]: if proposed_entry.phase AgentPhase.TOOL_CALL_REQUEST: if proposed_entry.content.get(tool_name) self.tool_name: now proposed_entry.timestamp window_start now - self.window # 查询账本统计在时间窗口内该工具的调用成功次数 call_count sum( 1 for e in ledger.get_all() if e.phase AgentPhase.TOOL_CALL_RESULT and e.content.get(tool_name) self.tool_name and e.content.get(success) is True and window_start e.timestamp now ) if call_count self.max_calls: return fRate limit exceeded for {self.tool_name}. Max {self.max_calls} calls per {self.window.seconds}s. return None4.2 复合策略与策略优先级现实中的策略往往是复合的。我们可以设计一个CompositePolicy它包含多个子策略并定义评估顺序如短路评估一个失败则整体失败和冲突解决机制。class CompositePolicy(Policy): 组合策略按顺序评估多个子策略 def __init__(self, policies: List[Policy], evaluation_mode: str all_pass): # all_pass 或 first_fail self.policies policies self.mode evaluation_mode def check(self, proposed_entry: StructuredStateEntry, ledger: Ledger) - Optional[str]: reasons [] for policy in self.policies: reason policy.check(proposed_entry, ledger) if reason: if self.mode first_fail: return fCompositePolicy(first_fail): {reason} reasons.append(f[{policy.__class__.__name__}] {reason}) if reasons: return fCompositePolicy(all_pass) violations:\n \n.join(reasons) return None4.3 策略的动态加载与更新在生产环境中策略可能需要热更新。我们可以将策略定义存储在外部如数据库、配置文件PolicyEngine定期 reload。或者更高级的做法是让策略本身也是一个可以被智能体查询的“工具”但对其修改权限施加更严格的管控。5. 性能、扩展性与生产级考量将每个动作都进行结构化记录和策略检查无疑会引入开销。在设计生产级 LedgerAgent 时以下几点至关重要5.1 账本存储的优化分级存储将高频访问的最新条目放在内存或 Redis 中历史条目归档到对象存储如 S3或时序数据库。索引优化为账本条目建立索引如phase,tool_name,timestamp加速策略引擎的历史查询。可以使用专门的文档数据库如 MongoDB、Elasticsearch来存储账本。异步写入策略检查必须同步以保证决策合规但条目的持久化存储可以异步进行以提高主循环的响应速度。需要处理好异步写入失败的重试和一致性补偿。5.2 策略检查的优化增量检查策略引擎不需要每次都对完整账本进行全量扫描。Ledger.get_since()方法提供了增量条目。许多策略如顺序依赖只需要检查最新的一部分条目或与当前条目相关的条目。策略编译与预计算将声明式策略编译成高效的可执行代码如生成决策树或状态机。对于限流类策略可以维护一个内存中的滑动窗口计数器而不是每次都查询账本。并行检查如果策略之间没有依赖可以在多个线程或进程中并行执行检查。5.3 与现有监控、审计系统的集成Ledger 本身就是一个强大的审计日志源。可以很容易地将其条目转发到现有的日志聚合系统如 ELK Stack、监控系统如 Prometheus或安全信息与事件管理SIEM系统。例如每一个POLICY_VIOLATION条目都可以触发一个高优先级的告警。5.4 对智能体“思考”过程的策略约束目前的策略主要约束“行动”工具调用。更进一步的我们可以尝试约束智能体的“思考”推理过程。例如可以要求智能体在思考链中必须包含某些关键词如“检查权限”或者禁止其思考某些敏感话题。这可以通过对THINKING阶段的内容进行自然语言处理NLP或嵌入向量相似度检查来实现虽然技术上更具挑战性但代表了更深度的策略控制。6. 总结LedgerAgent 的价值与未来展望通过以上深入的探讨和实战构建我们可以看到 LedgerAgent 模式为 Tool-Calling Agents 带来了根本性的改变可审计性结构化的账本提供了机器可读、不可篡改的完整执行轨迹满足合规与审计的硬性要求。可控性策略引擎在决策点进行实时拦截将安全与合规要求从“事后补救”变为“事前预防”和“事中控制”。可解释性当智能体行为出现偏差时开发者、审计员或用户可以通过审查账本和关联的策略精准定位问题根源是工具问题、策略缺陷还是智能体逻辑错误。可测试性可以基于账本回放智能体的执行过程进行回归测试和压力测试验证策略修改的影响。在实际操作中我发现在引入 LedgerAgent 模式后调试复杂智能体的效率显著提升。以前需要反复查看冗长的非结构化日志现在可以直接查询账本数据库用 SQL 或特定查询语言快速定位“所有调用了敏感工具 X 的会话”或“所有违反了策略 Y 的请求”。同时它也促使我们在项目早期就更加严谨地思考智能体的行为边界定义清晰的策略而不是等到出了问题再打补丁。当然这套架构也会增加系统的复杂性。对于简单的、非关键的智能体应用可能显得有些“杀鸡用牛刀”。但对于企业级、金融级、涉及敏感操作的应用这种对“结构化状态”和“策略遵从”的投入是必要且值得的。未来我期待看到更多开源框架和云服务将 LedgerAgent 的理念内置为一种标准选项并提供更强大的策略语言、可视化审计工具和性能优化方案让构建安全、可靠、可信的智能体应用成为所有开发者的标配能力。 
长河君 · 2026/8/24 17:10:12 · 阅读 1.2k LedgerAgent：为AI智能体构建结构化状态管理与策略执行框架
1. 从“失控的工具调用”到“结构化状态管理”的必然演进最近在设计和实现一些需要调用外部工具Tool-Calling的智能体Agent时我反复被一个问题困扰如何确保这些智能体在复杂的、多步骤的任务执行过程中始终如一地遵守预设的策略和约束比如一个负责处理财务数据的智能体必须确保它调用的每一个API、生成的每一条查询都符合数据访问权限和审计规则一个自动化客服智能体在与用户交互时其每一步操作都必须遵循服务协议和隐私政策。这不仅仅是简单的“调用前检查”而是在一个可能包含分支、循环、条件判断的漫长执行轨迹中对智能体的“行为记忆”和“决策上下文”进行持续、可信的追踪与验证。传统的智能体架构其内部状态往往是“黑盒”或“非结构化”的。状态可能散落在对话历史、临时的内存变量或未经验证的执行日志中。当需要判断“智能体到目前为止的行为是否合规”时我们缺乏一个权威的、不可篡改的“账本”来作为依据。这就好比让一个会计手工记账每一笔收支都记在不同的便签纸上事后审计几乎无法进行。LedgerAgent这个概念正是为了解决这一核心痛点而提出的。它本质上是一种为工具调用型智能体引入“结构化状态”Structured State的设计范式其核心思想是将智能体的每一次工具调用、每一次状态变更都像会计记账一样记录在一个结构化的、可验证的“账本”Ledger中并基于此账本实时执行策略Policy adherence 检查。简单来说LedgerAgent 不是一个具体的库或框架虽然我们可以基于此理念构建而是一种架构模式。它旨在将智能体从“凭感觉行事”的松散状态转变为“每一步都有据可查、有规可循”的严谨系统。这对于金融、医疗、法律、企业级自动化等对合规性、安全性和可审计性要求极高的场景具有至关重要的意义。接下来我将深入拆解 LedgerAgent 的核心组件、工作原理并分享一个从零开始构建简易 LedgerAgent 原型的设计思路与实战代码。2. 解构 LedgerAgent账本、状态与策略的三位一体要理解 LedgerAgent必须厘清三个核心概念结构化状态Structured State、策略Policy和账本Ledger。这三者构成了一个闭环的控制系统。2.1 结构化状态从混沌到秩序智能体在运行过程中会产生大量信息用户输入、自身思考Chain-of-Thought、工具调用请求、工具调用结果、环境反馈等。在普通智能体中这些信息可能以非结构化的文本序列或简单的字典形式存在。LedgerAgent 要求我们将这些信息结构化。一个典型的结构化状态条目State Entry可能包含以下字段entry_id: 条目的唯一标识符UUID或自增ID。timestamp: 条目创建的时间戳。agent_phase: 智能体所处的阶段如“THINKING”、“TOOL_CALL_REQUEST”、“TOOL_CALL_RESULT”、“ACTION”、“OBSERVATION”。content: 该阶段的具体内容。例如在TOOL_CALL_REQUEST阶段content可能是一个结构化的 JSON包含tool_name、tool_parameters在THINKING阶段可能是推理链文本。parent_entry_id: 指向引发当前条目的父条目ID用于构建状态树State Tree追溯决策链。metadata: 其他元数据如会话ID、用户ID、环境变量等。这种结构化的好处是显而易见的状态变得可查询、可分析、可验证。我们可以轻易地回答“智能体在过去的5分钟里调用了哪些工具”、“它是基于哪一步的思考结果做出了这个工具调用决策”2.2 策略定义行为的“交通规则”策略Policy是一组规则或约束条件用于规定智能体什么能做、什么不能做、以及应该以何种方式做。策略可以非常具体例如工具调用策略“禁止调用delete_database工具”“调用query_sensitive_data工具前必须已成功调用user_authentication工具”。数据流策略“工具A的输出结果若包含‘PII’个人身份信息标签则只能作为工具B的输入且工具B必须具有‘PII处理权限’”。会话策略“单次会话中调用付费API的次数不得超过10次”。业务逻辑策略“生成报告前必须已收集‘数据源A’和‘数据源B’的结果”。策略通常以声明式的语言或领域特定语言DSL来定义也可以通过在代码中嵌入检查函数来实现。策略引擎是 LedgerAgent 的“大脑”它负责解读并执行这些规则。2.3 账本不可篡改的执行轨迹记录账本Ledger是结构化状态的持久化存储和序列。它按时间顺序记录了智能体生命周期内的所有状态条目。账本的核心特性是“只追加”Append-Only。一旦一个状态条目被写入账本就不应被修改或删除。这保证了执行轨迹的完整性和可审计性就像区块链的分布式账本一样为事后追溯和问责提供了铁证。账本可以是内存中的数据结构如列表、数据库中的一张表、一个文件如JSON Lines格式甚至是分布式的日志系统如Apache Kafka。选择哪种实现取决于对持久性、性能和查询能力的需求。三者的工作流关系智能体产生一个新的意图或行动例如决定调用一个工具。该意图被格式化为一个结构化的状态条目TOOL_CALL_REQUEST。在真正执行调用之前这个待定的状态条目连同当前的完整账本历史被提交给策略引擎进行验证。策略引擎遍历所有相关策略检查如果将此条目加入账本是否会违反任何规则。例如检查该工具是否在允许列表内参数是否合规调用频率是否超限等。验证通过该条目被正式“记账”写入账本智能体随后执行真正的工具调用并将调用结果作为新的状态条目TOOL_CALL_RESULT再次记入账本。验证拒绝条目被拒绝写入账本智能体收到一个策略违规错误必须调整其决策例如选择其他工具或参数。这个“提议 - 验证 - 记账 - 执行”的循环是 LedgerAgent 确保策略遵从性的关键机制。3. 实战构建一个简易 LedgerAgent 原型设计与实现理论讲完了我们来点实际的。我将设计一个基于 Python、使用 LangChain 框架因其在工具调用方面的生态成熟的简易 LedgerAgent 原型。这个原型将清晰地展示上述核心概念。3.1 系统架构与核心类设计我们将创建几个核心类StructuredStateEntry: 表示一个状态条目。Ledger: 负责存储和管理状态条目序列。Policy与PolicyEngine: 定义策略规则和验证引擎。LedgerAgent: 继承或包装一个基础智能体如 LangChain 的AgentExecutor在其决策循环中插入账本记录和策略检查。from datetime import datetime from enum import Enum from typing import Any, Dict, List, Optional, Tuple from uuid import uuid4, UUID from pydantic import BaseModel, Field from langchain.agents import AgentExecutor, BaseSingleActionAgent from langchain.schema import AgentAction, AgentFinish class AgentPhase(str, Enum): 智能体执行阶段枚举 THINKING THINKING TOOL_CALL_REQUEST TOOL_CALL_REQUEST TOOL_CALL_RESULT TOOL_CALL_RESULT ACTION ACTION OBSERVATION OBSERVATION ERROR ERROR POLICY_VIOLATION POLICY_VIOLATION class StructuredStateEntry(BaseModel): 结构化状态条目数据模型 entry_id: UUID Field(default_factoryuuid4) timestamp: datetime Field(default_factorydatetime.utcnow) phase: AgentPhase content: Dict[str, Any] # 结构化内容如 {tool_name: search, args: {...}} parent_entry_id: Optional[UUID] None # 形成状态树 metadata: Dict[str, Any] Field(default_factorydict) class Config: use_enum_values True class Ledger: 账本管理结构化状态条目的只追加存储 def __init__(self): self._entries: List[StructuredStateEntry] [] def append(self, entry: StructuredStateEntry) - None: 只追加写入条目 self._entries.append(entry) def get_all(self) - List[StructuredStateEntry]: 获取所有条目按时间顺序 return self._entries.copy() def get_since(self, entry_id: Optional[UUID] None) - List[StructuredStateEntry]: 获取自某个条目之后的所有条目用于增量策略检查 if entry_id is None: return self.get_all() for i, entry in enumerate(self._entries): if entry.entry_id entry_id: return self._entries[i1:] return [] class PolicyViolation(Exception): 策略违反异常 def __init__(self, message: str, entry: StructuredStateEntry): super().__init__(message) self.violating_entry entry class Policy: 策略基类所有具体策略需继承此类 def check(self, proposed_entry: StructuredStateEntry, ledger: Ledger) - Optional[str]: 检查待添加的条目是否违反本策略。 返回 None 表示通过返回 str 表示违反原因。 此方法接收待添加条目和当前完整账本。 raise NotImplementedError class ToolWhitelistPolicy(Policy): 工具白名单策略只允许调用指定工具 def __init__(self, allowed_tools: List[str]): self.allowed_tools set(allowed_tools) def check(self, proposed_entry: StructuredStateEntry, ledger: Ledger) - Optional[str]: if proposed_entry.phase AgentPhase.TOOL_CALL_REQUEST: tool_name proposed_entry.content.get(tool_name) if tool_name and tool_name not in self.allowed_tools: return fTool {tool_name} is not in the whitelist. Allowed: {list(self.allowed_tools)} return None class SequentialDependencyPolicy(Policy): 顺序依赖策略调用工具B之前必须先成功调用工具A def __init__(self, prerequisite_tool: str, dependent_tool: str): self.prerequisite prerequisite_tool self.dependent dependent_tool def check(self, proposed_entry: StructuredStateEntry, ledger: Ledger) - Optional[str]: if proposed_entry.phase AgentPhase.TOOL_CALL_REQUEST: tool_name proposed_entry.content.get(tool_name) if tool_name self.dependent: # 检查历史账本中是否有成功的 prerequisite 工具调用 has_prerequisite any( e.phase AgentPhase.TOOL_CALL_RESULT and e.content.get(tool_name) self.prerequisite and e.content.get(success) is True for e in ledger.get_all() ) if not has_prerequisite: return fCannot call {self.dependent} before {self.prerequisite} has been successfully called. return None class PolicyEngine: 策略引擎组合并执行所有策略 def __init__(self, policies: List[Policy]): self.policies policies def validate(self, proposed_entry: StructuredStateEntry, ledger: Ledger) - None: 验证待添加条目违反任何策略则抛出 PolicyViolation 异常 for policy in self.policies: violation_reason policy.check(proposed_entry, ledger) if violation_reason: # 记录策略违反条目 violation_entry StructuredStateEntry( phaseAgentPhase.POLICY_VIOLATION, content{ policy_type: policy.__class__.__name__, reason: violation_reason, blocked_entry: proposed_entry.dict() }, parent_entry_idproposed_entry.entry_id ) ledger.append(violation_entry) # 将违规记录也记入账本 raise PolicyViolation(violation_reason, proposed_entry)3.2 集成 LangChain Agent打造 LedgerAgentExecutor现在我们将上述组件与 LangChain 的AgentExecutor集成。关键点在于重写或 Hook_take_next_step这类核心方法在智能体决定调用工具AgentAction和得到工具结果后插入账本记录和策略检查。class LedgerAgentExecutor(AgentExecutor): 带账本和策略检查的 AgentExecutor。 继承自 LangChain 的 AgentExecutor重写关键步骤以集成结构化状态管理。 def __init__(self, agent: BaseSingleActionAgent, tools: List, policy_engine: PolicyEngine, **kwargs): super().__init__(agentagent, toolstools, **kwargs) self.ledger Ledger() self.policy_engine policy_engine self._current_parent_id: Optional[UUID] None def _record_entry(self, phase: AgentPhase, content: Dict, parent_id: Optional[UUID] None) - StructuredStateEntry: 创建并记录一个状态条目 entry StructuredStateEntry(phasephase, contentcontent, parent_entry_idparent_id or self._current_parent_id) # 如果是 TOOL_CALL_REQUEST需通过策略引擎验证 if phase AgentPhase.TOOL_CALL_REQUEST: try: self.policy_engine.validate(entry, self.ledger) except PolicyViolation as e: # 验证失败记录异常并直接返回阻止后续工具调用 self.ledger.append(e.violating_entry) raise e # 验证通过或无需验证的阶段正式记账 self.ledger.append(entry) # 更新当前父ID用于链式记录 if phase in [AgentPhase.TOOL_CALL_REQUEST, AgentPhase.THINKING]: self._current_parent_id entry.entry_id elif phase AgentPhase.TOOL_CALL_RESULT: self._current_parent_id None # 结果条目后重置父ID return entry async def _atake_next_step(self, name_to_tool_map: Dict[str, Any], inputs: Dict[str, str]) - List[Tuple[AgentAction, str]]: 重写异步的下一步执行方法。 这是核心拦截点在智能体决定行动和得到观察结果时插入记账逻辑。 # 1. 智能体“思考”并决定下一步行动 # 这里调用父类方法获取原始的 AgentAction 或 AgentFinish next_step_output await super()._atake_next_step(name_to_tool_map, inputs) # next_step_output 可能是一个 AgentAction调用工具或 AgentFinish结束 for agent_action, output in next_step_output: if isinstance(agent_action, AgentAction): # 2. 记录工具调用请求TOOL_CALL_REQUEST tool_call_entry self._record_entry( phaseAgentPhase.TOOL_CALL_REQUEST, content{ tool_name: agent_action.tool, tool_input: agent_action.tool_input, log: agent_action.log } ) # 3. 执行工具调用父类逻辑 # 注意实际的工具调用在父类的 _execute_agent_action 中我们通过重写 _record_entry 已经做了策略检查。 # 如果策略检查失败_record_entry 会抛出 PolicyViolation此处不会执行到。 observation await self._arun_tool(agent_action, name_to_tool_map) # 4. 记录工具调用结果TOOL_CALL_RESULT self._record_entry( phaseAgentPhase.TOOL_CALL_RESULT, content{ tool_name: agent_action.tool, input: agent_action.tool_input, observation: observation, success: True # 简化处理实际应根据工具执行是否异常判断 }, parent_idtool_call_entry.entry_id ) # 5. 将观察结果返回给智能体进行下一轮思考 return [(agent_action, observation)] elif isinstance(agent_action, AgentFinish): # 任务结束记录最终动作 self._record_entry( phaseAgentPhase.ACTION, content{return_values: agent_action.return_values, log: agent_action.log} ) return [(agent_action, )] return [] # 同步方法 _take_next_step 也需要类似重写此处省略原理相同。3.3 运行示例与账本分析假设我们有两个工具get_user_profile获取用户资料和query_payment_history查询支付记录。我们制定策略必须成功调用get_user_profile后才能调用query_payment_history。from langchain.agents import initialize_agent, Tool from langchain.llms import OpenAI # 1. 定义工具模拟 def get_user_profile(user_id: str) - str: return fProfile of user {user_id}: Active, VIP member. def query_payment_history(user_id: str) - str: return fPayment history for {user_id}: 3 transactions in 2024. tools [ Tool(nameGetUserProfile, funcget_user_profile, descriptionGet user profile by ID), Tool(nameQueryPaymentHistory, funcquery_payment_history, descriptionQuery users payment history), ] # 2. 初始化基础LLM和智能体 llm OpenAI(temperature0) # 假设使用OpenAI模型 base_agent initialize_agent(tools, llm, agentzero-shot-react-description, verboseTrue) # 3. 定义策略 policies [ SequentialDependencyPolicy(prerequisite_toolGetUserProfile, dependent_toolQueryPaymentHistory), ] policy_engine PolicyEngine(policies) # 4. 创建 LedgerAgentExecutor ledger_agent LedgerAgentExecutor(agentbase_agent.agent, toolstools, policy_enginepolicy_engine, max_iterations5, verboseTrue) # 5. 运行一个任务 try: result ledger_agent.run(先获取用户123的资料然后查询他的支付记录。) print(任务结果:, result) except PolicyViolation as e: print(f策略违规原因{e}) # 6. 查看完整的执行账本 print(\n 完整账本记录 ) for entry in ledger_agent.ledger.get_all(): print(f[{entry.timestamp.isoformat()}] {entry.phase}: {entry.content})预期输出与账本分析 如果智能体首先尝试调用QueryPaymentHistory策略引擎会在TOOL_CALL_REQUEST阶段拦截抛出PolicyViolation异常并在账本中记录一条POLICY_VIOLATION条目。智能体如果设计良好应该能根据这个错误调整其计划转而先调用GetUserProfile。一个成功的账本可能如下所示[2024-05-27T10:00:00] THINKING: {reasoning: 用户要求先获取资料再查记录我应该先调用GetUserProfile。} [2024-05-27T10:00:01] TOOL_CALL_REQUEST: {tool_name: GetUserProfile, tool_input: {user_id: 123}} [2024-05-27T10:00:02] TOOL_CALL_RESULT: {tool_name: GetUserProfile, observation: Profile of user 123..., success: true} [2024-05-27T10:00:03] THINKING: {reasoning: 已获取资料现在可以安全地查询支付记录了。} [2024-05-27T10:00:04] TOOL_CALL_REQUEST: {tool_name: QueryPaymentHistory, tool_input: {user_id: 123}} [2024-05-27T10:00:05] TOOL_CALL_RESULT: {tool_name: QueryPaymentHistory, observation: Payment history for 123..., success: true} [2024-05-27T10:00:06] ACTION: {return_values: {output: 已完成查询。}}这个账本清晰地展示了智能体的完整思维和行动轨迹并且每一步都隐含了策略合规的证明因为违规的请求根本无法被记录。4. 深入策略引擎从简单规则到复杂逻辑上面的例子展示了基于当前待记条目和完整历史账本的策略检查。但在实际生产中策略可能复杂得多。4.1 基于状态的策略Stateful Policies有些策略需要维护自己的内部状态。例如一个“限流策略”需要记录单位时间内的调用次数。class RateLimitPolicy(Policy): 限流策略限制特定工具在时间窗口内的调用次数 def __init__(self, tool_name: str, max_calls: int, window_seconds: int): self.tool_name tool_name self.max_calls max_calls self.window timedelta(secondswindow_seconds) def check(self, proposed_entry: StructuredStateEntry, ledger: Ledger) - Optional[str]: if proposed_entry.phase AgentPhase.TOOL_CALL_REQUEST: if proposed_entry.content.get(tool_name) self.tool_name: now proposed_entry.timestamp window_start now - self.window # 查询账本统计在时间窗口内该工具的调用成功次数 call_count sum( 1 for e in ledger.get_all() if e.phase AgentPhase.TOOL_CALL_RESULT and e.content.get(tool_name) self.tool_name and e.content.get(success) is True and window_start e.timestamp now ) if call_count self.max_calls: return fRate limit exceeded for {self.tool_name}. Max {self.max_calls} calls per {self.window.seconds}s. return None4.2 复合策略与策略优先级现实中的策略往往是复合的。我们可以设计一个CompositePolicy它包含多个子策略并定义评估顺序如短路评估一个失败则整体失败和冲突解决机制。class CompositePolicy(Policy): 组合策略按顺序评估多个子策略 def __init__(self, policies: List[Policy], evaluation_mode: str all_pass): # all_pass 或 first_fail self.policies policies self.mode evaluation_mode def check(self, proposed_entry: StructuredStateEntry, ledger: Ledger) - Optional[str]: reasons [] for policy in self.policies: reason policy.check(proposed_entry, ledger) if reason: if self.mode first_fail: return fCompositePolicy(first_fail): {reason} reasons.append(f[{policy.__class__.__name__}] {reason}) if reasons: return fCompositePolicy(all_pass) violations:\n \n.join(reasons) return None4.3 策略的动态加载与更新在生产环境中策略可能需要热更新。我们可以将策略定义存储在外部如数据库、配置文件PolicyEngine定期 reload。或者更高级的做法是让策略本身也是一个可以被智能体查询的“工具”但对其修改权限施加更严格的管控。5. 性能、扩展性与生产级考量将每个动作都进行结构化记录和策略检查无疑会引入开销。在设计生产级 LedgerAgent 时以下几点至关重要5.1 账本存储的优化分级存储将高频访问的最新条目放在内存或 Redis 中历史条目归档到对象存储如 S3或时序数据库。索引优化为账本条目建立索引如phase,tool_name,timestamp加速策略引擎的历史查询。可以使用专门的文档数据库如 MongoDB、Elasticsearch来存储账本。异步写入策略检查必须同步以保证决策合规但条目的持久化存储可以异步进行以提高主循环的响应速度。需要处理好异步写入失败的重试和一致性补偿。5.2 策略检查的优化增量检查策略引擎不需要每次都对完整账本进行全量扫描。Ledger.get_since()方法提供了增量条目。许多策略如顺序依赖只需要检查最新的一部分条目或与当前条目相关的条目。策略编译与预计算将声明式策略编译成高效的可执行代码如生成决策树或状态机。对于限流类策略可以维护一个内存中的滑动窗口计数器而不是每次都查询账本。并行检查如果策略之间没有依赖可以在多个线程或进程中并行执行检查。5.3 与现有监控、审计系统的集成Ledger 本身就是一个强大的审计日志源。可以很容易地将其条目转发到现有的日志聚合系统如 ELK Stack、监控系统如 Prometheus或安全信息与事件管理SIEM系统。例如每一个POLICY_VIOLATION条目都可以触发一个高优先级的告警。5.4 对智能体“思考”过程的策略约束目前的策略主要约束“行动”工具调用。更进一步的我们可以尝试约束智能体的“思考”推理过程。例如可以要求智能体在思考链中必须包含某些关键词如“检查权限”或者禁止其思考某些敏感话题。这可以通过对THINKING阶段的内容进行自然语言处理NLP或嵌入向量相似度检查来实现虽然技术上更具挑战性但代表了更深度的策略控制。6. 总结LedgerAgent 的价值与未来展望通过以上深入的探讨和实战构建我们可以看到 LedgerAgent 模式为 Tool-Calling Agents 带来了根本性的改变可审计性结构化的账本提供了机器可读、不可篡改的完整执行轨迹满足合规与审计的硬性要求。可控性策略引擎在决策点进行实时拦截将安全与合规要求从“事后补救”变为“事前预防”和“事中控制”。可解释性当智能体行为出现偏差时开发者、审计员或用户可以通过审查账本和关联的策略精准定位问题根源是工具问题、策略缺陷还是智能体逻辑错误。可测试性可以基于账本回放智能体的执行过程进行回归测试和压力测试验证策略修改的影响。在实际操作中我发现在引入 LedgerAgent 模式后调试复杂智能体的效率显著提升。以前需要反复查看冗长的非结构化日志现在可以直接查询账本数据库用 SQL 或特定查询语言快速定位“所有调用了敏感工具 X 的会话”或“所有违反了策略 Y 的请求”。同时它也促使我们在项目早期就更加严谨地思考智能体的行为边界定义清晰的策略而不是等到出了问题再打补丁。当然这套架构也会增加系统的复杂性。对于简单的、非关键的智能体应用可能显得有些“杀鸡用牛刀”。但对于企业级、金融级、涉及敏感操作的应用这种对“结构化状态”和“策略遵从”的投入是必要且值得的。未来我期待看到更多开源框架和云服务将 LedgerAgent 的理念内置为一种标准选项并提供更强大的策略语言、可视化审计工具和性能优化方案让构建安全、可靠、可信的智能体应用成为所有开发者的标配能力。