Agent 架构反思：为什么文件不是万能的上下文容器

pip install pymongo openai

import pymongo
import openai
import json
from datetime import datetime

# ── 1. 连接 MongoDB，创建带 schema 校验的集合 ──

client = pymongo.MongoClient("mongodb://localhost:27017/")
db = client["agent_workflow"]

# 用 MongoDB 的 JSON Schema 校验确保状态结构不漂移
db.create_collection("task_state", validator={
    "$jsonSchema": {
        "bsonType": "object",
        "required": ["task_id", "agent", "status", "output", "created_at"],
        "properties": {
            "task_id":    {"bsonType": "string"},
            "agent":      {"bsonType": "string"},
            "status":     {"enum": ["pending", "done", "failed"]},
            "output":     {"bsonType": "string"},
            "created_at": {"bsonType": "date"},
            "metadata":   {"bsonType": "object"}  # 可选的附加字段
        }
    }
})

col = db["task_state"]

# ── 2. 研究 Agent：产出调研摘要并写入数据库 ──

openai.api_key = "sk-..."  # 替换为你的 key

def research_agent(topic: str) -> str:
    """调用 LLM 做调研，返回摘要文本"""
    resp = openai.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "你是一名技术调研员，输出简洁的要点摘要。"},
            {"role": "user", "content": f"调研主题：{topic}。列出 3-5 个核心要点。"}
        ],
        temperature=0.3
    )
    return resp.choices[0].message.content

task_id = "ctx-eng-001"
summary = research_agent("上下文工程在 Agent 系统中的实践")

# 写入状态——schema 校验会拦截缺失字段或类型错误
col.insert_one({
    "task_id":    task_id,
    "agent":      "research",
    "status":     "done",
    "output":     summary,
    "created_at": datetime.utcnow(),
    "metadata":   {"topic": "上下文工程", "tokens_used": "~200"}
})

print("研究 Agent 输出已写入 MongoDB")
print(summary[:120], "...")

# ── 3. 写作 Agent：从数据库精确读取上游状态 ──

def writing_agent(task_id: str) -> str:
    """只拉取指定 task 的 output，不读全表"""
    state = col.find_one(
        {"task_id": task_id, "agent": "research", "status": "done"},
        {"output": 1}  # 只取 output 字段，不取 metadata 等无关内容
    )
    if not state:
        raise ValueError(f"未找到 task {task_id} 的研究产出")

    research_output = state["output"]

    # 只把相关摘要放进 prompt，而不是整段对话历史
    resp = openai.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "你是一名技术作者，基于调研摘要写一段通俗解读。"},
            {"role": "user", "content": f"调研摘要如下：\n{research_output}\n\n请写一段 200 字以内的通俗解读。"}
        ],
        temperature=0.5
    )
    return resp.choices[0].message.content

draft = writing_agent(task_id)

# 写入写作 Agent 的状态
col.insert_one({
    "task_id":    task_id,
    "agent":      "writing",
    "status":     "done",
    "output":     draft,
    "created_at": datetime.utcnow(),
    "metadata":   {"model": "gpt-4o-mini"}
})

print("\n写作 Agent 草稿已写入 MongoDB")
print(draft)

# MongoDB Atlas Vector Search 示例（需 Atlas 集群）
# 先为集合添加向量字段
from openai import OpenAI
oai = OpenAI()

def embed_text(text: str) -> list[float]:
    resp = oai.embeddings.create(model="text-embedding-3-small", input=text)
    return resp.data[0].embedding

# 写入时同步生成向量
doc = col.find_one({"task_id": task_id, "agent": "research"})
col.update_one(
    {"_id": doc["_id"]},
    {"$set": {"embedding": embed_text(doc["output"])}
})

# 查询时：用当前任务的描述向量做相似搜索，只取 top-3
query_embedding = embed_text("如何减少 Agent 上下文中的噪声")
pipeline = [
    {
        "$vectorSearch": {
            "index": "vector_index",       # 需在 Atlas 中预先创建
            "path": "embedding",
            "queryVector": query_embedding,
            "numCandidates": 50,
            "limit": 3
        }
    },
    {"$project": {"output": 1, "metadata": 1, "score": {"$meta": "vectorSearchScore"}}}
]
results = list(db["task_state"].aggregate(pipeline))
for r in results:
    print(f"[score={r['score']:.3f}] {r['output'][:80]}...")