用 Anthropic 的方法构建有效 Agents：工程化笔记

GAIA/
├── config.py              # 环境变量驱动的全局配置
├── agent.py               # 核心：System Prompt + StateGraph + 答案提取
├── tools.py               # 基础工具层（搜索/文件/计算）
├── extension_tools.py     # 扩展工具层（PDF/Excel/OCR/音频/视觉）
├── rag.py                 # RAG 层（FAISS 向量检索 + 解题建议）
├── app.py                 # Gradio UI + 批量评测入口
└── data_clean.csv         # RAG 知识库源数据

# Workflow：代码控制流程，LLM 只做子任务
def translate_and_review(text):
    # 第一步：翻译（代码决定先翻译）
    translation = llm.invoke("请将以下英文翻译为中文：" + text)
 
    # 第二步：校对（代码决定翻译完再校对）
    review = llm.invoke("请校对这段翻译，指出错误：" + translation)
 
    # 第三步：修正（代码决定校对完再修正）
    if "错误" in review:
        final = llm.invoke("根据校对意见修正翻译：" + review)
    else:
        final = translation
 
    return final  # 代码决定何时结束

# Agent：LLM 控制流程，代码只提供循环框架
def agent_loop(goal):
    messages = [SystemMessage("你是翻译助手，可以使用以下工具...")]
    while True:
        response = llm.invoke(messages)      # LLM 自己决定下一步
        if not response.tool_calls:          # LLM 自己决定何时结束
            return response.content
        results = execute_tools(response.tool_calls)  # 执行 LLM 选择的工具
        messages.append(results)

# config.py — 核心配置项
MAX_ITERATIONS = int(os.getenv("MAX_ITERATIONS", "10"))     # Agent 最大迭代次数
LLM_TIMEOUT = int(os.getenv("LLM_TIMEOUT", "120"))          # LLM 单次调用超时
TOOL_TIMEOUT = int(os.getenv("TOOL_TIMEOUT", "30"))          # 工具调用超时
MAX_FILE_SIZE = int(os.getenv("MAX_FILE_SIZE", "10000"))     # 工具输出截断长度
RATE_LIMIT_RETRY_MAX = int(os.getenv("RATE_LIMIT_RETRY_MAX", "5"))        # 429 重试次数
RATE_LIMIT_RETRY_BASE_DELAY = float(os.getenv("RATE_LIMIT_RETRY_BASE_DELAY", "10"))  # 指数退避基数
BATCH_QUESTION_DELAY = float(os.getenv("BATCH_QUESTION_DELAY", "5"))      # 批量评测间隔

# agent.py — System Prompt 中的工具使用策略（摘录）
## 工具使用策略
### 优先级顺序
0. **先查知识库**【最高优先级】:
   - 首先调用 rag_query(question) 查询知识库
   - 如果返回"知识库匹配成功"，直接使用该答案
1. **有附件的问题**【重要】:
   - 第一步：用 fetch_task_files(task_id) 下载文件
   - 第二步：根据扩展名选择正确的解析工具
   - 第三步：分析文件内容，进行必要的计算
2. **需要外部信息**: web_search / wikipedia_page / arxiv_search
3. **需要计算**: 简单算术用 calc，复杂处理用 run_python

# agent.py — RAG 前置短路
def __call__(self, question: str, task_id: str = None) -> str:
    # ===== RAG 前置短路：高置信度匹配直接返回 =====
    if rag_lookup_answer is not None:
        hit = rag_lookup_answer(question, min_similarity=0.85)
        if hit and hit.get("answer"):
            return str(hit["answer"]).strip()
    # ===== 未命中才进入 LangGraph 循环 =====
    result = self.graph.invoke(initial_state)
    answer = extract_final_answer(result)
    # 检查答案是否需要格式化
    if self._needs_reformatting(answer):
        answer = self._force_format_answer(result)
    return answer

根据文件扩展名选择正确的读取工具：
  .xlsx / .xls → 必须用 parse_excel(file_path)
  .pdf         → 必须用 parse_pdf(file_path)
  .txt / .csv  → 用 read_file(file_path)
  .png / .jpg  → 用 image_ocr() 或 analyze_image()
  .mp3 / .wav  → 用 transcribe_audio(file_path)

# agent.py — 路由判断
def should_continue(state: AgentState) -> Literal["tools", "end"]:
    last_message = state["messages"][-1]
    iteration = state.get("iteration_count", 0)
 
    if iteration >= MAX_ITERATIONS:        # 达到上限 → 强制结束
        return "end"
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"                     # 有工具调用 → 继续
    return "end"                           # 无工具调用 → 返回答案

# agent.py — 答案格式评审器
def _needs_reformatting(self, answer: str) -> bool:
    if not answer or answer == "无法获取答案":
        return False
    indicators = [
        answer.startswith('http'),           # 答案是 URL
        len(answer) > 300,                   # 答案过长
        answer.count('\n') > 3,              # 多段落
        answer.startswith('1.') and '2.' in answer,  # 列表格式
        '...' in answer and len(answer) > 100,       # 省略号 + 过长
    ]
    return any(indicators)
 
