AI Agent运行时架构：从Function Calling到多层级ReAct的工程实践

2024年，AI Agent从概念验证走向生产部署。Claude、GPT-4、Gemini等大模型通过Function Calling暴露工具调用能力，但如何围绕这一能力构建可靠的运行时架构，仍是工程实践中的核心挑战。本文深入解析AI Agent的运行时设计，涵盖从单循环ReAct到多Agent协作的完整工程化路径。

一、Agent核心循环：不只是"思考-行动"

最简Agent架构是一个循环：接收目标→LLM推理→执行工具→观察结果→再次推理。这个模式看似简单，但在生产环境中会遇到三大工程挑战：状态爆炸、不可恢复错误和上下文窗口泄漏。

class MinimalAgent:
    def __init__(self, llm, tools, system_prompt):
        self.llm = llm
        self.tools = {t.name: t for t in tools}
        self.system_prompt = system_prompt

    def run(self, user_query: str, max_steps: int = 10):
        messages = [
            {"role": "system", "content": self.system_prompt},
            {"role": "user", "content": user_query}
        ]

        for step in range(max_steps):
            response = self.llm.chat(
                messages=messages,
                tools=self._tool_definitions()
            )

            # 终止条件：模型未请求工具调用
            if not response.tool_calls:
                return response.content

            messages.append(response.to_message())

            for call in response.tool_calls:
                result = self._execute_tool(call)
                messages.append({
                    "role": "tool",
                    "tool_call_id": call.id,
                    "content": str(result)
                })

        return "达到最大步数限制"

这段代码能工作，但距离生产级部署差距甚远。真实场景需要处理：工具执行超时、LLM幻觉导致的无效工具参数、上下文窗口溢出、中间状态持久化等问题。

二、工具注册体系：从Decorator到类型安全

工具的注册方式决定了Agent的可维护性。一个生产级工具系统需要满足三个条件：类型安全、权限边界和调用审计。

from dataclasses import dataclass
from typing import Callable, Any, get_type_hints
import inspect
import json

@dataclass
class ToolSpec:
    name: str
    description: str
    parameters_schema: dict
    handler: Callable
    permission_level: str = "user"  # user, elevated, admin

    def execute(self, **kwargs) -> Any:
        # 类型校验
        hints = get_type_hints(self.handler)
        for key, value in kwargs.items():
            if key in hints and not isinstance(value, hints[key]):
                raise TypeError(
                    f"参数 {key} 期望类型 {hints[key].__name__}, "
                    f"实际得到 {type(value).__name__}"
                )
        return self.handler(**kwargs)

class ToolRegistry:
    def __init__(self):
        self._tools: dict[str, ToolSpec] = {}
        self._call_log: list[dict] = []

    def register(self, name: str, description: str, 
                 permission: str = "user"):
        def decorator(func: Callable):
            sig = inspect.signature(func)
            schema = self._build_schema(sig, func)
            self._tools[name] = ToolSpec(
                name=name,
                description=description,
                parameters_schema=schema,
                handler=func,
                permission_level=permission
            )
            return func
        return decorator

    def _build_schema(self, sig, func) -> dict:
        """从函数签名自动生成JSON Schema"""
        hints = get_type_hints(func)
        properties = {}
        required = []

        for param_name, param in sig.parameters.items():
            ptype = hints.get(param_name, str)
            prop = {"type": self._map_type(ptype)}
            if param_name in hints:
                prop["description"] = f"参数 {param_name}"
            if param.default is inspect.Parameter.empty:
                required.append(param_name)
            properties[param_name] = prop

        return {
            "type": "object",
            "properties": properties,
            "required": required
        }

    def execute(self, name: str, **kwargs) -> Any:
        if name not in self._tools:
            raise ValueError(f"未知工具: {name}")

        spec = self._tools[name]
        self._call_log.append({
            "tool": name,
            "params": kwargs,
            "permission": spec.permission_level
        })

        return spec.execute(**kwargs)

    @staticmethod
    def _map_type(ptype) -> str:
        mapping = {
            str: "string", int: "integer", 
            float: "number", bool: "boolean",
            list: "array", dict: "object"
        }
        return mapping.get(ptype, "string")

