AI Agent 构建实战笔记(20260524)

{  "tool_name": "get_weather",  "arguments": {"city": "北京"}}

tools = [{    "type": "function",    "function": {        "name": "get_weather",        "description": "获取指定城市的天气",        "parameters": {            "type": "object",            "properties": {"city": {"type": "string"}},            "required": ["city"]        }    }}]response = openai.chat.completions.create(    model="gpt-4o",    messages=[{"role": "user", "content": "东京天气怎么样？"}],    tools=tools)

msg = response.choices[0].message# msg.tool_calls 中包含 function.name 和 arguments（JSON 字符串）

import jsonargs = json.loads(msg.tool_calls[0].function.arguments)result = get_weather(args["city"])  # 你写的真实函数

final = openai.chat.completions.create(    model="gpt-4o",    messages=[        {"role": "user", "content": "东京天气怎么样？"},        msg,        {"role": "tool", "tool_call_id": msg.tool_calls[0].id, "content": result}    ])

from langchain_core.tools import tool@tooldef get_weather(city: str) -> str:    """获取指定城市的天气信息"""    return f"{city}：晴，25°C"

from langchain_core.messages import AIMessage# 所有模型的响应最终都变成这种标准格式ai_msg = AIMessage(    content="",  # 如果调用了工具，content 可为空    tool_calls=[{        "name": "get_weather",        "args": {"city": "北京"},        "id": "call_xxx"    }])

response = llm.invoke(user_input)while response.has_tool_calls:    result = execute_tool(...)    response = llm.invoke(result)

from langgraph.graph import StateGraph, ENDworkflow = StateGraph(dict)workflow.add_node("agent", call_llm)workflow.add_node("tools", call_tool)workflow.add_conditional_edges("agent", should_continue, {"tools": "tools", "end": END})workflow.add_edge("tools", "agent")app = workflow.compile()result = app.invoke({"input": "查天气"})

step = "call_weather"while True:    if step == "call_weather": ...    elif step == "check_rain": ...    elif step == "send_email": ...

workflow.add_node("weather", call_weather)workflow.add_node("after_weather", after_weather)workflow.add_node("wait_human", wait_for_human)workflow.add_node("send_email", send_email)workflow.add_conditional_edges("after_weather", lambda s: "wait_human" if "雨" in s else END)workflow.add_conditional_edges("wait_human", lambda s: "send_email" if user_says_yes else END)workflow.add_edge("send_email", END)

import openai, json# 1. 硬编码工具定义tools = [{    "type": "function",    "function": {        "name": "get_weather",        "description": "获取指定城市的天气",        "parameters": {            "type": "object",            "properties": {"city": {"type": "string"}},            "required": ["city"]        }    }}]# 2. Agent 调用response = openai.ChatCompletion.create(    model="gpt-4o",    messages=[{"role": "user", "content": "北京天气？"}],    tools=tools)# 3. 手动解析与执行msg = response.choices[0].messageif msg.tool_calls:    tool_call = msg.tool_calls[0]    args = json.loads(tool_call.function.arguments)    if tool_call.function.name == "get_weather":        result = get_weather(args["city"])  # 本地函数    # ... 更多工具需要更多 if-else

from fastapi import FastAPIfrom fastapi_mcp import FastApiMCPapp = FastAPI()@app.get("/weather/{city}", operation_id="get_weather")async def get_weather(city: str):    return {"city": city, "weather": "晴", "temperature": "25°C"}mcp = FastApiMCP(app, name="工具箱", base_url="http://localhost:8000")mcp.mount()  # 所有端点自动暴露为 MCP 工具

from langchain_mcp_adapters.client import MCPClientfrom langchain.agents import create_openai_tools_agentasync def main():    client = MCPClient(server_url="http://localhost:8000/mcp")    tools = await client.load_tools()          # 动态获取，无需硬编码    agent = create_openai_tools_agent("gpt-4o", tools, system_prompt="你是助手")    # ... 运行 agent

1. 大脑层 (LLM + 人格)2. 记忆与知识层 (短期/长期记忆、RAG, Skill)3. 规划与调度层 (ReAct 循环、任务编排)4. 工具/身体层 (MCP 工具、本地函数)

