生产级 AI Agent 从零搭建：从 Prompt、Memory、Tool、RAG 到 Runtime 的全链路工程实践

┌──────────────────────────────────────────────────────────────┐
│                        接入与治理层                           │
│ API Gateway / Auth / Rate Limit / Tenant Isolation / Audit  │
└──────────────────────────────┬───────────────────────────────┘
                               │
┌──────────────────────────────▼───────────────────────────────┐
│                         Agent Runtime                         │
│ Session / Planner / Tool Executor / State Machine / Stream  │
└───────┬──────────────────────┬──────────────────────┬────────┘
        │                      │                      │
┌───────▼────────┐   ┌─────────▼────────┐   ┌────────▼─────────┐
│ Prompt & Policy │   │ Memory & Context │   │ Knowledge & RAG  │
│ Role/Rules/Budget│  │ Window/Summary/Profile│ Recall/Rerank/Cite│
└───────┬────────┘   └─────────┬────────┘   └────────┬─────────┘
        │                      │                      │
┌───────▼──────────────────────▼──────────────────────▼────────┐
│                       Tool & Workflow Plane                   │
│ Registry / Permission / Idempotency / Compensation / Saga   │
└───────┬──────────────────────────────────────────────────────┘
        │
┌───────▼──────────────────────────────────────────────────────┐
│                        Infrastructure                         │
│ Redis / PostgreSQL / Kafka / VectorDB / Object Store / OTel │
└──────────────────────────────────────────────────────────────┘

User Request
  -> 鉴权与租户识别
  -> Session 装载
  -> Memory 加载
  -> Query 分类与 Skill 路由
  -> RAG 检索与上下文构建
  -> Prompt 编译
  -> LLM 推理
  -> Tool Call 决策
  -> 工具执行 / 工作流执行
  -> 结果验证与后处理
  -> 流式返回 / 最终回答
  -> 事件落库与异步记忆写回

from dataclasses import dataclass
from enum import Enum
from typing import List
 
class PromptLayerType(str, Enum):
    ROLE = "role"
    POLICY = "policy"
    TOOL = "tool"
    CONTEXT = "context"
    MEMORY = "memory"
    TASK = "task"
 
@dataclass
class PromptLayer:
    layer_type: PromptLayerType
    content: str
    priority: int
    token_budget: int
    required: bool = True
 
class PromptCompiler:
    def __init__(self, token_counter):
        self.token_counter = token_counter
 
    def compile(self, layers: List[PromptLayer], total_budget: int) -> str:
        ordered = sorted(layers, key=lambda x: x.priority)
        compiled: list[str] = []
        used_tokens = 0
 
        for layer in ordered:
            if not layer.content:
                continue
 
            current_tokens = self.token_counter.count(layer.content)
            if current_tokens > layer.token_budget:
                layer_content = self.token_counter.truncate(
                    layer.content, layer.token_budget
                )
                current_tokens = self.token_counter.count(layer_content)
            else:
                layer_content = layer.content
 
            if used_tokens + current_tokens > total_budget:
                if layer.required:
                    raise ValueError(f"Prompt budget exceeded at {layer.layer_type}")
                continue
 
            compiled.append(layer_content)
            used_tokens += current_tokens
 
        return "\n\n".join(compiled)

Layer 1: Role
你是谁，负责什么，不负责什么
 
Layer 2: Policy
合规规则、禁止行为、答复约束、升级策略
 
Layer 3: Tool Contract
可以使用哪些工具，工具的输入输出约束是什么
 
Layer 4: Memory Context
用户近期上下文、用户画像、当前任务状态
 
Layer 5: Knowledge Context
RAG 检索结果、来源与置信度
 
Layer 6: Task Instruction
本轮用户请求与输出格式要求

请谨慎调用工具，必要时才调用。

仅当回答需要外部实时信息、用户授权后的业务操作、或必须引用内部知识时才允许调用工具。
如果用户请求只是解释概念、总结信息或润色文本，不要调用任何工具。
同一工具在一轮会话中最多调用 2 次；若两次结果仍不足以回答，应明确说明信息不足并请求人工协助。