# 使用示例
registry = ToolRegistry()

@registry.register("github_search", "搜索GitHub仓库", permission="user")
def github_search(query: str, language: str = "", max_results: int = 5):
    """调用GitHub API搜索仓库"""
    # API调用实现...
    return {"repositories": []}

@registry.register("deploy_service", "部署服务到Kubernetes", permission="elevated")
def deploy_service(service_name: str, image: str, replicas: int = 3):
    """部署容器服务"""
    # K8s API调用实现...
    return {"status": "deployed"}

这套注册机制的核心优势在于：类型注解自动驱动Schema生成，权限级别在注册时声明而非执行时判断，调用日志自动记录使每次Agent决策都可追溯到具体参数。

三、ReAct的进阶工程化

Anthropic提出的ReAct模式（Reason + Act）是Agent推理的基石。但在实际部署中，纯ReAct会遇到"推理漂移"问题——Agent在长链路推理后忘记原始目标。

3.1 层级化ReAct架构

解决方案是引入计划层，将Agent分为三个抽象层级：

class HierarchicalReActAgent:
    """三层Agent：规划器 → 分解器 → 执行器"""

    def __init__(self):
        self.planner = PlannerLLM()     # 高层：制定子目标
        self.decomposer = TaskLLM()     # 中层：分解为步骤
        self.executor = ExecutorLLM()   # 低层：执行原子操作
        self.global_state = StateStore()

    def run(self, goal: str) -> dict:
        # 第一层：规划
        plan = self.planner.create_plan(goal, context=self.global_state)
        self.global_state.set("current_plan", plan)
        results = []

        for subgoal in plan.subgoals:
            # 第二层：分解
            steps = self.decomposer.decompose(subgoal)

            for step in steps:
                # 第三层：执行 + 自我修正
                result = self._execute_with_backoff(step)

                if result.status == "failed":
                    # 回退到分解层重新规划
                    alternative = self.decomposer.handle_failure(
                        step, result.error, context=steps
                    )
                    result = self._execute_with_backoff(alternative)

                results.append(result)
                self.global_state.append("execution_trace", result)

        return self.planner.synthesize(goal, results)

    def _execute_with_backoff(self, step, max_retries=3):
        """带回退的执行器"""
        for attempt in range(max_retries):
            try:
                return self.executor.execute(step)
            except RateLimitError:
                wait = 2 ** attempt * 10
                time.sleep(wait)
            except ToolExecutionError as e:
                return ExecutionResult(status="failed", error=str(e))

        return ExecutionResult(status="failed", error="超过最大重试次数")

3.2 防止推理漂移：目标锚定机制

class GoalAnchor:
    """在每轮推理中注入原始目标，防止推理漂移"""

    def __init__(self, original_goal: str):
        self.goal = original_goal
        self.drift_threshold = 0.7

    def wrap_messages(self, messages: list, context_window: int) -> list:
        """在消息列表末尾加入目标锚定"""
        anchor = (
            f"[目标锚定] 原始任务: {self.goal}\n"
            f"[进度检查] 当前执行是否仍服务于上述目标？"
            f"如果偏离，请回到正确轨道。"
        )

        # 在最新用户消息后插入锚定
        return messages + [{
            "role": "system",
            "content": anchor
        }]

    def check_drift(self, current_action: str) -> bool:
        """检测当前动作是否偏离原始目标"""
        # 使用小模型或关键词匹配检测偏离
        drift_keywords = ["偏离", "无关", "不确定是否"]
        return any(kw in current_action for kw in drift_keywords)

四、上下文窗口管理：Agent的"工作记忆"