def _force_format_answer(self, result: dict) -> str:
    format_prompt = (
        "根据上述对话收集的信息，输出最终答案。\n\n"
        "【强制要求】只输出答案本身，不要解释、不要前缀。\n"
        "- 数字：直接输出（如 42）\n"
        "- 人名/地名：直接输出（如 Albert Einstein）\n"
        "- 日期：YYYY-MM-DD\n"
        "- 是/否：Yes 或 No\n"
    )
    # 用不绑定工具的 LLM 重新格式化
    response = invoke_llm_with_retry(get_llm(), full_messages)
    return extract_final_answer({"messages": [response]})

# agent.py — Graph 构建
class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]
    iteration_count: int
 
def build_agent_graph():
    graph = StateGraph(AgentState)
    graph.add_node("assistant", assistant)
    graph.add_node("tools", ToolNode(ALL_TOOLS))
    graph.set_entry_point("assistant")
    graph.add_conditional_edges(
        "assistant", should_continue,
        {"tools": "tools", "end": END}
    )
    graph.add_edge("tools", "assistant")
    return graph.compile()

# agent.py — assistant 节点（完整逻辑）
def assistant(state: AgentState) -> dict:
    messages = state["messages"]
    iteration = state.get("iteration_count", 0) + 1
    full_messages = [SystemMessage(content=SYSTEM_PROMPT)] + list(messages)
 
    if iteration >= MAX_ITERATIONS - 1:
        # 第三级：最后一轮，解绑工具，强制输出纯文本答案
        warning = f"⚠️ 【最后机会】已进行 {iteration} 次迭代，达到上限。"
               +  "你必须立即给出最终答案！不要再调用任何工具！"
        full_messages.append(SystemMessage(content=warning))
        response = invoke_llm_with_retry(get_llm(), full_messages)  # 注意：不绑定工具
    elif iteration >= MAX_ITERATIONS - 2:
        # 第二级：倒数第二轮，提醒收敛，但仍允许工具调用
        warning = f"⚠️ 警告：已进行 {iteration} 次迭代，接近上限，请尽快给出最终答案。"
        full_messages.append(SystemMessage(content=warning))
        response = invoke_llm_with_retry(get_llm_with_tools(), full_messages)
    else:
        # 第一级：正常轮次，允许自由调用工具
        response = invoke_llm_with_retry(get_llm_with_tools(), full_messages)
 
    return {"messages": [response], "iteration_count": iteration}

# agent.py — 带重试的 LLM 调用
def invoke_llm_with_retry(llm, messages, max_retries=None, base_delay=None):
    from openai import RateLimitError
    if max_retries is None:
        max_retries = RATE_LIMIT_RETRY_MAX       # 默认 5 次
    if base_delay is None:
        base_delay = RATE_LIMIT_RETRY_BASE_DELAY  # 默认 10 秒
 
    for attempt in range(max_retries + 1):
        try:
            return llm.invoke(messages)
        except RateLimitError:
            if attempt < max_retries:
                delay = base_delay * (2 ** attempt)  # 指数退避：10s, 20s, 40s, 80s, 160s
                time.sleep(delay)
            else:
                raise

# agent.py — 工具渐进加载
from tools import BASE_TOOLS                        # 第一层：总是可用
 
try:
    from extension_tools import EXTENSION_TOOLS      # 第二层：需要额外依赖
    ALL_TOOLS = BASE_TOOLS + EXTENSION_TOOLS
except ImportError:
    ALL_TOOLS = BASE_TOOLS
 
try:
    from rag import RAG_TOOLS                        # 第三层：需要向量库
    ALL_TOOLS = ALL_TOOLS + RAG_TOOLS
except ImportError:
    pass

# tools.py — fetch_task_files 的类型路由提示
next_step_hint = ""
if file_ext in ['.xlsx', '.xls']:
    next_step_hint = "\n\n⚠️ 下一步：请立即使用 parse_excel(file_path) 读取此 Excel 文件"
elif file_ext == '.pdf':
    next_step_hint = "\n\n⚠️ 下一步：请立即使用 parse_pdf(file_path) 读取此 PDF 文件"
elif file_ext in ['.png', '.jpg', '.jpeg']:
    next_step_hint = "\n\n⚠️ 下一步：请使用 image_ocr() 或 analyze_image() 处理此图片"
elif file_ext in ['.mp3', '.wav']:
    next_step_hint = "\n\n⚠️ 下一步：请使用 transcribe_audio() 转写此音频文件"
 
return f"文件已下载到: {file_path}\n文件大小: {file_size} 字节{next_step_hint}"

# tools.py — run_python 沙箱核心
ALLOWED_MODULES = {
    'math': math, 're': re_module, 'json': json_module,
    'datetime': datetime_module, 'collections': collections_module,
    'random': random_module, 'string': string_module,
    'itertools': itertools_module, 'functools': functools_module,
}
 
