2025年，几乎每家公司都做了一个AI Agent的Demo。到了2026年，那些Demo中只有不到30%能真正进入生产环境——而成功的那部分，靠的不是更聪明的模型，而是更扎实的基础设施与运维能力。

把Agent放进生产环境只是开始。它能不能在真实业务中持续被观测、被评估、被优化，才决定了企业能否真正吃到AI红利。本文将从提示设计（Prompt Engineering）、记忆管控（Memory Management）和可观测性（Observability） 三个维度，系统拆解生产级Agent开发的核心技术与实战代码。

一、提示设计：从“写指令”到“搭框架”

1.1 为什么Prompt在“生产级”语境下变了

在Demo阶段，Prompt的任务是“让模型理解并完成任务”。但在生产环境中，Prompt的任务变成了“让模型在不可预测的输入下稳定输出可预期的结果” 。

两者的区别在于：Demo追求“最好的一次”，生产追求“最差的一次也不能崩”。企业Agent从Demo到生产的“拉胯”根因，往往在于Prompt只解决了“说清任务”，却没有解决“当模型犯错时系统怎么兜底”。

1.2 结构化输出：生产环境的硬要求

生产系统需要机器可解析的输出。以下是一个使用Pydantic强制结构化输出的示例：

from pydantic import BaseModel, Field
from typing import List, Optional
from langchain.output_parsers import PydanticOutputParser
from langchain.prompts import PromptTemplate
from langchain.chat_models import ChatOpenAI

# 1. 定义输出结构
class AgentResponse(BaseModel):
    """Agent的标准输出格式"""
    reasoning: str = Field(description="推理过程摘要")
    action: str = Field(description="要执行的动作名称")
    action_input: dict = Field(description="动作的参数")
    confidence: float = Field(ge=0.0, le=1.0, description="置信度")
    fallback: Optional[str] = Field(default=None, description="备用方案")

class BatchResult(BaseModel):
    """批量处理结果"""
    results: List[AgentResponse]
    total_tokens: int
    errors: List[str]

# 2. 构建带解析器的Prompt
parser = PydanticOutputParser(pydantic_object=AgentResponse)

prompt = PromptTemplate(
    template="""你是一个任务执行Agent。请根据用户输入完成任务。
    
    用户输入: {input}
    
    输出格式要求:
    {format_instructions}
    
    注意: 如果无法确定如何行动，将action设为"ask_clarification"并在fallback中说明。
    """,
    input_variables=["input"],
    partial_variables={"format_instructions": parser.get_format_instructions()}
)

# 3. 执行并解析
llm = ChatOpenAI(model="gpt-4", temperature=0)
chain = prompt | llm | parser

result = chain.invoke({"input": "帮我查询今天北京的天气"})
print(f"动作: {result.action}, 参数: {result.action_input}, 置信度: {result.confidence}")

关键设计原则：

• 强制字段兜底：每个输出都必须包含fallback字段，当模型不确定时给出备用方案
• 置信度自评：让模型对自己的输出打分，下游系统可根据置信度决定是否人工介入
• 类型安全：使用Pydantic在运行时校验输出格式，避免JSON解析异常

1.3 系统提示的工程化版本管理

生产环境中，Prompt和代码一样需要版本管理。以下是一个Prompt版本控制的实现框架：

import json
import hashlib
from datetime import datetime
from pathlib import Path

class PromptRegistry:
    """Prompt版本注册与管理"""
    
    def __init__(self, storage_path: str = "./prompts"):
        self.storage_path = Path(storage_path)
        self.storage_path.mkdir(exist_ok=True)
    
    def register(self, name: str, template: str, variables: list, metadata: dict = None):
        """注册一个新版本的Prompt"""
        version_hash = hashlib.sha256(template.encode()).hexdigest()[:8]
        version_info = {
            "name": name,
            "version": version_hash,
            "template": template,
            "variables": variables,
            "metadata": metadata or {},
            "created_at": datetime.now().isoformat(),
            "deprecated": False
        }
        
        version_file = self.storage_path / f"{name}_{version_hash}.json"
        with open(version_file, "w") as f:
            json.dump(version_info, f, indent=2)
        
        # 更新latest指针
        latest_file = self.storage_path / f"{name}_latest.json"
        with open(latest_file, "w") as f:
            json.dump({"latest": version_hash}, f)
        
        return version_hash
    
    def get(self, name: str, version: str = "latest"):
        """获取指定版本的Prompt"""
        if version == "latest":
            latest_file = self.storage_path / f"{name}_latest.json"
            if not latest_file.exists():
                raise ValueError(f"Prompt {name} not found")
            with open(latest_file) as f:
                version = json.load(f)["latest"]
        
        version_file = self.storage_path / f"{name}_{version}.json"
        if not version_file.exists():
            raise ValueError(f"Version {version} not found")
        
        with open(version_file) as f:
            return json.load(f)
    
    def rollback(self, name: str, target_version: str):
        """回滚到指定版本"""
        # 验证目标版本存在
        version_file = self.storage_path / f"{name}_{target_version}.json"
        if not version_file.exists():
            raise ValueError(f"Version {target_version} not found")
        
        latest_file = self.storage_path / f"{name}_latest.json"
        with open(latest_file, "w") as f:
            json.dump({"latest": target_version}, f)
        
        return target_version