L1: Working Memory
- 当前任务内的短期状态
- 生命周期：本次会话 / 当前任务
- 存储：Redis / In-Memory
 
L2: Conversation Memory
- 近期对话窗口
- 生命周期：分钟到小时
- 存储：Redis List / Stream
 
L3: Summary Memory
- 历史摘要、关键决策、阶段性结论
- 生命周期：天级
- 存储：Redis + PostgreSQL
 
L4: Profile / Fact Memory
- 用户画像、长期偏好、稳定业务事实
- 生命周期：周到月
- 存储：PostgreSQL / KV / Graph

from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Optional
import time
 
class MemoryType(str, Enum):
    MESSAGE = "message"
    SUMMARY = "summary"
    FACT = "fact"
    TASK_STATE = "task_state"
 
@dataclass
class MemoryRecord:
    tenant_id: str
    user_id: str
    session_id: str
    memory_type: MemoryType
    content: str
    metadata: dict[str, Any] = field(default_factory=dict)
    created_at: float = field(default_factory=lambda: time.time())
    expire_at: Optional[float] = None
 
class MemoryManager:
    def __init__(self, redis_client, repository, summary_queue):
        self.redis = redis_client
        self.repository = repository
        self.summary_queue = summary_queue
 
    async def append_message(self, record: MemoryRecord, window_size: int = 20) -> None:
        key = f"agent:mem:{record.tenant_id}:{record.session_id}:window"
        payload = self._serialize(record)
 
        pipe = self.redis.pipeline()
        pipe.rpush(key, payload)
        pipe.ltrim(key, -window_size, -1)
        pipe.expire(key, 3600)
        await pipe.execute()
 
    async def load_context(
        self,
        tenant_id: str,
        session_id: str,
        max_messages: int = 10,
    ) -> list[dict[str, Any]]:
        key = f"agent:mem:{tenant_id}:{session_id}:window"
        values = await self.redis.lrange(key, -max_messages, -1)
        return [self._deserialize(item) for item in values]
 
    async def trigger_summary_if_needed(self, tenant_id: str, session_id: str) -> None:
        key = f"agent:mem:{tenant_id}:{session_id}:window"
        size = await self.redis.llen(key)
        if size >= 18:
            await self.summary_queue.publish(
                {
                    "tenant_id": tenant_id,
                    "session_id": session_id,
                    "event": "memory.summary.generate",
                }
            )
 
    def _serialize(self, record: MemoryRecord) -> str:
        import json
        return json.dumps(record.__dict__, ensure_ascii=False)
 
    def _deserialize(self, payload: str) -> dict[str, Any]:
        import json
        return json.loads(payload)

1. 当前用户目标
2. 已确认事实
3. 已执行动作
4. 待执行动作
5. 禁止或限制条件
6. 需要人工接管的风险点

import asyncio
import json
import time
from dataclasses import dataclass, field
from typing import Any, Callable, Optional
 
from pydantic import BaseModel, ValidationError
 
class ToolRequest(BaseModel):
    tenant_id: str
    user_id: str
    session_id: str
    trace_id: str
    idempotency_key: str
    arguments: dict[str, Any]
 
@dataclass
class ToolDefinition:
    name: str
    description: str
    schema_model: type[BaseModel]
    handler: Callable[..., Any]
    timeout_seconds: float = 5.0
    max_retries: int = 0
    need_approval: bool = False
    side_effect: bool = False
    allowed_roles: list[str] = field(default_factory=list)
 
    def to_openai_tool(self) -> dict[str, Any]:
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": self.schema_model.model_json_schema(),
            },
        }
 
class ToolRegistry:
    def __init__(self):
        self._tools: dict[str, ToolDefinition] = {}
 
    def register(self, tool: ToolDefinition) -> None:
        if tool.name in self._tools:
            raise ValueError(f"tool already exists: {tool.name}")
        self._tools[tool.name] = tool
 
    def get(self, name: str) -> Optional[ToolDefinition]:
        return self._tools.get(name)
 
    def list_openai_tools(self) -> list[dict[str, Any]]:
        return [tool.to_openai_tool() for tool in self._tools.values()]
 
