AI Agent Harness 架构详解：构建可靠的智能体执行环境【2026】

import OpenAI from "openai";

const openai = new OpenAI({ apiKey: process.env.OPENAI_API_KEY });

// 1. 工具注册表
const tools = {
  getWeather: async ({ location }) => {
    // 模拟 API 调用
    return `${location} 的天气是 22°C，晴朗。`;
  }
};

const toolDefinitions = [{
  type: "function",
  function: {
    name: "getWeather",
    description: "获取指定位置的当前天气",
    parameters: {
      type: "object",
      properties: { location: { type: "string" } },
      required: ["location"],
    },
  },
}];

// 2. Harness 核心循环
async function runAgentHarness(userPrompt) {
  let messages = [{ role: "user", content: userPrompt }];
  let isDone = false;
  let maxSteps = 5; // 核心安全约束
  
  while (!isDone && maxSteps > 0) {
    maxSteps--;
    
    // 调用 LLM（"大脑"）
    const response = await openai.chat.completions.create({
      model: "gpt-4o",
      messages: messages,
      tools: toolDefinitions,
    });
    
    const responseMessage = response.choices[0].message;
    messages.push(responseMessage);
    
    // 3. 执行引擎
    if (responseMessage.tool_calls) {
      for (const toolCall of responseMessage.tool_calls) {
        const tool = tools[toolCall.function.name];
        if (!tool) throw new Error("unsupported_tool");
        const args = JSON.parse(toolCall.function.arguments);
        if (
          typeof args?.location !== "string" ||
          args.location.length < 1 ||
          args.location.length > 100
        ) {
          throw new Error("invalid_tool_arguments");
        }
        const result = await tool(args);
        
        // 将状态和结果反馈给 LLM
        messages.push({
          role: "tool",
          tool_call_id: toolCall.id,
          name: toolCall.function.name,
          content: result,
        });
      }
    } else {
      isDone = true;
      return responseMessage.content;
    }
  }
  
  if (maxSteps === 0) throw new Error("Agent 达到最大执行步数限制（无限循环保护触发）");
  return messages[messages.length - 1].content;
}

// 启动 Harness
const finalOutput = await runAgentHarness("旧金山的天气怎么样？");
console.log(finalOutput); 
// 预期输出: "旧金山目前的当前天气是 22°C，晴朗。"

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI

# 1. 定义状态
class AgentState(TypedDict):
    messages: list
    scratchpad: str

# 2. 定义 Harness 节点
def call_model(state: AgentState):
    llm = ChatOpenAI(model="gpt-4o")
    response = llm.invoke(state["messages"])
    return {"messages": state["messages"] + [response]}

def should_continue(state: AgentState):
    last_message = state["messages"][-1]
    if "tool_calls" in last_message.additional_kwargs:
        return "execute_tools"
    return END

# 3. 构建 Harness 图
workflow = StateGraph(AgentState)
workflow.add_node("agent", call_model)
# ... 此处添加工具执行节点 ...

workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_continue)

# 编译 Harness 运行环境
app = workflow.compile()