1.4 从Prompt到Harness：四层工程演进

生产级Agent的Prompt策略需要放在更大的工程框架中理解。企业Agent从Demo到生产的演进可分为四层：

层级	核心关注	关键实践
Prompt	说清任务、输出格式	结构化输出、置信度自评、Fallback机制
Context	提供正确背景、最优信息策略	RAG检索优化、上下文窗口管理
Harness	引导、感知、执行约束与可观测性	错误结构性不可重犯、执行沙箱
Loop	自动化心跳、持续自主迭代	工作树隔离、子Agent制衡、外部状态同步

真正生产级的Prompt设计，不是写好一句话就完事了，而是设计好一个“当Prompt失效时系统如何自动降级”的闭环。

二、记忆管控：从“无状态”到“有记忆”

2.1 为什么记忆是生产级Agent的命门

一个没有记忆的Agent，每次对话都是一次“失忆”后的重新开始。这在生产环境中意味着：

• 用户需要反复陈述自己的偏好和背景
• Agent无法从历史错误中学习
• 跨会话的任务连续性完全断裂

更严重的是，没有策展的记忆会把一次性错误固化成永久谎言。Databricks在2026年的研究发现，Agent会引用之前运行中错误的输出，再以更高的信心复用。记忆系统需要的不只是“存储”，更是“策展”——知道什么该记住、什么该忘记、什么该修正。

2.2 记忆的五阶段流水线

构建生产级持久化记忆系统，需要遵循抽取→整合→存储→检索→遗忘五阶段流水线。记忆类型可分为四种：

记忆类型	定义	存储方式	生命周期
工作记忆	当前任务执行的上下文	内存/会话缓存	单次会话
情景记忆	完整的交互历史与事件	向量数据库 + 结构化存储	长期持久
语义记忆	提取的知识、事实、用户画像	向量数据库	长期持久
过程记忆	任务执行的经验与技能	结构化规则 + 示例库	长期持久

以下是一个结合短期工作记忆与长期向量记忆的完整实现：

from typing import List, Dict, Any, Optional
from langchain.memory import ConversationBufferMemory
from langchain.vectorstores import Chroma
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.schema import Document
import json
import hashlib
from datetime import datetime