def restricted_import(name, globals=None, locals=None, fromlist=(), level=0):
    if name not in ALLOWED_MODULES:
        raise ImportError(f"不允许导入模块 '{name}'")
    return ALLOWED_MODULES[name]
 
safe_builtins = {
    'list': list, 'dict': dict, 'set': set, 'tuple': tuple,
    'str': str, 'int': int, 'float': float, 'bool': bool,
    'print': print, 'len': len, 'range': range, 'enumerate': enumerate,
    'zip': zip, 'map': map, 'filter': filter, 'sorted': sorted,
    'sum': sum, 'min': min, 'max': max, 'abs': abs, 'round': round,
    '__import__': restricted_import,
    'True': True, 'False': False, 'None': None,
}
 
namespace = {"__builtins__": safe_builtins}
namespace.update(preloaded)  # 预注入 math, re, datetime, Counter 等
 
old_stdout = sys.stdout
sys.stdout = io.StringIO()
try:
    exec(code, namespace)
    output = sys.stdout.getvalue()
finally:
    sys.stdout = old_stdout

# rag.py — GAIARAGManager 核心结构
class GAIARAGManager:
    def __init__(self):
        self._embeddings = None   # 延迟加载 HuggingFace Embeddings
        self._llm = None          # 延迟加载 LLM
        self._vectorstore = None  # 延迟加载 FAISS 索引
 
    def retrieve_with_scores(self, query, k=3):
        """带相似度分数的检索"""
        return self.vectorstore.similarity_search_with_score(query, k=k)
 
    def query(self, question, k=3):
        """检索 + LLM 生成解题建议"""
        docs = self.retrieve(question, k=k)
        chain = self.rag_prompt | self.llm
        return chain.invoke({"context": context, "question": question})

# rag.py — rag_query 三级响应
best_doc, best_score = results[0]
similarity = 1 / (1 + best_score)  # FAISS L2 距离转相似度
 
if similarity > 0.85:
    # 高相似度：直接返回答案（知识库匹配成功）
    return f"【知识库匹配成功】直接答案: {answer}"
 
if similarity > 0.6:
    # 中等相似度：返回答案 + 解题参考
    return "【知识库参考】\n" + 参考文档
 
# 低相似度：调用 LLM 生成解题建议
return manager.query(question)

# rag.py — 短路查找
def rag_lookup_answer(question: str, min_similarity: float = 0.85):
    manager = get_rag_manager()
    results = manager.retrieve_with_scores(question.strip(), k=1)
    if not results:
        return None
    best_doc, best_score = results[0]
    similarity = 1.0 / (1.0 + float(best_score))
    answer = (best_doc.metadata.get("answer") or "").strip()
    if answer and similarity > min_similarity:
        return {"answer": answer, "similarity": similarity}
    return None

# agent.py — 答案提取管道
def extract_final_answer(result: dict) -> str:
    # Step 0: 优先选择"无 tool_calls 的 AIMessage"（真正的最终答案）
    for msg in reversed(messages):
        if isinstance(msg, AIMessage) and msg.content:
            if not (hasattr(msg, "tool_calls") and msg.tool_calls):
                content = msg.content
                break
 
    # Step 1: 移除常见前缀
    prefix_patterns = [
        r'^(?:the\s+)?(?:final\s+)?answer\s*(?:is|:)\s*',
        r'^(?:therefore|thus|so|hence)[,:]?\s*',
        r'^(?:最终)?答案[是为：:]\s*',
        r'^根据(?:以上)?(?:分析|信息|计算)[，,：:]?\s*',
    ]
 
    # Step 2: 移除尾部解释
    suffix_patterns = [
        r'\s*(?:This|That)\s+(?:is|was|represents).*$',
        r'\s*[（\(].*[）\)]$',
    ]
 
    # Step 3: 提取 JSON 格式答案
    # Step 4: 清理空白和引号
    # Step 5: 数字格式处理（移除千分位逗号）

# app.py — 批量评测
def on_run_evaluation(username: str, num_questions: int, progress=gr.Progress()):
    questions = get_questions()[:num_questions]
    correct = 0
 
    for i, q in enumerate(questions):
        # 问题间延迟（避免触发速率限制）
        if i > 0 and BATCH_QUESTION_DELAY > 0:
            time.sleep(BATCH_QUESTION_DELAY)
 
        task_id, question = q["task_id"], q["question"]
        answer = solve_question(question, task_id)
        submit_result = submit_answer(task_id, answer, username)
        correct += int(submit_result.get("is_correct", False))
 
    accuracy = correct / len(questions) * 100
    return f"正确: {correct}/{len(questions)} ({accuracy:.1f}%)"

# app.py — 后台预加载
def preload_agent():
    def _preload():
        agent = get_agent()
        get_llm_with_tools()  # 触发 LLM 单例初始化
    thread = threading.Thread(target=_preload, daemon=True)
    thread.start()
 
preload_agent()  # 启动时执行