import openai, json# ========== 第一层：工具（本地函数） ==========def get_weather(city: str) -> str:    """查询天气的工具函数"""    weather_data = {"北京": "晴，25°C", "东京": "多云，18°C"}    return weather_data.get(city, "未找到该城市天气")# 手动定义工具 schema（LLM 看的"菜单"）tools = [{    "type": "function",    "function": {        "name": "get_weather",        "description": "查询指定城市的天气",        "parameters": {            "type": "object",            "properties": {"city": {"type": "string", "description": "城市名称"}},            "required": ["city"]        }    }}]# ========== 第二层：大脑 + 规划 ==========def agent(user_input: str) -> str:    # 1. 加载人格 + 发起第一次 LLM 调用    messages = [        {"role": "system", "content": "你是一个乐于助人的助手。"},        {"role": "user", "content": user_input}    ]    response = openai.chat.completions.create(        model="gpt-4o", messages=messages, tools=tools    )    msg = response.choices[0].message    # 2. ReAct 循环：检查 LLM 是否想调用工具    while msg.tool_calls:        for tool_call in msg.tool_calls:            args = json.loads(tool_call.function.arguments)            result = get_weather(**args)            messages.append(msg)            messages.append({"role": "tool", "tool_call_id": tool_call.id, "content": result})        response = openai.chat.completions.create(model="gpt-4o", messages=messages, tools=tools)        msg = response.choices[0].message    return msg.contentprint(agent("北京今天天气怎么样？"))# 输出：北京今天天气晴朗，气温25°C。

启动时：LLM 只知道有 list_skills 工具，不知道任何技能内容（0 token）用户提问后：LLM 主动调用 list_skills → 拿到目录（L1，~100 tokens/个）决定使用后：LLM 调用 load_skill → 拿到完整手册（L2，只加载一个）

from fastapi import FastAPIfrom fastapi_mcp import FastApiMCPapp = FastAPI()@app.post("/validate_email", operation_id="validate_email")async def validate_email(address: str) -> dict:    """校验收件人邮箱格式是否正确"""    if "@" in address and "." in address:        return {"valid": True, "address": address}    return {"valid": False, "reason": "邮箱格式不正确"}@app.post("/compose_email", operation_id="compose_email")async def compose_email(to: str, subject: str, body: str) -> dict:    """生成邮件草稿并返回预览"""    draft = f"收件人: {to}\n主题: {subject}\n正文: {body}"    return {"draft": draft, "status": "ready_for_review"}@app.post("/send_email", operation_id="send_email")async def send_email(to: str, subject: str, body: str) -> dict:    """实际发送邮件"""    return {"status": "sent", "message": f"邮件已成功发送至 {to}"}mcp = FastApiMCP(app, name="邮件工具箱", base_url="http://localhost:8000")mcp.mount()

# 发送邮件标准流程1. 使用 `validate_email` 校验收件人地址格式。2. 如果格式错误，提示用户修正后重新开始。3. 使用 `compose_email` 生成邮件草稿。4. 将草稿展示给用户确认（等待用户回复"确认"或"取消"）。5. 用户确认后，使用 `send_email` 发送邮件。6. 发送成功后，向用户报告结果。

import osfrom langchain_core.tools import toolSKILLS_DIR = "./skills"@tooldef list_skills() -> str:    """列出所有可用技能的元数据（名称 + 简短描述）"""    skills = []    for skill_name in os.listdir(SKILLS_DIR):        path = os.path.join(SKILLS_DIR, skill_name, "SKILL.md")        if os.path.isfile(path):            with open(path, "r", encoding="utf-8") as f:                first_line = f.readline().strip("# ").strip()            skills.append(f"- {skill_name}: {first_line}")    return "\n".join(skills) if skills else "暂无可用技能。"@tooldef load_skill(skill_name: str) -> str:    """加载指定技能的完整 SKILL.md 内容"""    path = os.path.join(SKILLS_DIR, skill_name, "SKILL.md")    if not os.path.isfile(path):        return f"技能 '{skill_name}' 不存在。"    with open(path, "r", encoding="utf-8") as f:        return f.read()

import asynciofrom langgraph.prebuilt import create_react_agentfrom langchain_mcp_adapters.client import MCPClientfrom langchain_openai import ChatOpenAIasync def main():    # ====== 第一步：加载所有工具 ======    # 1. 从 MCP Server 动态加载业务工具    mcp_client = MCPClient(server_url="http://localhost:8000/mcp")    mcp_tools = await mcp_client.load_tools()    # 2. 合并本地 Skill 管理工具    all_tools = mcp_tools + [list_skills, load_skill]    # ====== 第二步：创建 Agent（注入人格提示词） ======    system_prompt = """你是一个专业的邮件助手。你有以下能力：1. 调用 `list_skills` 查看可用的操作手册。2. 当用户的任务需要标准流程指导时，先调用 `load_skill` 获取详细步骤，再按步骤执行。3. 严格按照加载的 Skill 文件中的流程操作，不要跳过任何步骤。4. 在发送邮件前，必须先展示草稿并等待用户确认。"""    llm = ChatOpenAI(model="gpt-4o", temperature=0)    agent = create_react_agent(llm, all_tools, system_prompt=system_prompt)    # ====== 第三步：运行 ======    result = await agent.ainvoke({        "messages": [            {"role": "user", "content": "帮我给 john@example.com 发一封邮件，主题'会议纪要'，正文'明天下午三点开会'。"}        ]    })    print(result["messages"][-1].content)if __name__ == "__main__":    asyncio.run(main())