class ToolExecutor:
    def __init__(self, permission_service, idempotency_store, audit_logger):
        self.permission_service = permission_service
        self.idempotency_store = idempotency_store
        self.audit_logger = audit_logger
 
    async def execute(self, tool: ToolDefinition, request: ToolRequest) -> dict[str, Any]:
        start = time.time()
 
        try:
            validated_args = tool.schema_model(**request.arguments)
        except ValidationError as exc:
            return {"status": "invalid_argument", "error": exc.errors()}
 
        allowed = await self.permission_service.check(
            tenant_id=request.tenant_id,
            user_id=request.user_id,
            tool_name=tool.name,
            roles=tool.allowed_roles,
        )
        if not allowed:
            return {"status": "forbidden", "error": "permission denied"}
 
        if tool.side_effect:
            existed = await self.idempotency_store.try_acquire(
                request.idempotency_key, ttl_seconds=600
            )
            if not existed:
                return {"status": "duplicate", "error": "duplicated request"}
 
        try:
            result = await asyncio.wait_for(
                tool.handler(validated_args, request),
                timeout=tool.timeout_seconds,
            )
            return {"status": "ok", "result": result}
        except asyncio.TimeoutError:
            return {"status": "timeout", "error": f"{tool.name} timeout"}
        except Exception as exc:
            return {"status": "error", "error": str(exc)}
        finally:
            await self.audit_logger.log(
                trace_id=request.trace_id,
                tool_name=tool.name,
                cost_ms=int((time.time() - start) * 1000),
                side_effect=tool.side_effect,
            )

class AgentLoopGuard:
    def __init__(
        self,
        max_iterations: int = 8,
        max_tool_calls: int = 6,
        max_elapsed_ms: int = 15000,
    ):
        self.max_iterations = max_iterations
        self.max_tool_calls = max_tool_calls
        self.max_elapsed_ms = max_elapsed_ms
 
class SafeAgentExecutor:
    def __init__(self, llm_client, tool_registry: ToolRegistry, tool_executor: ToolExecutor):
        self.llm_client = llm_client
        self.tool_registry = tool_registry
        self.tool_executor = tool_executor
 
    async def run(self, state, messages, guard: AgentLoopGuard):
        started_at = time.time()
        tool_call_count = 0
 
        for iteration in range(guard.max_iterations):
            elapsed_ms = int((time.time() - started_at) * 1000)
            if elapsed_ms > guard.max_elapsed_ms:
                return {"status": "timeout", "answer": "请求处理超时，请稍后重试。"}
 
            response = await self.llm_client.chat(messages, state.available_tools)
            assistant_message = response["message"]
 
            if not assistant_message.get("tool_calls"):
                return {"status": "ok", "answer": assistant_message.get("content", "")}
 
            for tool_call in assistant_message["tool_calls"]:
                tool_call_count += 1
                if tool_call_count > guard.max_tool_calls:
                    return {"status": "overflow", "answer": "工具调用次数超限，已停止执行。"}
 
                tool = self.tool_registry.get(tool_call["name"])
                if not tool:
                    messages.append(
                        {
                            "role": "tool",
                            "content": json.dumps({"status": "unknown_tool"}, ensure_ascii=False),
                        }
                    )
                    continue
 
                result = await self.tool_executor.execute(
                    tool=tool,
                    request=state.build_tool_request(tool_call["arguments"]),
                )
                messages.append(
                    {
                        "role": "tool",
                        "name": tool.name,
                        "content": json.dumps(result, ensure_ascii=False),
                    }
                )
 
        return {"status": "exhausted", "answer": "达到最大推理次数，建议转人工处理。"}

用户问题
  -> Query Rewrite
  -> 召回策略选择
  -> Dense Recall
  -> Sparse Recall
  -> RRF 融合
  -> Rerank
  -> 权限过滤
  -> 上下文压缩
  -> 回答生成
  -> 引用来源输出

from dataclasses import dataclass
from typing import Any
 
@dataclass
class RetrievedDocument:
    doc_id: str
    content: str
    score: float
    source: str
    tenant_id: str
    metadata: dict[str, Any]
 
