基于 FastAPI + LangChain 的简单agent实现

GitHub 仓库: https://github.com/caozhaoqi/jd_agent

摘要：在 AI Agent 开发中，如何平衡大模型的生成质量与接口的响应速度？本文将详细拆解 jd_agent 项目。这是一个基于岗位描述（JD）自动生成结构化面试指南的智能体。我们采用 FastAPI 作为后端，利用 LangChain 进行逻辑编排，并通过 Asyncio 实现并行推理，将任务耗时从 30s+ 优化至 10s 级。此外，本文还记录了从 OpenAI 迁移至 DeepSeek 的实战经验及工程踩坑记录。

1. 项目背景与痛点

在求职过程中，针对每个岗位进行深度准备（分析技术栈、预测面试题、调研公司背景）是非常耗时的。
传统的 ChatGPT 交互方式虽然可行，但存在以下问题：

Prompt 重复劳动：每次都要输入一大段 Prompt。
输出非结构化：生成的文本难以直接用于后续的数据存储或前端展示。
响应缓慢：分析 JD、出技术题、出 HR 题、背调公司，如果串行执行，等待时间极长。

本项目旨在通过工程化手段解决上述问题，构建一个可扩展、高并发、结构化输出的 AI Agent 后端。

2. 技术架构设计 (System Architecture)

为了保证系统的可维护性和扩展性，本项目采用了分层架构 (Layered Architecture)。

2.1 架构分层图

graph TD
    Client[前端/客户端] -->|HTTP/SSE| API[API Interface Layer]
    
    subgraph Backend [FastAPI Backend]
        API --> Service[Service Layer (Orchestration)]
        
        subgraph Parallel_Execution [异步并行执行区]
            Task_A[JD Meta Parser]
            Task_B[Tech Question Generator]
            Task_C[Company Research Agent]
        end
        
        Service --> Task_A
        Task_A --> Task_B
        Task_A --> Task_C
        
        Task_B & Task_C --> Task_D[HR Question Generator]
        
        Task_B --> Chains[LangChain Logic]
        Task_C --> Chains
        
        Chains --> LLM_Factory[LLM Factory (OpenAI/DeepSeek)]
    end
    
    Service -->|Structured JSON| Client

2.2 核心组件

Web 框架: FastAPI (利用其原生异步特性处理 IO 密集型任务)。
LLM 编排: LangChain (管理 Prompt、Chain 和 OutputParser)。
数据验证: Pydantic (强制 LLM 输出符合 Schema 的 JSON)。
并发控制: Python Asyncio (实现多 Chain 并行调用)。
模型层: 支持 OpenAI 协议，实测完美兼容 DeepSeek-V3。

3. 核心实现细节 (Implementation Details)

3.1 强制结构化输出 (Structured Output)

大模型天生喜欢“聊天”，为了让它输出 JSON，我们摒弃了不稳定的正则表达式，转而使用 LangChain 的 PydanticOutputParser。

Schema 定义 (schemas/interview.py):

class InterviewReport(BaseModel):
    meta: JDMetaData
    tech_questions: List[InterviewQuestion]
    hr_questions: List[InterviewQuestion]
    # 使用 Optional 处理非必填项，增强鲁棒性
    system_design_question: Optional[InterviewQuestion] = None

Chain 实现 (chains/tech_gen.py):

parser = PydanticOutputParser(pydantic_object=QuestionList)
prompt = ChatPromptTemplate.from_template(
    """
    ...
    请严格按照 JSON 格式输出:
    {format_instructions}
    """
)
# 注入 Pydantic 生成的 Schema 指令
chain = prompt | llm | parser

3.2 异步并发编排 (Asyncio Optimization)

这是本项目性能提升的关键。

串行模式：解析 JD (5s) -> 查公司 (10s) -> 出技术题 (10s) -> 出 HR 题 (5s) = 30s+
并行模式：解析 JD (5s) -> [查公司 | 出技术题] (同时进行, 取最大值 10s) -> 出 HR 题 = 约 15s

