Agent 工程化：当大模型走出“Chat”的温室，走进“铁与血”的生产线

from pydantic import BaseModel, Field
from typing import Literal

# 工程化的核心：定义“铁律”般的行动指令集
class ActionCommand(BaseModel):
    tool_name: Literal["query_weather", "send_email", "create_ticket"]
    target_city: str = Field(..., max_length=20, description="必须是中国省会城市")
    priority: Literal[1, 2, 3] = Field(..., description="1最高，3最低")
    content: str = Field(..., min_length=1)

# 代码中的这一句，就是工程化的“保险丝”
# 只要 LLM 输出的字段不对，解析器会直接抛出异常并触发重试机制，
# 而不是把一个错误的 Json 灌入下游系统。

import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
async def call_agent_with_retry(user_query):
    try:
        # 尝试调用核心 Agent 推理逻辑
        result = await agent.execute(user_query)
        return result
    except (TimeoutError, ConnectionError) as e:
        # 工程化的智慧：不盲目死磕，尝试简化上下文重新请求
        if "token_overflow" in str(e):
            compressed_query = summarize_context(user_query) # 触发上下文压缩
            return await agent.execute(compressed_query)
        raise  # 若依旧失败，透传至最外层兜底机制

class MemoryGuard:
    def __init__(self, max_tokens=8000):
        self.max_tokens = max_tokens
    
    def check_and_trim(self, message_list):
        # 工程化的“水位线”判断：一旦逼近极限，立即触发硬性截断
        if estimate_tokens(message_list) > self.max_tokens * 0.8:
            # 保留系统指令 + 最近 5 轮对话，其余暴力压缩为 200 字的文本摘要
            return force_summarize(message_list[:-5]) + message_list[-5:]
        return message_list

def orchestrate_loop(initial_input):
    state = {"phase": "PLAN", "retry_count": 0}
    
    while state["phase"] != "DONE":
        if state["phase"] == "PLAN":
            plan = llm.generate_plan(initial_input)
            state["phase"] = "ACTION"
        
        elif state["phase"] == "ACTION":
            result = execute_tool(plan.tool_id)
            # 工程化关键：引入评审节点，不信任任何一次执行结果
            if llm.critic(result) < 0.6 and state["retry_count"] < 2:
                state["phase"] = "REFLECT"  # 转向反思修正
                state["retry_count"] += 1
            else:
                state["phase"] = "DONE"
        
        elif state["phase"] == "REFLECT":
            # 修正错过的参数，重回 PLAN
            plan = llm.correct_plan(result, plan)
            state["phase"] = "ACTION"