class RAGPipeline:
    def __init__(self, dense_store, sparse_store, reranker, permission_filter):
        self.dense_store = dense_store
        self.sparse_store = sparse_store
        self.reranker = reranker
        self.permission_filter = permission_filter
 
    async def query(
        self,
        tenant_id: str,
        user_id: str,
        query: str,
        top_k: int = 8,
    ) -> tuple[str, list[RetrievedDocument]]:
        rewritten = await self._rewrite_query(query)
        dense = await self.dense_store.search(tenant_id=tenant_id, query=rewritten, top_k=20)
        sparse = await self.sparse_store.search(tenant_id=tenant_id, query=rewritten, top_k=20)
 
        fused = self._rrf_merge(dense, sparse)
        permitted = await self.permission_filter.filter(
            tenant_id=tenant_id,
            user_id=user_id,
            documents=fused,
        )
 
        ranked = await self.reranker.rank(query=rewritten, documents=permitted[:12])
        selected = ranked[:top_k]
        context = self._build_context(selected)
        return context, selected
 
    async def _rewrite_query(self, query: str) -> str:
        return query
 
    def _rrf_merge(self, dense: list[RetrievedDocument], sparse: list[RetrievedDocument]):
        scores: dict[str, float] = {}
        doc_map: dict[str, RetrievedDocument] = {}
 
        for rank, doc in enumerate(dense):
            scores[doc.doc_id] = scores.get(doc.doc_id, 0.0) + 1 / (60 + rank + 1)
            doc_map[doc.doc_id] = doc
 
        for rank, doc in enumerate(sparse):
            scores[doc.doc_id] = scores.get(doc.doc_id, 0.0) + 1 / (60 + rank + 1)
            doc_map[doc.doc_id] = doc
 
        sorted_ids = sorted(scores.keys(), key=lambda x: scores[x], reverse=True)
        return [doc_map[doc_id] for doc_id in sorted_ids]
 
    def _build_context(self, documents: list[RetrievedDocument]) -> str:
        blocks = []
        for index, doc in enumerate(documents, start=1):
            blocks.append(
                f"[来源{index}] source={doc.source} score={doc.score:.4f}\n{doc.content}"
            )
        return "\n\n---\n\n".join(blocks)

from abc import ABC, abstractmethod
 
class BaseSkill(ABC):
    @property
    @abstractmethod
    def name(self) -> str:
        ...
 
    @abstractmethod
    async def match(self, request: dict) -> float:
        ...
 
    @abstractmethod
    async def execute(self, request: dict, runtime_context: dict) -> dict:
        ...
 
class SkillRouter:
    def __init__(self, skills: list[BaseSkill], intent_classifier):
        self.skills = skills
        self.intent_classifier = intent_classifier
 
    async def route(self, request: dict) -> BaseSkill:
        scored = []
        for skill in self.skills:
            score = await skill.match(request)
            if score >= 0.85:
                return skill
            if score >= 0.30:
                scored.append((score, skill))
 
        if not scored:
            raise RuntimeError("no skill matched")
 
        top_candidates = [skill for _, skill in sorted(scored, key=lambda x: x[0], reverse=True)[:3]]
        selected_name = await self.intent_classifier.select(
            text=request["message"],
            candidates=[skill.name for skill in top_candidates],
        )
        for skill in top_candidates:
            if skill.name == selected_name:
                return skill
        return top_candidates[0]

from enum import Enum
 
class RuntimeState(str, Enum):
    RECEIVED = "received"
    SESSION_READY = "session_ready"
    CONTEXT_READY = "context_ready"
    MODEL_THINKING = "model_thinking"
    TOOL_RUNNING = "tool_running"
    POST_PROCESSING = "post_processing"
    COMPLETED = "completed"
    FAILED = "failed"
    HUMAN_REVIEW = "human_review"

 
# app/config.py
 
from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict
 
class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")
 
    app_name: str = "agent-platform"
    env: str = "dev"
    host: str = "0.0.0.0"
    port: int = 8080
 
    openai_api_key: str = ""
    openai_base_url: str = "https://api.openai.com/v1"
    llm_primary_model: str = "gpt-4.1"
    llm_fast_model: str = "gpt-4.1-mini"
 
    redis_url: str = "redis://localhost:6379/0"
    postgres_dsn: str = ""
    vector_backend: str = "pgvector"
 
    agent_max_iterations: int = 8
    agent_max_tool_calls: int = 6
    agent_request_timeout_ms: int = 15000
    agent_context_budget: int = 24000
 
    rag_top_k: int = 6
    rag_candidate_k: int = 20
 
    enable_trace_log: bool = True
    enable_sensitive_masking: bool = True
 