当Agent执行数十步操作后，上下文窗口成为最稀缺资源。一个典型的5步工具调用链产生约15条消息（每步：assistant call + tool result + 推理追加），每条200-500 tokens，快速逼近窗口上限。

4.1 滑动窗口 + 摘要压缩

class ContextManager:
    """管理Agent的上下文窗口"""

    def __init__(self, max_tokens: int = 100000, 
                 llm_for_summary=None):
        self.max_tokens = max_tokens
        self.summarizer = llm_for_summary
        self.messages = []
        self.summaries = []  # 历史摘要栈

    def add(self, message: dict):
        self.messages.append(message)
        if self._total_tokens() > self.max_tokens * 0.8:
            self._compress()

    def _compress(self):
        """将早期消息压缩为摘要"""
        # 保留最近N条消息，其余压缩
        keep_recent = 6
        to_summarize = self.messages[:-keep_recent]

        summary_prompt = (
            f"将以下Agent交互历史压缩为结构化摘要：\n"
            f"- 已完成的动作（按时间顺序）\n"
            f"- 关键发现和中间结果\n"
            f"- 当前子目标进展\n"
            f"- 待完成的事项\n\n"
            f"{json.dumps(to_summarize, ensure_ascii=False, indent=2)}"
        )

        summary = self.summarizer.complete(summary_prompt)
        self.summaries.append(summary)
        self.messages = self.messages[-keep_recent:]

    def get_context(self) -> list:
        """组装最终上下文"""
        context = []
        if self.summaries:
            context.append({
                "role": "system",
                "content": f"[历史摘要]\n" + "\n---\n".join(self.summaries)
            })
        context.extend(self.messages)
        return context

    def _total_tokens(self) -> int:
        # 简化token计算，实际使用tiktoken
        return sum(len(m.get("content", "")) for m in self.messages) // 4

4.2 结构化记忆存储

除滑动窗口外，Agent还需要外部记忆来持久化关键信息：

class AgentMemory:
    """Agent的外部记忆系统"""

    def __init__(self):
        self.working_memory = {}      # 当前任务上下文
        self.episodic_memory = []      # 交互历史记录
        self.semantic_memory = {}      # 结构化知识库

    def remember(self, key: str, value: Any, 
                 memory_type: str = "working"):
        if memory_type == "working":
            self.working_memory[key] = value
        elif memory_type == "episodic":
            self.episodic_memory.append({
                "key": key, "value": value, "timestamp": time.time()
            })
        elif memory_type == "semantic":
            self.semantic_memory[key] = value

    def recall(self, key: str, memory_type: str = "working") -> Any:
        store = {
            "working": self.working_memory,
            "semantic": self.semantic_memory
        }.get(memory_type, {})
        return store.get(key)

    def format_for_context(self) -> str:
        """格式化为LLM可消费的上下文片段"""
        lines = ["[工作记忆]"]
        for k, v in self.working_memory.items():
            lines.append(f"  {k}: {v}")
        if self.semantic_memory:
            lines.append("[知识库]")
            for k, v in list(self.semantic_memory.items())[:5]:
                lines.append(f"  {k}: {v}")
        return "\n".join(lines)

五、多Agent协作：从编排到涌现

当任务复杂度超过单个Agent的处理能力时，需要引入多Agent协作。主流架构有三种模式：

5.1 Orchestrator-Worker模式

class OrchestratorAgent:
    """编排型Agent：分解任务并分配给Worker"""

    def __init__(self, workers: dict[str, BaseAgent]):
        self.workers = workers  # {"code_reviewer": CodeReviewer(), ...}

    def run(self, task: ComplexTask):
        subtasks = self.decompose(task)
        results = {}

        # 并行执行无依赖的子任务
        with ThreadPoolExecutor(max_workers=4) as pool:
            futures = {}
            for subtask in subtasks:
                if self._can_run(subtask, results):
                    worker = self.workers[subtask.assigned_to]
                    future = pool.submit(worker.run, subtask)
                    futures[future] = subtask

            for future in as_completed(futures):
                subtask = futures[future]
                results[subtask.id] = future.result()

        return self.merge_results(results)

