大语言模型 LLM 与 Agent：从下一个 token 到可验证行动闭环

这个例子用一个可替换的 fake_llm 展示 Agent 结构。真实系统中，fake_llm 可以替换为任意 LLM API，但 schema、工具执行和验证边界应该保留。

import math
from collections import Counter

DOCUMENTS = [
    {"id": "did", "text": "Difference-in-differences compares treated and control changes over time."},
    {"id": "psm", "text": "Propensity score matching balances observed covariates before comparing outcomes."},
    {"id": "uat", "text": "Universal approximation says a wide neural network can approximate continuous functions."},
]

def tokenize(text):
    return [word.strip(".,:;!?").lower() for word in text.split()]

def vectorize(text):
    return Counter(tokenize(text))

def cosine(a, b):
    shared = set(a) & set(b)
    dot = sum(a[key] * b[key] for key in shared)
    norm_a = math.sqrt(sum(value * value for value in a.values()))
    norm_b = math.sqrt(sum(value * value for value in b.values()))
    return 0.0 if norm_a == 0 or norm_b == 0 else dot / (norm_a * norm_b)

def retrieve(query, k=2):
    qv = vectorize(query)
    ranked = sorted(
        DOCUMENTS,
        key=lambda doc: cosine(qv, vectorize(doc["text"])),
        reverse=True,
    )
    return ranked[:k]

def validate_action(action):
    allowed = {"search": {"query"}, "draft": {"claim", "evidence_ids"}}
    if action.get("name") not in allowed:
        raise ValueError("unknown tool")
    missing = allowed[action["name"]] - set(action.get("arguments", {}))
    if missing:
        raise ValueError(f"missing arguments: {missing}")
    return action

def run_tool(action):
    args = action["arguments"]
    if action["name"] == "search":
        return retrieve(args["query"])
    if action["name"] == "draft":
        ids = ", ".join(args["evidence_ids"])
        return f"{args['claim']} Evidence: {ids}."
    raise ValueError("unreachable")

def fake_llm(state):
    # Replace this with an LLM call. Keep the action schema and verifier outside the model.
    if not state["docs"]:
        return {"name": "search", "arguments": {"query": state["task"]}}
    evidence_ids = [doc["id"] for doc in state["docs"]]
    return {
        "name": "draft",
        "arguments": {
            "claim": "Use RAG before answering empirical-method questions.",
            "evidence_ids": evidence_ids,
        },
    }

def verify(answer, docs):
    cited = {doc["id"] for doc in docs}
    return all(doc_id in answer for doc_id in cited)

state = {"task": "Explain DID and PSM for a research assistant.", "docs": [], "trace": []}

for step in range(3):
    action = validate_action(fake_llm(state))
    observation = run_tool(action)
    state["trace"].append({"action": action, "observation": observation})
    if action["name"] == "search":
        state["docs"] = observation
    else:
        if verify(observation, state["docs"]):
            state["answer"] = observation
            break
        state["task"] += " Cite retrieved evidence explicitly."

print(state["answer"])
print("trace length:", len(state["trace"]))

• 用户提出：“帮我解释 DID 和 PSM，并给出 Stata/R 实操提醒。” 如果只是一次 LLM 生成，答案可能流畅但无法确认依据、版本和执行路径。
• Agent 版本会先把任务写入 state，检索课程知识页和方法文档，再生成结构化工具调用，例如 search(query)、open_file(path)、run_code(cmd) 或 render_table(model)。
• 每次工具调用都会产生 observation：检索到的文档、代码输出、表格路径或错误日志。Agent 把 observation 写回上下文，再决定继续检索、写草稿、运行验证还是请求人工确认。
• 最终答案必须带有可检查证据：引用了哪些知识页、运行了哪些命令、是否通过 route/build/test、哪些假设仍需人类判断。这就是 Agent 与普通聊天机器人的分界线。

参考资料

• Brown et al. (2020), Language Models are Few-Shot Learnershttps://arxiv.org/abs/2005.14165
• Lewis et al. (2020), Retrieval-Augmented Generation for Knowledge-Intensive NLP Taskshttps://arxiv.org/abs/2005.11401
• Yao et al. (2022), ReAct: Synergizing Reasoning and Acting in Language Modelshttps://arxiv.org/abs/2210.03629
• Schick et al. (2023), Toolformer: Language Models Can Teach Themselves to Use Toolshttps://arxiv.org/abs/2302.04761