Agent 智能体

import operator
from typing import Annotated, TypedDict

class PlanState(TypedDict):
    objective: str
    steps: list[str]                        # 剩余步骤（planner 产出）
    results: Annotated[list[str], operator.add]   # 每步执行结果
    final: str

# planner（结构化输出 list[str]）：
#   "调研 X 并写报告" -> ["检索资料", "提炼要点", "撰写初稿", "自检修订"]
# executor：
#   取 steps[0]，用工具或 ReAct 子图完成，结果追加进 results
# replan（看完剩余 steps 与全部 results 再决策）：
#   继续下一步 / 重排剩余步骤 / 判定目标已达成，直接产出 final

class ReflexionState(TypedDict):
    task: str
    attempts: int                 # 已尝试次数，硬上限 3
    reflections: Annotated[list[str], operator.add]   # 每次失败的教训
    output: str

def reflect(state: ReflexionState) -> dict:
    lesson = llm.invoke(
        "任务：{task}\n上次输出：{output}\n失败原因：{feedback}\n"
        "用一两句话总结下次尝试应改进的具体做法。"
    ).content
    return {"reflections": [lesson]}   # 下一轮 system prompt 注入全部 lessons

# 差：模型不知道何时用、单位是什么、结果长什么样
@tool
def search_orders(q: str) -> str:
    """搜索订单。"""
    return order_db.search(q)

# 好：适用边界、参数语义、返回结构、失败行为全部写清
@tool
def search_orders(
    query: str,
    status: Literal["pending", "shipped", "all"] = "all",
    limit: int = 10,
) -> str:
    """按关键词搜索当前用户的历史订单。

    适用：用户询问自己的订单状态、金额、物流单号。
    不适用：查商品库存（用 check_stock）；查他人订单（无权限）。
    query 可用订单号、商品名或收货人姓名；limit 最大 50。
    返回 JSON 数组，字段：order_id, status, total, tracking_no。
    无匹配时返回空数组，不报错。
    """
    rows = order_db.search(query, status=status, limit=limit)
    return json.dumps(rows, ensure_ascii=False)

@tool
def cancel_order(order_id: str) -> str:
    """取消订单。仅限 24 小时内、未发货的订单。"""
    result = order_db.try_cancel(order_id)
    if not result.ok:
        # 失败原因 + 补救建议，模型可以据此向用户解释并改道
        return f"取消失败：{result.reason}。建议：{result.suggestion}"
    return "已取消，退款将在 3 个工作日原路退回。"

SYSTEM = f"""你是 {PRODUCT} 的订单助理。

[角色与边界]
只处理订单与物流问题，其余礼貌引导到人工客服。

[环境状态]
当前时间：{now}（Asia/Shanghai）
当前用户：{user.name}，VIP 等级 {user.tier}
已知偏好：{user.prefs_summary}

[稳定规则]
1. 金额一律人民币，保留两位小数
2. 不确定订单号时先用 search_orders 确认，不要猜
3. 退款类操作必须先征得用户明确确认
"""

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()

# 写：命名空间（用户维度隔离）+ 键 + 值
store.put(("user", "u-42"), "profile",
          {"value": {"name": "林可", "timezone": "Asia/Shanghai"}})
store.put(("user", "u-42"), "prefs",
          {"value": {"style": "简洁，优先给代码示例"}})

# 读：每轮预取，拼进系统提示的环境层
profile = store.get(("user", "u-42"), "profile").value
prefs = store.get(("user", "u-42"), "prefs").value

@tool
def save_memory(content: str,
                kind: Literal["fact", "preference", "event"]) -> str:
    """把值得长期记住的信息写入用户记忆。

    仅当信息跨会话仍然有效且用户明确提及时使用；
    临时性内容、可从工具查到的数据不要存。
    """
    store.put(("user", current_user_id), f"mem-{uuid4().hex[:8]}",
              {"value": {"content": content, "kind": kind, "ts": now()}})
    return "已记住。"

from pydantic import BaseModel, Field

class Critique(BaseModel):
    score: float = Field(description="0 到 10，达标线 7")
    issues: list[str] = Field(description="具体问题清单；无则空列表")

critic = llm.with_structured_output(Critique)

def review_node(state: MessagesState) -> dict:
    c = critic.invoke(
        "对照任务要求审查产出：\n" + state["messages"][-1].content
    )
    if c.score >= 7 or state.get("revisions", 0) >= 2:
        return {"messages": [approve_and_finish(state)]}
    # 带着具体问题清单重写，而不是“请再好一点”
    return {"messages": [revise_with(state, c.issues)],
            "revisions": state.get("revisions", 0) + 1}

from langgraph.types import interrupt

@tool
def transfer_money(payee: str, amount: float) -> str:
    """向指定收款人转账。高危操作，需要用户确认。"""
    decision = interrupt({
        "question": "确认转账？",
        "payee": payee,
        "amount": amount,
    })
    if decision != "approve":
        return "用户已取消转账。"
    return bank.transfer(payee, amount)   # 副作用在确认之后

import subprocess, sys

@tool
def run_python(code: str) -> str:
    """在隔离沙箱中执行 Python 代码，返回 stdout。

    无网络访问；最长 10 秒；仅 /data 目录可读写。
    需要“循环处理多个文件”“多步计算”时用它，
    单次查询请直接用对应工具。
    """
    proc = subprocess.run(
        [sys.executable, "-I", "-c", code],    # -I：隔离模式，不加载用户环境
        capture_output=True, text=True, timeout=10,
    )
    out = proc.stdout[-4000:] or proc.stderr[-2000:]
    return out or "(no output)"

class TrajectoryVerdict(BaseModel):
    correct_tools: bool = Field(description="工具选择是否正确")
    efficient: bool = Field(description="是否存在多余或重复步骤")
    final_correct: bool = Field(description="最终结果是否完成任务")
    notes: str = Field(description="问题说明，无则空字符串")

judge = llm.with_structured_output(TrajectoryVerdict)

def eval_trajectory(task: str, trace: list[dict]) -> TrajectoryVerdict:
    steps = "\n".join(
        f"{i}. {t['tool']}({t['args']}) -> {t['result_summary']}"
        for i, t in enumerate(trace, 1)
    )
    return judge.invoke(
        f"任务：{task}\n执行轨迹：\n{steps}\n"
        "评估工具选择、路径效率与最终结果。"
    )