字数 4962,阅读大约需 25 分钟
Python分享:实现一个人工智能agent(生成求职简历的AI助手)
今天,我们将使用python、pydantic_ai、deepseek实现简历生成智能体,实现根据我们的要求,生成求职简历。本地使用到的工具和模型为:
1. 工具:PydanticAI和Python
2. 模型:DeepSeek-Chat(即DeepSeek-V3)
3. 代码仓库:
[email protected]:FunkyGod/fastapi-demo.git4. commit:d7b878502b2de235fa82afee4be5120fe6a44ed8
一、系统目标
• 自动化简历生成:
根据用户输入的简单求职需求(如“3年经验的Python高级工程师”),自动生成一份完整、专业、符合国内就业的求职简历。• 智能化内容填充:
利用大模型(LLM)的能力,结合预定义的工具(Tools),智能推断并填充简历的各个模块,如个人信息、教育背景、工作经验等。• 工具化架构:
将简历的每个部分(个人信息、教育、工作经验等)抽象为独立的工具函数,由Agent根据任务需求自主选择并调用。• 高鲁棒性:设计了回退机制(Fallback),当主Agent(依赖外部API)执行失败时,能够切换到本地模拟数据模式,确保系统可用性。
二、核心组件
1. AI模型配置 (AI Model)
• 模型:使用
deepseek-chat作为核心的语言模型。• Provider:通过
pydantic-ai的OpenAIProvider接入DeepSeek API。• 关键代码:实例化模型,给后续agent使用
model = OpenAIChatModel( model_name="deepseek-chat", provider=OpenAIProvider( base_url="https://api.deepseek.com/beta", api_key="YOUR_API_KEY", ), )
2. 数据模型 (Pydantic Models)
系统使用Pydantic模型来定义简历的结构,确保了数据的类型安全和格式一致性。
•
PersonalInfo: 个人信息•
Education: 教育背景•
WorkExperience: 工作经验•
Skill: 掌握的技能•
Achievement: 突出贡献•
Resume: 最终简历,完整的简历模型,聚合了以上所有部分。
3. Agent核心 (resume_agent)
• 类型:
pydantic_ai.Agent• 输出类型:指定为
Resume模型,确保Agent的最终输出是结构化的简历数据。• 系统提示词 (System Prompt):
•
定义了Agent的角色:“一个专业的简历生成助手”。• 指示
Agent可以调用哪些工具来完成任务。• 要求Agent根据用户需求(如工作年限、技术栈)调整内容,并保证信息的真实性和专业性。
4. 工具函数 (Tools)
简历各部分的生成工具
每个工具函数负责生成简历的一个特定部分,这些函数被 @resume_agent.tool 装饰器标记,以便Agent可以发现并调用它们。
•
generate_personal_info: 生成个人信息。•
generate_education: 生成教育背景。•
generate_work_experience: 生成工作经验。•
generate_skills: 生成技能列表。•
generate_achievements: 生成突出贡献。
这些工具内部包含简单的业务逻辑,例如:
• 根据用户输入中的“3年”、“5年”等关键词,动态调整年龄、毕业年份、工作开始日期。
• 根据“高级”、“初级”等关键词,确定职位级别和成就数量。
• 根据技术栈关键词(如“Python”、“Java”),填充相应的工作描述和技能列表。
• 使用预设的数据池(如姓名、学校、公司列表)增加生成内容的多样性。
动态System Prompt增强
•
add_job_requirements: 使用@resume_agent.system_prompt装饰器,在Agent执行前动态地将用户的原始输入注入到系统提示词中,让Agent更清晰地了解用户意图。
三、执行流程
Please use '!option handwritten true' to enable handwritten简历生成Agent执行流程用户输入需求调用 resume_agent.run()动态注入PromptAgent分析任务API调用成功?yesno任务分解调用 generate_personal_info调用 generate_education调用 其他工具...聚合所有工具返回的数据组装成Resume对象返回RunResult-> "异常处理"捕获异常切换到Fallback模式调用 _fallback_generate_resume生成本地预设简历序列化为JSON并输出
1. 用户输入:用户提供需求,例如
"我需要一份高级软件工程师的简历,有3年工作经验,熟悉Python和Java"。2. Agent启动:
resume_agent.run()方法被调用,deps参数同时传入用户输入,供工具函数上下文使用。3. 动态Prompt注入:
add_job_requirements函数首先执行,将用户需求补充到System Prompt中。4. 任务分解与工具调用:
• Agent接收到包含用户需求的完整提示词后,理解其任务是生成一份
Resume。• 它分析
Resume模型的结构,发现需要personal_info,education,work_experience等字段。• 根据System Prompt中描述的工具能力,Agent自主决定调用
generate_personal_info,generate_education等一系列工具来获取所需数据。• 在调用每个工具时,工具函数可以从
RunContext的deps中获取到原始的用户输入,从而生成与需求匹配的内容。
5. 数据聚合与最终输出:
• Agent收集所有工具返回的数据。
• 将收集到的数据组装成一个完整的
Resume对象。•
run()方法返回RunResult,其中output字段包含了最终生成的Resume对象。
6. 结果展示:程序将
Resume对象序列化为JSON格式并打印输出。
四、异常处理与回退机制
为了增强系统的健壮性,设计了 ResumeAgentWithFallback 类。
• 主流程:尝试使用
resume_agent(真实Agent)生成简历。• 失败捕获:如果
resume_agent.run()在执行过程中(如API调用失败、模型不支持Function Calling等)抛出任何异常,该异常会被捕获。• 切换回退模式:
• 捕获异常后,系统会打印错误信息,并宣布切换到回退模式。
• 调用内部的
_fallback_generate_resume方法。
• 回退实现:此方法不依赖外部API,而是使用本地预设的、固定的数据快速生成一份基础简历,保证了即时响应和系统可用性。
五、启动与执行
• 入口:
if __name__ == "__main__"块是程序的启动点。• 执行逻辑:
1. 首先,尝试在主函数
main()中运行真实的Agent流程。2. 如果
main()失败(即真实Agent执行失败),except块会捕获异常。3. 在
except块中,实例化ResumeAgentWithFallback并调用其generate_resume方法,从而触发带回退机制的流程。
通过这种设计,系统在理想情况(网络通畅、API可用)下提供高质量的智能服务,在异常情况下则能保证基础功能的可用性,实现了优雅降级。
完整代码
"""
真实Agent + Tools简历生成系统
基于Agent_Tools_系统设计文档的架构,使用真实的pydantic_ai Agent实现
"""
import asyncio
import random
from typing import Any, List, Optional
from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext
from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.providers.openai import OpenAIProvider
# 配置AI模型
model = OpenAIChatModel(
model_name="deepseek-chat",
provider=OpenAIProvider(
base_url="https://api.deepseek.com/beta",
api_key="",
),
)
# ============================================================================
# 数据模型定义
# ============================================================================
class PersonalInfo(BaseModel):
"""个人信息模型"""
name: str = Field(description="姓名")
age: int = Field(description="年龄")
gender: str = Field(description="性别")
phone: str = Field(description="电话号码")
email: str = Field(description="邮箱地址")
address: str = Field(description="现居住址")
marital_status: str = Field(description="婚姻状况")
class Education(BaseModel):
"""教育背景模型"""
school: str = Field(description="学校名称")
degree: str = Field(description="学位")
major: str = Field(description="专业")
graduation_year: int = Field(description="毕业年份")
gpa: Optional[float] = Field(default=None, description="GPA成绩")
class WorkExperience(BaseModel):
"""工作经验模型"""
company: str = Field(description="公司名称")
position: str = Field(description="职位")
start_date: str = Field(description="开始日期")
end_date: str = Field(description="结束日期")
description: str = Field(description="工作描述")
achievements: List[str] = Field(description="工作成就")
class Skill(BaseModel):
"""技能模型"""
category: str = Field(description="技能类别")
skills: List[str] = Field(description="技能列表")
proficiency: str = Field(description="熟练程度")
class Achievement(BaseModel):
"""突出贡献模型"""
title: str = Field(description="成就标题")
description: str = Field(description="成就描述")
year: int = Field(description="年份")
impact: str = Field(description="影响程度")
class Resume(BaseModel):
"""完整简历模型"""
personal_info: PersonalInfo = Field(description="个人信息")
education: List[Education] = Field(description="教育背景")
work_experience: List[WorkExperience] = Field(description="工作经验")
skills: List[Skill] = Field(description="技能列表")
achievements: List[Achievement] = Field(description="突出贡献")
summary: str = Field(description="个人总结")
desired_position: str = Field(description="期望职位")
expected_salary: str = Field(description="期望薪资")
# ============================================================================
# 创建Agent
# ============================================================================
resume_agent = Agent(
model=model,
output_type=Resume,
system_prompt=(
"你是一个专业的简历生成助手。用户会描述他们的需求,你需要根据这些需求生成一份完整、专业的中国式求职简历。"
"你可以调用以下工具来获取或生成简历的各个部分:"
"1. generate_personal_info - 生成个人信息"
"2. generate_education - 生成教育背景"
"3. generate_work_experience - 生成工作经验"
"4. generate_skills - 生成技能列表"
"5. generate_achievements - 生成突出贡献"
"根据用户描述的工作经验年限和技术栈来调整简历内容。"
"确保生成的简历信息真实可信,符合中国职场标准和格式。"
),
)
# ============================================================================
# 工具函数定义
# ============================================================================
@resume_agent.system_prompt
async def add_job_requirements(ctx: RunContext[Any]) -> str:
"""添加职位要求到系统提示词"""
user_input = ctx.deps.get("user_input", "生成一份完整的求职简历")
print(f"用户需求:{user_input}")
return f"用户期望生成简历的描述:{user_input}"
@resume_agent.tool
async def generate_personal_info(ctx: RunContext[Any]) -> PersonalInfo:
"""生成个人信息"""
print("=== 工具调用:生成个人信息 ===")
# 从用户需求中提取信息
user_input = ctx.deps.get("user_input", "")
print(f"分析用户需求:{user_input}")
# 数据池
names = ["张三", "李四", "王五", "赵六", "钱七", "孙八", "周九", "吴十"]
addresses = ["北京市朝阳区", "上海市浦东新区", "广州市天河区", "深圳市南山区", "杭州市西湖区"]
# 智能推断信息
if "3年" in user_input:
age = random.randint(25, 30)
elif "5年" in user_input:
age = random.randint(28, 35)
else:
age = random.randint(25, 35)
name = random.choice(names)
gender = random.choice(["男", "女"])
phone = f"1{random.randint(3,9)}{''.join([str(random.randint(0,9)) for _ in range(9)])}"
email = f"{name.lower()}@email.com"
address = random.choice(addresses)
marital_status = random.choice(["未婚", "已婚"])
result = PersonalInfo(
name=name,
age=age,
gender=gender,
phone=phone,
email=email,
address=address,
marital_status=marital_status,
)
print(f"生成个人信息:{result.name}, {result.age}岁, {result.gender}")
return result
@resume_agent.tool
async def generate_education(ctx: RunContext[Any]) -> List[Education]:
"""生成教育背景"""
print("=== 工具调用:生成教育背景 ===")
user_input = ctx.deps.get("user_input", "")
print(f"分析工作经验年限:{user_input}")
schools = [
"清华大学",
"北京大学",
"上海交通大学",
"复旦大学",
"浙江大学",
"中山大学",
"华中科技大学",
]
# 根据工作经验推断毕业年份
if "3年" in user_input:
graduation_year = 2021
elif "5年" in user_input:
graduation_year = 2019
else:
graduation_year = 2020
education_list = []
# 生成本科教育
education_list.append(
Education(
school=random.choice(schools),
degree="本科",
major="计算机科学与技术",
graduation_year=graduation_year,
gpa=round(random.uniform(3.2, 4.0), 2),
)
)
# 高级职位可能包含硕士(40%概率)
if "高级" in user_input and random.random() < 0.4:
education_list.append(
Education(
school=random.choice(schools),
degree="硕士",
major="计算机科学与技术",
graduation_year=graduation_year - 2,
gpa=round(random.uniform(3.4, 4.0), 2),
)
)
print(f"生成教育背景:{len(education_list)}个教育经历")
return education_list
@resume_agent.tool
async def generate_work_experience(ctx: RunContext[Any]) -> List[WorkExperience]:
"""生成工作经验"""
print("=== 工具调用:生成工作经验 ===")
user_input = ctx.deps.get("user_input", "")
print(f"分析用户工作经验:{user_input}")
companies = [
"腾讯",
"阿里巴巴",
"字节跳动",
"百度",
"华为",
"小米",
"京东",
"美团",
"滴滴",
"网易",
]
# 确定职位级别
if "高级" in user_input:
position = "高级软件工程师"
elif "初级" in user_input:
position = "初级软件工程师"
elif "中级" in user_input:
position = "中级软件工程师"
else:
position = "软件工程师"
# 生成工作经历
company = random.choice(companies)
# 根据经验年限生成时间
if "3年" in user_input:
start_date = "2021-07"
elif "5年" in user_input:
start_date = "2019-07"
else:
start_date = "2020-07"
# 生成工作描述
description = f"负责{position}相关工作"
if "Python" in user_input:
description += ",主要使用Python技术栈开发后端服务"
if "Java" in user_input:
description += ",熟悉Java生态系统开发"
description += ",参与系统设计、开发、测试和部署全流程"
work_experience = [
WorkExperience(
company=company,
position=position,
start_date=start_date,
end_date="至今",
description=description,
achievements=[
f"成功完成{company}重要项目,系统稳定性达99.9%",
"获得公司年度优秀员工称号",
"主导技术架构优化,提升系统性能30%",
],
)
]
print(f"生成工作经验:{len(work_experience)}个工作经历,职位:{position}")
return work_experience
@resume_agent.tool
async def generate_skills(ctx: RunContext[Any]) -> List[Skill]:
"""生成技能列表"""
print("=== 工具调用:生成技能列表 ===")
user_input = ctx.deps.get("user_input", "")
print(f"分析技术栈:{user_input}")
skills_list = []
# 编程语言
programming_languages = []
if "Python" in user_input:
programming_languages.append("Python")
if "Java" in user_input:
programming_languages.append("Java")
programming_languages.extend(["JavaScript", "Go"])
skills_list.append(
Skill(
category="编程语言",
skills=programming_languages,
proficiency="熟练" if "高级" in user_input else "了解",
)
)
# 框架技术
frameworks = []
if "Python" in user_input:
frameworks.extend(["Django", "Flask", "FastAPI"])
if "Java" in user_input:
frameworks.extend(["Spring", "Spring Boot", "MyBatis"])
frameworks.extend(["React", "Vue", "Angular"])
skills_list.append(Skill(category="框架技术", skills=frameworks, proficiency="熟练"))
# 数据库
skills_list.append(
Skill(
category="数据库",
skills=["MySQL", "PostgreSQL", "Redis", "MongoDB"],
proficiency="了解",
)
)
# 工具平台
skills_list.append(
Skill(
category="工具平台",
skills=["Git", "Docker", "Kubernetes", "Linux", "Nginx"],
proficiency="熟练",
)
)
print(f"生成技能列表:{len(skills_list)}个技能分类")
return skills_list
@resume_agent.tool
async def generate_achievements(ctx: RunContext[Any]) -> List[Achievement]:
"""生成突出贡献"""
print("=== 工具调用:生成突出贡献 ===")
user_input = ctx.deps.get("user_input", "")
print(f"分析职位级别:{user_input}")
achievements = [
Achievement(
title="技术创新奖",
description="主导开发的新算法提升系统处理效率40%",
year=2023,
impact="为公司节省成本50万元",
),
Achievement(
title="最佳团队协作奖",
description="带领团队完成重要项目,获得客户高度认可",
year=2022,
impact="提升团队整体士气,获得公司表扬",
),
Achievement(
title="优秀员工",
description="连续两年获得公司优秀员工称号",
year=2022,
impact="成为部门技术骨干",
),
Achievement(
title="技术突破奖",
description="突破核心技术难题,获得专利申请",
year=2023,
impact="为公司技术发展做出重要贡献",
),
]
# 根据职位级别选择成就数量
if "高级" in user_input:
num_achievements = random.randint(2, 3)
else:
num_achievements = 1
result = random.sample(achievements, min(num_achievements, len(achievements)))
print(f"生成突出贡献:{len(result)}个成就")
return result
# ============================================================================
# 主函数
# ============================================================================
async def main():
"""主函数:使用真实Agent生成智能简历"""
print("=== 真实Agent + Tools简历生成系统 ===")
print("架构:基于Agent_Tools_系统设计文档的真实实现")
# 用户需求
user_input = "我需要一份高级软件工程师的简历,有3年工作经验,熟悉Python和Java"
print(f"\\n用户需求:{user_input}")
try:
print("\\n=== 开始Agent执行流程 ===")
# 运行Agent
result = await resume_agent.run(user_input, deps={"user_input": user_input})
print("\\n=== Agent执行完成 ===")
print("所有相关工具已成功调用")
# 输出结果
print("\\n=== 生成的完整简历 ===")
print(result.output.model_dump_json(indent=2, ensure_ascii=False))
print("\\n=== 系统执行统计 ===")
print("✓ Agent + Tools架构展示完成")
print("✓ 真实工具调用功能正常")
print("✓ 智能工具选择机制工作")
print("✓ 数据类型安全验证通过")
print("✓ 完整简历结构生成成功")
return result.output
except Exception as e:
print(f"\\nAgent执行过程中发生错误:{str(e)}")
print("\\n=== 错误分析 ===")
print("可能原因:")
print("1. DeepSeek-V3-0324模型不支持function calling")
print("2. API配置问题")
print("3. 网络连接问题")
print("\\n=== 备用方案 ===")
print("建议使用 ctyun_ai_agent_tools_demo.py 作为备用实现")
raise
# ============================================================================
# 异常处理和回退机制
# ============================================================================
class ResumeAgentWithFallback:
"""带回退机制的简历Agent"""
def __init__(self):
self.primary_agent = resume_agent
self.fallback_mode = False
async def generate_resume(self, user_input: str) -> Resume:
"""生成简历,失败时自动切换到回退模式"""
try:
# 尝试使用真实Agent
result = await self.primary_agent.run(user_input, deps={"user_input": user_input})
print("✓ 真实Agent模式成功")
return result.output
except Exception as e:
print(f"⚠️ 真实Agent失败:{str(e)}")
print("切换到回退模式...")
# 切换到回退模式(使用模拟实现)
self.fallback_mode = True
return await self._fallback_generate_resume(user_input)
async def _fallback_generate_resume(self, user_input: str) -> Resume:
"""回退模式的简历生成"""
print("=== 回退模式:使用本地数据生成简历 ===")
# 这里可以复用 ctyun_ai_agent_tools_demo.py 的逻辑
# 为了简洁,这里只提供基本实现
personal_info = PersonalInfo(
name="张三",
age=28,
gender="男",
phone="13800138000",
email="[email protected]",
address="北京市朝阳区",
marital_status="未婚",
)
# 其他字段类似...
print("✓ 回退模式简历生成完成")
return Resume(
personal_info=personal_info,
education=[],
work_experience=[],
skills=[],
achievements=[],
summary="具有3年经验的软件工程师",
desired_position="高级软件工程师",
expected_salary="25k-35k",
)
# ============================================================================
# 启动程序
# ============================================================================
if __name__ == "__main__":
try:
# 尝试真实Agent
asyncio.run(main())
except Exception:
print("\\n真实Agent失败,尝试使用带回退机制的Agent...")
# 使用带回退机制的Agent
agent_with_fallback = ResumeAgentWithFallback()
try:
result = asyncio.run(
agent_with_fallback.generate_resume(
"我需要一份高级软件工程师的简历,有3年工作经验,熟悉Python和Java"
)
)
print("\\n✓ 带回退机制的Agent成功生成简历")
except Exception as e:
print(f"\\n✗ 所有方法都失败了:{str(e)}")
输出日志
D:\Code\fastapi-demo\.venv\Scripts\python.exe D:\Code\fastapi-demo\app\intergration\agent\ctyun_ai_real_agent_resume.py
=== 真实Agent + Tools简历生成系统 ===
架构:基于Agent_Tools_系统设计文档的真实实现
\n用户需求:我需要一份高级软件工程师的简历,有3年工作经验,熟悉Python和Java
\n=== 开始Agent执行流程 ===
用户需求:我需要一份高级软件工程师的简历,有3年工作经验,熟悉Python和Java
=== 工具调用:生成个人信息 ===
分析用户需求:我需要一份高级软件工程师的简历,有3年工作经验,熟悉Python和Java
生成个人信息:张三, 29岁, 女
=== 工具调用:生成教育背景 ===
分析工作经验年限:我需要一份高级软件工程师的简历,有3年工作经验,熟悉Python和Java
生成教育背景:1个教育经历
=== 工具调用:生成工作经验 ===
分析用户工作经验:我需要一份高级软件工程师的简历,有3年工作经验,熟悉Python和Java
生成工作经验:1个工作经历,职位:高级软件工程师
=== 工具调用:生成技能列表 ===
分析技术栈:我需要一份高级软件工程师的简历,有3年工作经验,熟悉Python和Java
生成技能列表:4个技能分类
=== 工具调用:生成突出贡献 ===
分析职位级别:我需要一份高级软件工程师的简历,有3年工作经验,熟悉Python和Java
生成突出贡献:2个成就
\n=== Agent执行完成 ===
所有相关工具已成功调用
\n=== 生成的完整简历 ===
{
"personal_info": {
"name": "张三",
"age": 29,
"gender": "女",
"phone": "16318621890",
"email": "张三@email.com",
"address": "上海市浦东新区",
"marital_status": "已婚"
},
"education": [
{
"school": "复旦大学",
"degree": "本科",
"major": "计算机科学与技术",
"graduation_year": 2021,
"gpa": 3.34
}
],
"work_experience": [
{
"company": "百度",
"position": "高级软件工程师",
"start_date": "2021-07",
"end_date": "至今",
"description": "负责高级软件工程师相关工作,主要使用Python技术栈开发后端服务,熟悉Java生态系统开发,参与系统设计、开发、测试和部署全流程",
"achievements": [
"成功完成百度重要项目,系统稳定性达99.9%",
"获得公司年度优秀员工称号",
"主导技术架构优化,提升系统性能30%"
]
}
],
"skills": [
{
"category": "编程语言",
"skills": [
"Python",
"Java",
"JavaScript",
"Go"
],
"proficiency": "熟练"
},
{
"category": "框架技术",
"skills": [
"Django",
"Flask",
"FastAPI",
"Spring",
"Spring Boot",
"MyBatis",
"React",
"Vue",
"Angular"
],
"proficiency": "熟练"
},
{
"category": "数据库",
"skills": [
"MySQL",
"PostgreSQL",
"Redis",
"MongoDB"
],
"proficiency": "了解"
},
{
"category": "工具平台",
"skills": [
"Git",
"Docker",
"Kubernetes",
"Linux",
"Nginx"
],
"proficiency": "熟练"
}
],
"achievements": [
{
"title": "最佳团队协作奖",
"description": "带领团队完成重要项目,获得客户高度认可",
"year": 2022,
"impact": "提升团队整体士气,获得公司表扬"
},
{
"title": "技术突破奖",
"description": "突破核心技术难题,获得专利申请",
"year": 2023,
"impact": "为公司技术发展做出重要贡献"
}
],
"summary": "拥有3年高级软件工程师经验,精通Python和Java技术栈,熟悉前后端开发全流程。具备扎实的计算机理论基础和丰富的项目实践经验,能够独立完成系统设计、开发和部署。具有良好的团队协作能力和问题解决能力,追求技术创新和业务价值实现。",
"desired_position": "高级软件工程师",
"expected_salary": "25-35K"
}
\n=== 系统执行统计 ===
✓ Agent + Tools架构展示完成
✓ 真实工具调用功能正常
✓ 智能工具选择机制工作
✓ 数据类型安全验证通过
✓ 完整简历结构生成成功更多文章内容
访问我的专属博客:https://www.funkygod.vip/
评论区