class HybridMemory:
    """混合记忆系统：工作记忆 + 长期向量记忆"""
    
    def __init__(self, persist_dir: str = "./memory_store"):
        self.persist_dir = persist_dir
        self.working_memory = ConversationBufferMemory(
            memory_key="chat_history",
            return_messages=True,
            max_len=20  # 限制工作记忆长度
        )
        
        # 长期记忆：向量存储
        self.embeddings = HuggingFaceEmbeddings(
            model_name="sentence-transformers/all-MiniLM-L6-v2"
        )
        self.long_term_store = Chroma(
            persist_directory=f"{persist_dir}/vectors",
            embedding_function=self.embeddings
        )
        
        # 结构化记忆：用户画像等
        self.structured_memory = {}
        self._load_structured()
    
    def _load_structured(self):
        """加载结构化记忆"""
        try:
            with open(f"{self.persist_dir}/structured.json", "r") as f:
                self.structured_memory = json.load(f)
        except FileNotFoundError:
            self.structured_memory = {
                "user_profile": {},
                "preferences": {},
                "learned_facts": [],
                "error_history": []
            }
    
    def _save_structured(self):
        """保存结构化记忆"""
        with open(f"{self.persist_dir}/structured.json", "w") as f:
            json.dump(self.structured_memory, f, indent=2)
    
    def add_to_working(self, user_input: str, agent_output: str):
        """添加到工作记忆"""
        self.working_memory.save_context(
            {"input": user_input},
            {"output": agent_output}
        )
    
    def add_to_long_term(self, content: str, metadata: Dict[str, Any]):
        """添加到长期向量记忆"""
        doc = Document(
            page_content=content,
            metadata={
                **metadata,
                "timestamp": datetime.now().isoformat(),
                "id": hashlib.md5(content.encode()).hexdigest()[:8]
            }
        )
        self.long_term_store.add_documents([doc])
    
    def add_fact(self, fact: str, category: str = "general"):
        """添加结构化事实"""
        self.structured_memory["learned_facts"].append({
            "fact": fact,
            "category": category,
            "timestamp": datetime.now().isoformat()
        })
        # 限制事实库大小，防止膨胀
        if len(self.structured_memory["learned_facts"]) > 1000:
            self.structured_memory["learned_facts"] = \
                self.structured_memory["learned_facts"][-1000:]
        self._save_structured()
    
    def retrieve_relevant(self, query: str, k: int = 5) -> List[Dict]:
        """检索相关长期记忆"""
        results = self.long_term_store.similarity_search(query, k=k)
        return [{"content": doc.page_content, "metadata": doc.metadata} 
                for doc in results]
    
    def get_context(self, query: str) -> str:
        """构建完整上下文：工作记忆 + 长期记忆 + 结构化事实"""
        context_parts = []
        
        # 1. 工作记忆
        working_context = self.working_memory.load_memory_variables({})
        if working_context.get("chat_history"):
            context_parts.append("=== 近期对话历史 ===")
            context_parts.append(str(working_context["chat_history"]))
        
        # 2. 长期记忆检索
        relevant = self.retrieve_relevant(query, k=3)
        if relevant:
            context_parts.append("=== 相关历史知识 ===")
            for item in relevant:
                context_parts.append(f"- {item['content']}")
        
        # 3. 结构化事实（按相关性过滤）
        facts = self.structured_memory.get("learned_facts", [])[-10:]
        if facts:
            context_parts.append("=== 已学习的事实 ===")
            for f in facts[-5:]:
                context_parts.append(f"- {f['fact']}")
        
        return "\n".join(context_parts)
    
    def forget(self, older_than_days: int = 30):
        """记忆遗忘：清理过期的长期记忆"""
        # 向量存储的清理需要遍历删除，此处简化
        # 实际生产环境可使用带时间戳过滤的检索
        cutoff = datetime.now().timestamp() - older_than_days * 86400
        # 结构化记忆的清理
        self.structured_memory["learned_facts"] = [
            f for f in self.structured_memory["learned_facts"]
            if datetime.fromisoformat(f["timestamp"]).timestamp() > cutoff
        ]
        self._save_structured()

2.3 记忆的闭环：捕获→分析→更新

LangChain在2026年的记忆实践指南中提出了一个核心框架：一个运转良好的Agent记忆循环包含三个环节——捕获轨迹、分析轨迹、更新记忆。

from typing import List, Dict
import json

class MemoryLoop:
    """记忆闭环：自动从交互中学习"""
    
    def __init__(self, hybrid_memory: HybridMemory):
        self.memory = hybrid_memory
        self.traces = []
    
    def capture(self, user_input: str, agent_output: str, 
                success: bool, metadata: Dict = None):
        """捕获一次交互的完整轨迹"""
        trace = {
            "input": user_input,
            "output": agent_output,
            "success": success,
            "metadata": metadata or {},
            "timestamp": datetime.now().isoformat()
        }
        self.traces.append(trace)
        
        # 实时更新工作记忆
        self.memory.add_to_working(user_input, agent_output)
        
        # 如果是成功案例，提取关键信息进入长期记忆
        if success:
            self._extract_insight(trace)
        
        # 如果是失败案例，记录错误模式
        if not success:
            self._record_error_pattern(trace)
    
    def _extract_insight(self, trace: Dict):
        """从成功交互中提取可复用的洞察"""
        # 实际生产中可调用LLM进行摘要提取
        # 此处简化为关键词提取
        insight = f"成功处理: {trace['input'][:100]}"
        self.memory.add_fact(insight, category="success_pattern")
    
    def _record_error_pattern(self, trace: Dict):
        """记录失败模式"""
        self.memory.structured_memory["error_history"].append({
            "input": trace["input"],
            "output": trace["output"],
            "timestamp": trace["timestamp"]
        })
        self.memory._save_structured()
    
    def analyze_and_update(self):
        """批量分析轨迹并更新记忆（可定时执行）"""
        # 分析失败模式
        errors = self.memory.structured_memory.get("error_history", [])
        if len(errors) > 10:
            # 可调用LLM分析错误模式
            # 此处简化：生成一个聚合摘要
            error_summary = f"最近{len(errors)}次失败中有{len([e for e in errors if 'timeout' in str(e)])}次超时"
            self.memory.add_fact(error_summary, category="error_pattern")
            # 清理已处理的错误记录
            self.memory.structured_memory["error_history"] = []
            self.memory._save_structured()

