AI Agent开发实战：从0到1构建你的第一个智能助手

2026年，AI Agent赛道融资127亿美元。这个数字背后，是整个行业对"会自己干活"的AI的狂热追捧。

今天，我手把手教你从0到1构建一个AI Agent。不讲概念，直接上代码。

一、什么是AI Agent？

简单说，AI Agent = LLM（大模型）+ Tools（工具）+ Memory（记忆）。

传统AI：用户问，它答。回答完了，结束。

AI Agent：用户说"帮我筛选简历"，它自己调用招聘系统、分析简历、生成报告、把结果发给你。中间可能调用10个工具，做50步推理。

核心区别：一个是给建议，一个是帮你干活。

二、核心架构

class AIAgent:
    def __init__(self, model, tools, memory):
        self.llm = model           # 大模型
        self.tools = tools         # 可用工具
        self.memory = memory       # 记忆系统
    
    def run(self, task):
        # 1. 理解任务
        plan = self.llm.plan(task)
        
        # 2. 规划步骤
        steps = self.llm.decompose(plan)
        
        # 3. 执行循环
        for step in steps:
            if step.need_tool:
                result = self.execute_tool(step.tool, step.params)
            else:
                result = self.llm.think(step)
            
            # 4. 更新记忆
            self.memory.add(step, result)
        
        # 5. 返回结果
        return self.memory.get_result()

三、实战：构建简历筛选Agent

Step 1: 定义工具

from langchain.tools import tool

@tool
def read_resume(file_path: str) -> dict:
    """读取简历文件，返回结构化数据"""
    import json
    with open(file_path, 'r', encoding='utf-8') as f:
        data = json.load(f)
    return {
        'name': data.get('name', ''),
        'skills': data.get('skills', []),
        'experience': data.get('experience', []),
        'education': data.get('education', ''),
        'score': 0
    }

@tool
def score_resume(resume: dict, requirements: dict) -> dict:
    """根据岗位要求给简历打分"""
    score = 0
    
    # 技能匹配度 (40分)
    required_skills = set(requirements.get('skills', []))
    candidate_skills = set(resume.get('skills', []))
    skill_match = len(required_skills & candidate_skills) / len(required_skills)
    score += skill_match * 40
    
    # 经验年限 (30分)
    exp_years = resume.get('experience_years', 0)
    required_exp = requirements.get('min_experience', 0)
    if exp_years >= required_exp:
        score += 30
    elif exp_years >= required_exp * 0.8:
        score += 20
    
    # 学历匹配 (30分)
    edu_score = {'博士': 30, '硕士': 25, '本科': 20, '专科': 10}
    edu = edu_score.get(resume.get('education', ''), 0)
    required_edu = requirements.get('min_education', '本科')
    if edu >= edu_score.get(required_edu, 20):
        score += edu
    
    resume['score'] = round(score, 1)
    return resume

@tool
def send_notification(recipient: str, content: str):
    """发送通知（邮件/消息）"""
    # 实际场景：调用邮件API或消息队列
    print(f"通知已发送给 {recipient}: {content}")
    return {"status": "sent"}

Step 2: 构建Agent

from langchain.agents import AgentExecutor, create_react_agent
from langchain_openai import ChatOpenAI

# 初始化模型
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 定义工具列表
tools = [read_resume, score_resume, send_notification]

# 创建Agent
agent = create_react_agent(llm, tools)

# 创建执行器
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=10
)

# 运行
result = agent_executor.invoke({
    "input": """
    帮我筛选简历，岗位要求：
    - 技能：Python, SQL, 数据分析
    - 最低经验：3年
    - 最低学历：本科
    
    简历文件在 /data/resumes/ 目录下。
    把分数最高的前5份简历整理成报告发给我。
    """
})

print(result['output'])

Step 3: 添加记忆系统

from langchain.memory import ConversationBufferMemory

class ResumeAgentWithMemory:
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4o", temperature=0)
        self.tools = [read_resume, score_resume, send_notification]
        self.memory = ConversationBufferMemory(
            memory_key="chat_history",
            return_messages=True
        )
        self.agent = create_react_agent(self.llm, self.tools)
        self.executor = AgentExecutor(
            agent=self.agent,
            tools=self.tools,
            memory=self.memory,
            verbose=True
        )
    
    def run(self, task: str):
        # 加入历史上下文
        history = self.memory.chat_memory.messages
        context = f"之前的对话：{history}\n\n当前任务：{task}"
        
        result = self.executor.invoke({"input": context})
        
        # 记忆关键结果
        self.memory.chat_memory.add_user_message(task)
        self.memory.chat_memory.add_ai_message(result['output'])
        
        return result
    
    def get_candidates(self, min_score: float = 70):
        """查询分数>=70的候选人"""
        # 实际场景：从数据库查询
        pass

四、避坑指南

坑1: 工具调用死循环

# ❌ 错误：Agent可能在两个工具间无限循环
# tool_a -> tool_b -> tool_a -> tool_b ...

# ✅ 解决：设置 max_iterations + 工具优先级
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    max_iterations=10,  # 最多执行10步
    early_stopping=True   # 结果足够好就停止
)

坑2: Token溢出

# ❌ 错误：传入太多简历导致上下文溢出

# ✅ 解决：分批处理 + 摘要压缩
def batch_process_resumes(resumes, batch_size=10):
    results = []
    for i in range(0, len(resumes), batch_size):
        batch = resumes[i:i+batch_size]
        # 先摘要，再合并
        batch_summary = summarize_batch(batch)
        results.append(batch_summary)
    return merge_summaries(results)

坑3: 工具返回格式不一致

# ✅ 解决：统一返回格式 + Schema验证
from pydantic import BaseModel

class ResumeScoreResult(BaseModel):
    name: str
    score: float
    match_reasons: List[str]
    risks: List[str]

def score_resume(resume: dict, requirements: dict) -> ResumeScoreResult:
    # 标准化处理
    score_result = _calculate_score(resume, requirements)
    return ResumeScoreResult(
        name=resume['name'],
        score=score_result['total'],
        match_reasons=score_result['strengths'],
        risks=score_result['weaknesses']
    )

五、生产环境注意事项

1. 监控Agent行为：记录每一步的工具调用和结果，便于调试和审计
2. 设置熔断机制：单次任务超时、每日调用量限制
3. 结果校验：Agent输出可能有幻觉，重要决策需要人工复核
4. 成本控制：LLM调用是成本大户，做好Token统计和预算

六、总结

AI Agent开发的核心就三件事：

1. 定义好工具：工具越精准，Agent越能干
2. 设计好Prompt：让Agent理解什么时候用什么工具
3. 管理好记忆：让Agent在长对话中保持连贯

工具不会淘汰人，但会用工具的人会淘汰不会用的人。

与其担心被AI替代，不如学会让AI替你干活。