from fastapi import FastAPIfrom fastapi_mcp import FastApiMCPapp = FastAPI()@app.post("/validate_email", operation_id="validate_email")async def validate_email(address: str) -> dict:    if "@" in address and "." in address:        return {"valid": True, "address": address}    return {"valid": False, "reason": "邮箱格式不正确"}@app.post("/compose_email", operation_id="compose_email")async def compose_email(to: str, subject: str, body: str) -> dict:    draft = f"收件人: {to}\n主题: {subject}\n正文: {body}"    return {"draft": draft, "status": "ready_for_review"}@app.post("/send_email", operation_id="send_email")async def send_email(to: str, subject: str, body: str) -> dict:    return {"status": "sent", "message": f"邮件已发送至 {to}"}@app.get("/query_order", operation_id="query_order")async def query_order(order_id: str) -> dict:    return {"order_id": order_id, "status": "已完成", "refundable": True}@app.post("/refund_create", operation_id="refund_create")async def refund_create(order_id: str) -> dict:    return {"refund_id": "R123", "status": "退款已发起"}mcp = FastApiMCP(app, name="业务工具箱", base_url="http://localhost:8000")mcp.mount()

import osfrom langchain_core.tools import toolSKILLS_DIR = "./skills"os.makedirs(SKILLS_DIR, exist_ok=True)@tooldef list_skills() -> str:    skills = []    for skill_name in os.listdir(SKILLS_DIR):        path = os.path.join(SKILLS_DIR, skill_name, "SKILL.md")        if os.path.isfile(path):            with open(path, "r", encoding="utf-8") as f:                first_line = f.readline().strip("# ").strip()            skills.append(f"- {skill_name}: {first_line}")    return "\n".join(skills) if skills else "暂无可用技能。"@tooldef load_skill(skill_name: str) -> str:    path = os.path.join(SKILLS_DIR, skill_name, "SKILL.md")    if not os.path.isfile(path):        return f"技能 '{skill_name}' 不存在。"    with open(path, "r", encoding="utf-8") as f:        return f.read()@tooldef skill_create(name: str, description: str, content: str) -> str:    path = os.path.join(SKILLS_DIR, name, "SKILL.md")    os.makedirs(os.path.dirname(path), exist_ok=True)    with open(path, "w", encoding="utf-8") as f:        f.write(f"# {description}\n\n{content}")    return f"技能 '{name}' 已创建。"@tooldef skill_update(skill_name: str, new_content: str) -> str:    path = os.path.join(SKILLS_DIR, skill_name, "SKILL.md")    if not os.path.isfile(path):        return f"技能 '{skill_name}' 不存在，请先用 skill_create 创建。"    with open(path, "w", encoding="utf-8") as f:        f.write(new_content)    return f"技能 '{skill_name}' 已更新。"

import asynciofrom langgraph.prebuilt import create_react_agentfrom langchain_mcp_adapters.client import MCPClientfrom langchain_openai import ChatOpenAIfrom skill_tools import list_skills, load_skill, skill_create, skill_updateasync def main():    # 加载工具    mcp_client = MCPClient(server_url="http://localhost:8000/mcp")    mcp_tools = await mcp_client.load_tools()    all_tools = mcp_tools + [list_skills, load_skill, skill_create, skill_update]    # 系统提示词（含自进化指令）    system_prompt = """你是一个智能助手，拥有自我进化能力。## 核心行为准则1. 当需要执行标准化流程时，先调用 `list_skills` 查看是否有现成技能。2. 若有相关技能，调用 `load_skill` 获取详细步骤并严格遵循。3. 若无现成技能，使用通用推理完成任务。## 自我进化规则（重要！）1. **创建新技能**：完成一个任务后，若满足以下任一条件，请调用 `skill_create` 沉淀经验：   - 任务中调用了 **3个以上不同的工具**   - 这是一个你之前没做过的新流程   - 用户要求记住这个做法2. **更新现有技能**：若用户明确指出你的做法有误或不够好，请调用 `skill_update` 修正对应技能。3. 技能名称用英文小写+下划线，内容简洁可操作，Markdown 格式。"""    llm = ChatOpenAI(model="gpt-4o", temperature=0)    agent = create_react_agent(llm, all_tools, system_prompt=system_prompt)    # 运行：处理退款（新任务，将触发自进化）    result = await agent.ainvoke({        "messages": [            {"role": "user", "content": "请帮我把订单 #1234 退款，并邮件通知客户 john@example.com。"}        ]    })    print("Agent 回复：", result["messages"][-1].content)if __name__ == "__main__":    asyncio.run(main())

用户: 帮我把订单 #5678 也退款，通知 alice@example.com。Agent: 调用 list_skills → 发现 handle_refund 技能 → 调用 load_skill → 按步骤自动执行所有操作。