2.4 记忆管控的安全与合规

生产级记忆系统必须考虑数据安全与合规：

class MemorySecurity:
    """记忆安全管控"""
    
    def __init__(self, hybrid_memory: HybridMemory):
        self.memory = hybrid_memory
        self.sensitive_patterns = [
            r"\b\d{3}-\d{2}-\d{4}\b",  # SSN
            r"\b[\w\.-]+@[\w\.-]+\.\w+\b",  # Email
            r"\b\d{4}-\d{4}-\d{4}-\d{4}\b",  # 信用卡
        ]
    
    def sanitize(self, text: str) -> str:
        """脱敏处理"""
        import re
        for pattern in self.sensitive_patterns:
            text = re.sub(pattern, "[REDACTED]", text)
        return text
    
    def add_with_privacy(self, content: str, metadata: Dict):
        """带隐私保护的记忆添加"""
        # 脱敏
        sanitized = self.sanitize(content)
        # 标记数据敏感等级
        metadata["privacy_level"] = self._classify_privacy(content)
        # 存储
        self.memory.add_to_long_term(sanitized, metadata)
    
    def _classify_privacy(self, content: str) -> str:
        """分类隐私等级"""
        import re
        if re.search(r"\b\d{3}-\d{2}-\d{4}\b", content):
            return "high"
        if re.search(r"\b[\w\.-]+@[\w\.-]+\.\w+\b", content):
            return "medium"
        return "low"
    
    def get_user_consent(self, user_id: str) -> bool:
        """检查用户是否同意数据被记忆"""
        # 实际生产需对接用户偏好数据库
        return True

三、可观测性：让Agent从“黑盒”变“白盒”

3.1 为什么传统监控对Agent失效

传统监控关注的是“系统是否在运行”（Liveness）。但Agent的问题是**“它在运行，但它在做正确的事吗”**（Quality）。

当一个Coding Agent在生产环境中陷入循环时，问题往往不会以清晰的错误码出现——它可能为了修复一个失败的测试反复编辑同一个文件。非确定性、多步骤的Agent系统会打破传统的监控模式，可观测性必须在设计时内置，而非故障后才追加。

生产级Agent面临五大“黑盒”痛点：

• 运行黑盒：缺乏实时运行状态的监控与告警
• 链路黑盒：复杂调用链条导致故障定位困难
• 成本黑盒：缺乏细粒度的成本分摊与ROI评估
• 安全黑盒：高权限Agent的数据隐私与操作合规性难以审计
• 质量黑盒：缺乏系统性的评估指标与持续迭代的反馈闭环

3.2 可观测性的三层架构

生产级Agent的可观测性需要覆盖日志（Logging）、指标（Metrics）、链路追踪（Tracing） 三个层次：

import time
import json
from typing import Dict, Any, Optional
from datetime import datetime
from dataclasses import dataclass, field
from enum import Enum
import uuid

class AgentStatus(Enum):
    STARTED = "started"
    THINKING = "thinking"
    TOOL_CALLING = "tool_calling"
    COMPLETED = "completed"
    ERROR = "error"
    TIMEOUT = "timeout"