settings = Settings()

 
# app/runtime/context.py
 
from dataclasses import dataclass, field
from typing import Any
import uuid
 
@dataclass
class RequestContext:
    tenant_id: str
    user_id: str
    session_id: str
    message: str
    trace_id: str = field(default_factory=lambda: uuid.uuid4().hex)
    metadata: dict[str, Any] = field(default_factory=dict)
 
@dataclass
class RuntimeContext:
    request: RequestContext
    available_tools: list[dict[str, Any]]
    prompt_budget: int
    state: dict[str, Any] = field(default_factory=dict)
    citations: list[dict[str, Any]] = field(default_factory=list)

 
# app/llm/router.py
 
class LLMRouter:
    def __init__(self, primary_client, fast_client):
        self.primary_client = primary_client
        self.fast_client = fast_client
 
    async def select_model(self, scene: str, complexity: str) -> str:
        if scene in {"routing", "query_rewrite", "summary"}:
            return "fast"
        if complexity == "low":
            return "fast"
        return "primary"
 
    async def chat(self, scene: str, complexity: str, **kwargs):
        model_key = await self.select_model(scene=scene, complexity=complexity)
        client = self.fast_client if model_key == "fast" else self.primary_client
        return await client.chat.completions.create(**kwargs)

 
# app/runtime/agent_runtime.py
 
import json
import time
 
class AgentRuntime:
    def __init__(
        self,
        prompt_service,
        memory_service,
        rag_service,
        skill_router,
        tool_registry,
        tool_executor,
        llm_router,
        event_store,
    ):
        self.prompt_service = prompt_service
        self.memory_service = memory_service
        self.rag_service = rag_service
        self.skill_router = skill_router
        self.tool_registry = tool_registry
        self.tool_executor = tool_executor
        self.llm_router = llm_router
        self.event_store = event_store
 
    async def handle(self, request_context):
        started_at = time.time()
 
        recent_history = await self.memory_service.load_recent(
            tenant_id=request_context.tenant_id,
            session_id=request_context.session_id,
        )
        profile = await self.memory_service.load_profile(
            tenant_id=request_context.tenant_id,
            user_id=request_context.user_id,
        )
 
        skill = await self.skill_router.route(
            {
                "tenant_id": request_context.tenant_id,
                "user_id": request_context.user_id,
                "message": request_context.message,
            }
        )
 
        rag_context, sources = await self.rag_service.query(
            tenant_id=request_context.tenant_id,
            user_id=request_context.user_id,
            query=request_context.message,
        )
 
        prompt = await self.prompt_service.build(
            skill=skill.name,
            message=request_context.message,
            history=recent_history,
            profile=profile,
            rag_context=rag_context,
        )
 
        messages = [
            {"role": "system", "content": prompt},
            {"role": "user", "content": request_context.message},
        ]
 
        tool_schemas = self.tool_registry.list_openai_tools()
        llm_response = await self.llm_router.chat(
            scene="agent_main",
            complexity="medium",
            model="gpt-4.1",
            messages=messages,
            tools=tool_schemas,
            tool_choice="auto",
            temperature=0.2,
        )
 
        final_answer = await self._continue_if_needed(
            request_context=request_context,
            messages=messages,
            response=llm_response,
        )
 
        await self.memory_service.write_back(
            tenant_id=request_context.tenant_id,
            user_id=request_context.user_id,
            session_id=request_context.session_id,
            user_message=request_context.message,
            assistant_message=final_answer,
        )
 
        await self.event_store.append(
            trace_id=request_context.trace_id,
            event_type="agent.completed",
            payload={
                "skill": skill.name,
                "elapsed_ms": int((time.time() - started_at) * 1000),
                "sources": sources,
            },
        )
 
        return {"answer": final_answer, "sources": sources}
 
    async def _continue_if_needed(self, request_context, messages, response):
        choice = response.choices[0].message
        if not choice.tool_calls:
            return choice.content or ""
 
        messages.append(
            {
                "role": "assistant",
                "content": choice.content or "",
                "tool_calls": [
                    {
                        "id": tc.id,
                        "type": "function",
                        "function": {
                            "name": tc.function.name,
                            "arguments": tc.function.arguments,
                        },
                    }
                    for tc in choice.tool_calls
                ],
            }
        )
 
        for tc in choice.tool_calls:
            tool = self.tool_registry.get(tc.function.name)
            if not tool:
                tool_result = {"status": "unknown_tool"}
            else:
                tool_result = await self.tool_executor.execute(
                    tool=tool,
                    request=request_context.to_tool_request(
                        arguments=json.loads(tc.function.arguments or "{}")
                    ),
                )
 
            messages.append(
                {
                    "role": "tool",
                    "tool_call_id": tc.id,
                    "content": json.dumps(tool_result, ensure_ascii=False),
                }
            )
 
        second_round = await self.llm_router.chat(
            scene="agent_main",
            complexity="medium",
            model="gpt-4.1",
            messages=messages,
            temperature=0.2,
        )
        return second_round.choices[0].message.content or ""

 