5.2 辩论模式（Debate Pattern）

适用于高风险决策场景，多个Agent从不同角度分析后达成共识：

class DebateOrchestrator:
    def run(self, question: str, rounds: int = 3):
        agents = [
            RiskAverseAgent(),    # 偏保守角度
            InnovationAgent(),    # 偏创新角度  
            PracticalAgent()      # 偏实用角度
        ]

        positions = {a.name: a.initial_position(question) 
                    for a in agents}

        for r in range(rounds):
            for agent in agents:
                others = {k: v for k, v in positions.items() 
                         if k != agent.name}
                positions[agent.name] = agent.respond_to_others(
                    question, others, positions[agent.name]
                )

        return self.synthesize_consensus(positions)

六、生产级Agent的可靠性工程

将Agent部署到生产环境，需要面对与传统软件不同的可靠性挑战：LLM的非确定性输出导致相同输入可能产生不同行为路径。

6.1 幂等工具设计

def idempotent_tool(func):
    """确保工具幂等性的装饰器"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        call_id = kwargs.get("__call_id")
        if call_id:
            cached = check_result_cache(call_id)
            if cached is not None:
                return cached
        result = func(*args, **kwargs)
        if call_id:
            store_result_cache(call_id, result, ttl=3600)
        return result
    return wrapper

6.2 可观测性体系

from contextlib import contextmanager
import uuid

class AgentTracer:
    """Agent运行时的全链路追踪"""

    @contextmanager
    def trace_run(self, agent_name: str, goal: str):
        run_id = str(uuid.uuid4())[:8]
        self._emit({
            "event": "agent_run_start",
            "run_id": run_id,
            "agent": agent_name,
            "goal_preview": goal[:100]
        })
        start = time.time()

        try:
            yield AgentRunContext(run_id=run_id, tracer=self)
            self._emit({
                "event": "agent_run_complete",
                "run_id": run_id,
                "duration_ms": (time.time() - start) * 1000
            })
        except Exception as e:
            self._emit({
                "event": "agent_run_failed",
                "run_id": run_id,
                "error": str(e),
                "duration_ms": (time.time() - start) * 1000
            })
            raise

    def trace_tool_call(self, tool_name: str, params: dict, 
                       result: Any, duration_ms: float):
        self._emit({
            "event": "tool_call",
            "tool": tool_name,
            "param_keys": list(params.keys()),
            "result_type": type(result).__name__,
            "duration_ms": duration_ms
        })

七、前沿方向：从Reactive到Proactive

2024年Agent架构正从"被动响应"向"主动感知"演进：

• Long-running Agent：突破单次会话限制，Agent可以跨小时/天持续运行，监控外部事件并主动触发工作流
• Sub-Agent Spawning：父Agent根据任务需要动态创建子Agent实例，形成Agent树结构，结束后自动回收资源
• Reflexive Self-Improvement：Agent将执行过程中遇到的错误和纠正策略写入持久记忆，在后续运行中自动避免重复犯错
• Multi-Modal Perception：Agent从纯文本输入扩展到视觉、音频、屏幕流等多模态感知，实现"看屏操作"

总结

AI Agent运行时的工程复杂度被严重低估。一个生产级Agent系统需要在工具注册、上下文管理、记忆系统、可靠性保障四个维度建立完整的工程化方案。核心设计原则是：将LLM视为一个强大但不可靠的核心，用工程手段兜底不确定性。

从单循环ReAct到层级化架构，从单Agent到多Agent协作，每一步演化都围绕同一个命题：如何让非确定性的LLM推理在确定性工程框架内可靠运行。这不仅是技术问题，更是工程哲学的体现。