@dataclass
class AgentSpan:
    """Agent执行的一个追踪跨度"""
    span_id: str
    parent_id: Optional[str]
    trace_id: str
    name: str
    start_time: float
    end_time: Optional[float] = None
    status: Optional[AgentStatus] = None
    metadata: Dict[str, Any] = field(default_factory=dict)
    events: List[Dict] = field(default_factory=list)
    error: Optional[str] = None
    
    def finish(self, status: AgentStatus, metadata: Dict = None):
        self.end_time = time.time()
        self.status = status
        if metadata:
            self.metadata.update(metadata)
    
    def add_event(self, event_type: str, data: Dict):
        self.events.append({
            "type": event_type,
            "data": data,
            "timestamp": time.time()
        })
    
    def to_dict(self) -> Dict:
        return {
            "span_id": self.span_id,
            "parent_id": self.parent_id,
            "trace_id": self.trace_id,
            "name": self.name,
            "duration_ms": (self.end_time - self.start_time) * 1000 if self.end_time else None,
            "status": self.status.value if self.status else None,
            "metadata": self.metadata,
            "events": self.events,
            "error": self.error
        }


class AgentObservability:
    """Agent可观测性系统"""
    
    def __init__(self, service_name: str = "agent"):
        self.service_name = service_name
        self.current_trace_id = None
        self.current_span_id = None
        self.spans: List[AgentSpan] = []
        self.metrics: Dict[str, Any] = {}
    
    def start_trace(self, trace_id: Optional[str] = None) -> str:
        """开始一个新的追踪"""
        self.current_trace_id = trace_id or str(uuid.uuid4())
        self.current_span_id = None
        self.spans = []
        return self.current_trace_id
    
    def start_span(self, name: str, metadata: Dict = None) -> AgentSpan:
        """开始一个新的跨度"""
        span = AgentSpan(
            span_id=str(uuid.uuid4())[:8],
            parent_id=self.current_span_id,
            trace_id=self.current_trace_id,
            name=name,
            start_time=time.time(),
            metadata=metadata or {}
        )
        self.current_span_id = span.span_id
        self.spans.append(span)
        return span
    
    def end_span(self, span: AgentSpan, status: AgentStatus, metadata: Dict = None):
        """结束当前跨度"""
        span.finish(status, metadata)
        # 记录关键指标
        duration = (span.end_time - span.start_time) * 1000
        self._record_metric(f"span.{span.name}.duration_ms", duration)
        self._record_metric(f"span.{span.name}.status", status.value)
    
    def _record_metric(self, name: str, value: Any):
        """记录指标"""
        if name not in self.metrics:
            self.metrics[name] = []
        self.metrics[name].append(value)
    
    def get_trace(self) -> Dict:
        """获取完整追踪"""
        return {
            "trace_id": self.current_trace_id,
            "service": self.service_name,
            "spans": [s.to_dict() for s in self.spans],
            "metrics": self.metrics,
            "total_spans": len(self.spans),
            "total_duration_ms": sum(
                (s.end_time - s.start_time) * 1000 
                for s in self.spans if s.end_time
            )
        }
    
    def log_llm_call(self, model: str, prompt: str, response: str, 
                     tokens: Dict, cost: float):
        """记录LLM调用的详细日志"""
        self._record_metric("llm.total_tokens", tokens.get("total", 0))
        self._record_metric("llm.prompt_tokens", tokens.get("prompt", 0))
        self._record_metric("llm.completion_tokens", tokens.get("completion", 0))
        self._record_metric("llm.cost", cost)
        
        # 详细日志可写入持久化存储
        log_entry = {
            "timestamp": datetime.now().isoformat(),
            "model": model,
            "prompt": prompt[:500],  # 截断以控制存储
            "response": response[:500],
            "tokens": tokens,
            "cost": cost,
            "trace_id": self.current_trace_id
        }
        # 实际生产可写入ES、ClickHouse等
        # self._write_to_log_store(log_entry)

3.3 集成OpenTelemetry的生产级追踪

以下是将Agent与OpenTelemetry集成的示例：

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.requests import RequestsInstrumentor
import functools

# 初始化OpenTelemetry
def setup_observability(endpoint: str = "http://localhost:4317"):
    """配置OpenTelemetry导出器"""
    provider = TracerProvider()
    processor = BatchSpanProcessor(
        OTLPSpanExporter(endpoint=endpoint, insecure=True)
    )
    provider.add_span_processor(processor)
    trace.set_tracer_provider(provider)
    
    # 自动instrument常见库
    RequestsInstrumentor().instrument()
    
    return trace.get_tracer(__name__)