Service 层实现 (services/interview_service.py):

import asyncio

async def generate_interview_guide(request: JDRequest):
    # 1. 前置依赖：必须先解析 JD
    jd_meta = await parse_jd_async(request.jd_text)
    
    # 2. 构造并行任务 (Fire tasks)
    # 技术题不依赖公司背景
    task_tech = generate_tech_async(jd_meta.tech_stack, jd_meta.years_required)
    # 公司背调是网络 IO 密集型
    task_research = research_company(jd_meta.company_name)
    
    # 3. 核心：并发等待
    tech_qs, company_info = await asyncio.gather(task_tech, task_research)
    
    # 4. 后置依赖：HR 题依赖公司背景
    hr_qs = await generate_hr_async(jd_meta.soft_skills, company_info)
    
    return InterviewReport(...)

3.3 兼容多模型的 LLM 工厂

为了在开发阶段节省成本（使用 DeepSeek），在生产阶段追求稳定（使用 OpenAI），我们设计了工厂模式。

配置解耦 (core/config.py):
通过 .env 文件动态切换底座，无需修改业务代码。

class Settings(BaseSettings):
    OPENAI_API_KEY: str
    OPENAI_API_BASE: str = "https://api.deepseek.com" # 默认指向 DeepSeek
    MODEL_NAME: str = "deepseek-chat"

4. 工程化踩坑与解决方案 (Troubleshooting)

在开发过程中，我们遇到了数个真实的工程挑战，以下是排查记录。

坑位 1：API 余额与鉴权混淆 (429 vs 402)

现象: 程序崩溃，报错 openai.RateLimitError: Error code: 429 或 402 Payment Required。
分析:
- 429 通常指速率限制，但 OpenAI 在余额不足时也会报这个错（insufficient_quota）。
- 402 是明确的余额为零。
解决:
- 不要死磕代码逻辑。
- 切换至 DeepSeek（充值 10 元人民币可用很久）或 SiliconFlow（免费额度）。
- 修改 .env 即可解决。

坑位 2：Pydantic 字段校验异常

现象: ValidationError: system_design_question Input should be a valid dictionary。
分析: 业务逻辑中某些情况下未生成系统设计题，返回了 None，但 Schema 定义默认为必填。

解决: 使用 Optional 类型提示。

1 2	from typing import Optional system_design_question: Optional[InterviewQuestion] = None

坑位 3：SSL 证书警告 (Mac 环境)

现象: NotOpenSSLWarning: urllib3 v2 only supports OpenSSL 1.1.1+
分析: macOS 自带的 LibreSSL 版本过低，与新版 urllib3 不兼容。
解决: 降级 urllib3 或使用 Homebrew 升级 OpenSSL。
1
pip install "urllib3<2.0"

坑位 4：Nginx 反向代理超时 (504 Gateway Time-out)

现象: 爬虫或长文本生成任务耗时超过 60s，前端收到 504 错误。
分析: Nginx 默认 proxy_read_timeout 为 60s。

解决: 在 Nginx 配置中针对 API 路由增加超时时间。

location /api/ {
    proxy_read_timeout 600s;
    proxy_pass http://127.0.0.1:8000;
}

5. 总结与展望

jd_agent 项目展示了如何将一个简单的 Prompt 交互转化为一个工业级的后端服务。

核心收益:

效率提升: 并行处理使响应速度提升 300%。
成本降低: 通过 LLM 工厂模式切换至 DeepSeek，Token 成本降低 90%。
稳定性: 强类型校验和重试机制保证了服务的鲁棒性。

后续规划:

引入 Redis 缓存 JD 解析结果，避免重复扣费。
集成 Celery 处理超长耗时的系统设计题生成。
开发 React/Next.js 前端，实现流式 Markdown 渲染。

如果你觉得这个项目对你有帮助，欢迎在 GitHub 上点个 Star ⭐️！

Zhaoqi.Cao static blog

A static personal blog.