# app/tools/reschedule_delivery.py
 
from pydantic import BaseModel, Field
 
class RescheduleDeliveryArgs(BaseModel):
    order_id: str = Field(min_length=8, max_length=64)
    expected_date: str
    time_slot: str
    reason: str = Field(default="user_request", max_length=128)
 
async def reschedule_delivery_handler(args: RescheduleDeliveryArgs, request):
    order = await order_gateway.get_order(
        tenant_id=request.tenant_id,
        order_id=args.order_id,
    )
    if not order:
        return {"status": "not_found", "message": "订单不存在"}
 
    if order["status"] not in {"PAID", "READY_TO_SHIP", "DELIVERING"}:
        return {"status": "rejected", "message": "当前订单状态不允许改约"}
 
    if not await risk_service.allow_reschedule(
        tenant_id=request.tenant_id,
        user_id=request.user_id,
        order_id=args.order_id,
    ):
        return {"status": "review_required", "message": "需要人工审核"}
 
    updated = await delivery_gateway.reschedule(
        tenant_id=request.tenant_id,
        order_id=args.order_id,
        expected_date=args.expected_date,
        time_slot=args.time_slot,
        operator=request.user_id,
        idempotency_key=request.idempotency_key,
    )
 
    await audit_service.record(
        trace_id=request.trace_id,
        action="delivery.reschedule",
        order_id=args.order_id,
        payload=args.model_dump(),
    )
    return {"status": "ok", "result": updated}

 
# app/main.py
 
from fastapi import FastAPI
from pydantic import BaseModel
import uuid
 
app = FastAPI(title="Production Agent Platform")
 
class ChatRequest(BaseModel):
    tenant_id: str
    user_id: str
    session_id: str | None = None
    message: str
 
class ChatResponse(BaseModel):
    trace_id: str
    session_id: str
    answer: str
    sources: list[dict]
 
@app.post("/api/v1/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    session_id = request.session_id or uuid.uuid4().hex
    request_context = RequestContext(
        tenant_id=request.tenant_id,
        user_id=request.user_id,
        session_id=session_id,
        message=request.message,
    )
    result = await agent_runtime.handle(request_context)
    return ChatResponse(
        trace_id=request_context.trace_id,
        session_id=session_id,
        answer=result["answer"],
        sources=result["sources"],
    )

订单昨天还没发货，帮我查一下原因，如果今天也发不了就改到明天下午配送。

用户请求
  -> Skill Router 命中 order_service
  -> 调 query_order 工具
  -> 调 query_shipment_reason 工具
  -> 模型判断是否需要进一步动作
  -> 若满足规则，则调 reschedule_delivery 工具
  -> 输出结果并附带操作结果说明
  -> 异步写审计与会话摘要

agent.request.received
agent.session.loaded
agent.skill.selected
agent.rag.recalled
agent.prompt.compiled
agent.llm.completed
agent.tool.called
agent.tool.completed
agent.response.returned
agent.memory.persisted