class ObservableAgent:
    """带可观测性的Agent基类"""
    
    def __init__(self, tracer):
        self.tracer = tracer
    
    def observe(self, func):
        """装饰器：自动追踪Agent方法执行"""
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            with self.tracer.start_as_current_span(func.__name__) as span:
                # 记录输入
                span.set_attribute("agent.method", func.__name__)
                span.set_attribute("agent.input", str(args)[:200])
                
                try:
                    result = func(*args, **kwargs)
                    span.set_attribute("agent.success", True)
                    return result
                except Exception as e:
                    span.set_attribute("agent.success", False)
                    span.set_attribute("agent.error", str(e))
                    span.record_exception(e)
                    raise
        return wrapper
    
    @observe
    def execute(self, user_input: str) -> str:
        """执行Agent任务（由子类实现）"""
        raise NotImplementedError

3.4 评估与反馈闭环

可观测性的终极目标不是“看”，而是 “看→评估→改进”的闭环。

from typing import List, Dict, Callable
import json

class AgentEvaluator:
    """Agent评估系统"""
    
    def __init__(self):
        self.metrics = {
            "correctness": [],
            "completeness": [],
            "latency": [],
            "tool_usage": [],
            "error_rate": []
        }
        self.bad_cases = []  # 低分案例自动收集
    
    def evaluate(self, trace: Dict) -> Dict:
        """评估一次Agent执行"""
        scores = {}
        
        # 1. 正确性：输出是否合理
        scores["correctness"] = self._eval_correctness(trace)
        
        # 2. 完整性：是否完成所有任务步骤
        scores["completeness"] = self._eval_completeness(trace)
        
        # 3. 延迟：是否在预期时间内完成
        scores["latency"] = trace.get("total_duration_ms", 0)
        
        # 4. 工具调用效率
        scores["tool_efficiency"] = self._eval_tool_usage(trace)
        
        # 5. 是否出错
        scores["has_error"] = any(
            s.get("status") == "error" for s in trace.get("spans", [])
        )
        
        # 记录指标
        for key, value in scores.items():
            if key not in ["latency", "has_error"]:
                self.metrics[key].append(value)
        
        # 自动收集低分案例
        if scores.get("correctness", 1.0) < 0.6:
            self.bad_cases.append({
                "trace_id": trace.get("trace_id"),
                "scores": scores,
                "timestamp": datetime.now().isoformat()
            })
        
        return scores
    
    def _eval_correctness(self, trace: Dict) -> float:
        """评估正确性（可接入LLM-as-Judge）"""
        # 实际生产可使用LLM进行评分
        # 此处返回模拟分数
        return 0.85
    
    def _eval_completeness(self, trace: Dict) -> float:
        """评估完整性"""
        # 检查是否所有预期步骤都已执行
        return 0.90
    
    def _eval_tool_usage(self, trace: Dict) -> float:
        """评估工具调用效率"""
        # 检查工具调用次数是否合理
        return 0.80
    
    def get_bad_cases(self, limit: int = 100) -> List[Dict]:
        """获取低分案例，用于迭代优化"""
        return self.bad_cases[-limit:]
    
    def export_to_dataset(self) -> str:
        """导出为评测数据集"""
        dataset = {
            "bad_cases": self.bad_cases,
            "metrics_summary": {
                k: {
                    "avg": sum(v) / len(v) if v else 0,
                    "min": min(v) if v else 0,
                    "max": max(v) if v else 0,
                    "count": len(v)
                }
                for k, v in self.metrics.items() if v
            }
        }
        return json.dumps(dataset, indent=2)

四、整合：一个生产级Agent的完整骨架

将上述三个维度整合，形成一个可投入生产的Agent骨架：

from typing import Dict, Any, Optional
from datetime import datetime

class ProductionAgent:
    """生产级Agent完整骨架"""
    
    def __init__(self, 
                 llm,
                 memory: HybridMemory,
                 observability: AgentObservability,
                 evaluator: AgentEvaluator,
                 max_iterations: int = 5):
        self.llm = llm
        self.memory = memory
        self.observability = observability
        self.evaluator = evaluator
        self.max_iterations = max_iterations
        self.tools = {}  # 工具注册表
    
    def register_tool(self, name: str, func: Callable, description: str):
        """注册工具"""
        self.tools[name] = {"func": func, "description": description}
    
    def run(self, user_input: str, user_id: Optional[str] = None) -> Dict:
        """执行Agent任务"""
        # 1. 开始追踪
        trace_id = self.observability.start_trace()
        root_span = self.observability.start_span("agent.run", {
            "user_input": user_input[:200],
            "user_id": user_id
        })
        
        try:
            # 2. 加载记忆上下文
            context = self.memory.get_context(user_input)
            
            # 3. 构建Prompt（使用版本化Prompt）
            prompt = self._build_prompt(user_input, context)
            
            # 4. Agent循环（思考-行动-观察）
            for i in range(self.max_iterations):
                iter_span = self.observability.start_span(
                    f"agent.iteration.{i}",
                    {"iteration": i}
                )
                
                # 4a. LLM推理
                llm_span = self.observability.start_span("llm.call")
                response = self.llm.invoke(prompt)
                self.observability.log_llm_call(
                    model="gpt-4",
                    prompt=prompt,
                    response=response.content,
                    tokens={"total": 100, "prompt": 80, "completion": 20},
                    cost=0.002
                )
                self.observability.end_span(llm_span, AgentStatus.COMPLETED)
                
                # 4b. 解析并执行
                action = self._parse_action(response.content)
                if action["type"] == "final_answer":
                    self.observability.end_span(iter_span, AgentStatus.COMPLETED)
                    result = action["content"]
                    break
                
                if action["type"] == "tool_call":
                    tool_result = self._execute_tool(action["name"], action["args"])
                    # 将工具执行结果加入上下文
                    prompt += f"\n工具执行结果: {tool_result}"
                
                self.observability.end_span(iter_span, AgentStatus.COMPLETED)
            else:
                # 达到最大迭代次数
                result = "达到最大迭代次数，任务未完成"
                root_span.error = "max_iterations_exceeded"
            
            # 5. 记录到记忆
            self.memory.add_to_working(user_input, result)
            if user_id:
                self.memory.add_fact(
                    f"用户{user_id}询问了{user_input[:50]}",
                    category="user_interaction"
                )
            
            # 6. 评估
            trace = self.observability.get_trace()
            scores = self.evaluator.evaluate(trace)
            
            # 7. 结束追踪
            self.observability.end_span(
                root_span, 
                AgentStatus.COMPLETED,
                {"result": result[:200], "scores": scores}
            )
            
            return {
                "result": result,
                "trace_id": trace_id,
                "scores": scores,
                "iterations": i + 1
            }
            
        except Exception as e:
            self.observability.end_span(
                root_span,
                AgentStatus.ERROR,
                {"error": str(e)}
            )
            raise
    
    def _build_prompt(self, user_input: str, context: str) -> str:
        """构建Prompt（使用版本化的模板）"""
        return f"""你是一个任务执行Agent。

{context}

用户输入: {user_input}

请分析任务并给出行动。你可以：
1. 直接回答用户（action: final_answer）
2. 调用工具获取信息（action: tool_call）

输出格式必须是JSON。
"""
    
    def _parse_action(self, response: str) -> Dict:
        """解析Agent输出"""
        # 实际生产使用结构化解析
        return {"type": "final_answer", "content": response}
    
    def _execute_tool(self, name: str, args: Dict) -> str:
        """执行工具"""
        if name not in self.tools:
            return f"错误: 工具{name}不存在"
        try:
            return str(self.tools[name]["func"](**args))
        except Exception as e:
            return f"工具执行失败: {str(e)}"

五、总结

生产级Agent开发的核心，可以归结为三个关键词：

提示设计决定Agent“能不能听懂话”——但生产环境中，更关键的是“听不懂时怎么办”。结构化输出、Fallback机制、版本化管理，是让Prompt从“艺术”走向“工程”的三大支柱。

记忆管控决定Agent“能不能记住事”——但生产环境中，更关键的是“记住什么、忘掉什么、怎么修正错误记忆”。五阶段流水线、混合记忆架构、安全与合规管控，是让记忆从“存储”走向“策展”的核心能力。

可观测性决定Agent“能不能被信任”——但生产环境中，更关键的是“能不能基于观测数据持续改进”。三层架构（日志/指标/追踪）、评估体系、反馈闭环，是让Agent从“黑盒”走向“白盒”的必经之路。

正如行业观察者所言：“2026年Agent的核心瓶颈不在模型，而在系统工程层——缺少一个能够承载’思考—执行—反馈—记忆’闭环的运行框架”。提示设计、记忆管控与可观测性，正是这个框架的三大支柱。三者兼备，Agent才能从“能跑”走向“可靠”，从Demo走向生产。