Prompt 工程已死，上下文工程当立

用户提问 -> 向量化 -> 从数据库检索 Top-K 相关文档 -> 塞进 Prompt -> LLM 生成答案

tools = [
    {
        "name": "search_question_bank",
        "description": "从面试题库中检索与指定技能相关的面试题。返回题目、难度、参考答案要点。",
        "parameters": {
            "skill": "技能标签，如 'Python', 'System Design', 'Kubernetes'",
            "difficulty": "难度等级：easy, medium, hard",
            "count": "返回题目数量"
        }
    },
    {
        "name": "evaluate_answer",
        "description": "评估候选人的回答质量，返回得分（1-10）和简评。",
        "parameters": {
            "question": "原始问题",
            "candidate_answer": "候选人的回答",
            "reference_points": "参考答案要点"
        }
    },
    {
        "name": "schedule_next_round",
        "description": "为候选人安排下一轮面试。",
        "parameters": {
            "candidate_id": "候选人ID",
            "interviewer_id": "面试官ID",
            "time_slot": "时间段"
        }
    },
    {
        "name": "send_rejection_email",
        "description": "发送委婉的拒绝邮件。",
        "parameters": {
            "candidate_email": "候选人邮箱",
            "feedback": "反馈内容"
        }
    }
]

{
  "candidate_id": "C-2026-001",
  "name": "张三",
  "position": "Senior Backend Engineer",
  "skills": ["Go", "Kubernetes", "PostgreSQL"],
  "previous_interviews": [
    {"date": "2026-01-10", "round": "Phone Screen", "result": "Pass", "notes": "基础扎实"}
  ]
}

{
  "session_id": "S-20260116-001",
  "questions_asked": [
    {"q": "请解释 Go 的 GMP 模型", "a": "...", "score": 8}
  ],
  "current_phase": "technical_deep_dive",
  "overall_impression": "沟通清晰，技术细节掌握较好"
}

def assemble_context(session_state: dict, user_message: str) -> str:
    """动态组装上下文"""
    context_parts = []

    # 1. 根据阶段选择不同的 System Prompt
    if session_state["current_phase"] == "opening":
        context_parts.append(PROMPT_OPENING)  # "你是友好的面试官，先做自我介绍..."
    elif session_state["current_phase"] == "technical_deep_dive":
        context_parts.append(PROMPT_TECHNICAL)  # "你是严谨的技术面试官，追问细节..."
    elif session_state["current_phase"] == "closing":
        context_parts.append(PROMPT_CLOSING)  # "总结面试，询问候选人问题..."

    # 2. 注入岗位 JD（压缩版，只保留关键要求）
    jd_summary = summarize_jd(session_state["job_description"])
    context_parts.append(f"【岗位要求摘要】\n{jd_summary}")

    # 3. 注入候选人简历（重点突出匹配项）
    resume_highlights = extract_highlights(
        session_state["candidate_resume"],
        session_state["job_description"]
    )
    context_parts.append(f"【候选人亮点】\n{resume_highlights}")

    # 4. 注入本场面试历史（只保留最近 3 轮问答，防止 Context Overflow）
    recent_qa = session_state["questions_asked"][-3:]
    context_parts.append(f"【本场面试记录】\n{format_qa(recent_qa)}")

    # 5. 如果需要出新题，调用 RAG 检索题库
    if should_ask_new_question(session_state, user_message):
        skill_to_test = infer_skill_gap(session_state)
        questions = search_question_bank(skill=skill_to_test, difficulty="medium", count=2)
        context_parts.append(f"【备选题目】\n{format_questions(questions)}")

    return "\n\n".join(context_parts)

AI 面试官：你好张三，我是今天的技术面试官。看了你的简历，你在云原生领域有不少经验。
          我们今天主要聊聊 Kubernetes 和 Go 相关的话题。准备好了吗？

候选人：Go 的 GMP 模型就是 Goroutine、Machine 和 Processor...

AI 面试官：[内部调用 evaluate_answer，得分 7/10，缺少对抢占式调度的描述]
          不错的解释。我追问一下，Go 1.14 之后引入的异步抢占机制，解决了什么问题？

候选人：呃...这个我不太确定...

AI 面试官：[内部记录：抢占式调度-不熟悉]
          没关系，这是一个比较深入的点。我们换个方向，聊聊你在上家公司做的 K8s 项目...

AI 面试官：[内部调用：汇总得分、生成评语、决定是否进入下一轮]
          好的张三，今天的面试就到这里。总体来说，你的基础扎实，项目经验也很匹配。
          HR 会在三个工作日内联系你。你有什么问题想问我吗？

[内部调用 schedule_next_round(...)]

pip install langchain langchain-openai chromadb pydantic

"""
AI 面试官 Agent - 上下文工程实战
展示 RAG + Memory + Tools + Dynamic Prompts 的组合使用
"""
import os
from typing import List, Dict, Optional
from pydantic import BaseModel, Field

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_core.tools import tool
from langchain_community.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.agents import AgentExecutor, create_openai_tools_agent

# ========== 1. 数据模型 ==========
class Candidate(BaseModel):
    """候选人档案（长期记忆）"""
    id: str
    name: str
    position: str
    skills: List[str]
    resume_text: str
    previous_interviews: List[Dict] = []

class InterviewSession(BaseModel):
    """面试会话（短期记忆）"""
    session_id: str
    candidate: Candidate
    questions_asked: List[Dict] = []
    current_phase: str = "opening"  # opening, technical, behavioral, closing
    scores: List[int] = []

# ========== 2. 模拟数据 ==========
# 面试题库（实际项目中存在 ES 或向量数据库）
QUESTION_BANK = [
    {"id": "q1", "skill": "Go", "difficulty": "medium", 
     "question": "请解释 Go 的 GMP 调度模型", 
     "key_points": ["Goroutine", "Machine", "Processor", "work stealing"]},
    {"id": "q2", "skill": "Go", "difficulty": "hard", 
     "question": "Go 1.14 引入的异步抢占机制解决了什么问题？", 
     "key_points": ["死循环", "GC STW", "信号抢占"]},
    {"id": "q3", "skill": "Kubernetes", "difficulty": "medium", 
     "question": "Pod 的生命周期有哪些阶段？", 
     "key_points": ["Pending", "Running", "Succeeded", "Failed", "Unknown"]},
    {"id": "q4", "skill": "System Design", "difficulty": "hard", 
     "question": "设计一个高可用的消息队列系统", 
     "key_points": ["持久化", "复制", "消费确认", "死信队列"]},
]

# 模拟候选人
SAMPLE_CANDIDATE = Candidate(
    id="C-2026-001",
    name="张三",
    position="Senior Backend Engineer",
    skills=["Go", "Kubernetes", "PostgreSQL"],
    resume_text="""
    张三 - 高级后端工程师
    工作经验：5年
    技能：Go, Kubernetes, PostgreSQL, Redis
    项目经历：
    1. 某大厂微服务改造项目，负责服务拆分和 K8s 迁移
    2. 高并发订单系统，QPS 达到 10w+
    """
)

# ========== 3. 定义工具 (Tools) ==========
@tool
def search_question_bank(skill: str, difficulty: str = "medium") -> str:
    """从面试题库中检索与指定技能相关的面试题。

    Args:
        skill: 技能标签，如 'Go', 'Kubernetes', 'System Design'
        difficulty: 难度等级：easy, medium, hard
    """
    results = [q for q in QUESTION_BANK 
               if q["skill"].lower() == skill.lower() 
               and q["difficulty"] == difficulty]
    if not results:
        results = [q for q in QUESTION_BANK if q["skill"].lower() == skill.lower()]

    if results:
        q = results[0]
        return f"题目: {q['question']}\n参考要点: {', '.join(q['key_points'])}"
    return "未找到相关题目"

@tool
def evaluate_answer(question: str, candidate_answer: str, key_points: str) -> str:
    """评估候选人的回答质量，返回得分（1-10）和简评。

    Args:
        question: 原始问题
        candidate_answer: 候选人的回答
        key_points: 参考答案要点，逗号分隔
    """
    # 实际项目中，这里会调用 LLM 进行评估
    # 这里简化为关键词匹配
    points = key_points.lower().split(",")
    answer_lower = candidate_answer.lower()
    matched = sum(1 for p in points if p.strip() in answer_lower)
    score = min(10, max(1, int(matched / len(points) * 10) + 3))

    if score >= 8:
        comment = "回答完整，掌握扎实"
    elif score >= 6:
        comment = "基本正确，但缺少部分细节"
    else:
        comment = "理解不够深入，建议加强"

    return f"得分: {score}/10\n评语: {comment}"

@tool
def record_interview_note(note: str) -> str:
    """记录面试笔记，用于后续评估。

    Args:
        note: 面试笔记内容
    """
    # 实际项目中会存入数据库
    return f"已记录笔记: {note}"

@tool
def decide_next_step(average_score: float) -> str:
    """根据平均得分决定下一步行动。

    Args:
        average_score: 候选人的平均得分（1-10）
    """
    if average_score >= 7:
        return "建议进入下一轮面试，安排与 Tech Lead 的深度技术面"
    elif average_score >= 5:
        return "建议安排补充面试，重点考察薄弱环节"
    else:
        return "建议发送婉拒邮件，感谢候选人的时间"

# ========== 4. 动态 Prompt 模板 ==========
PROMPTS = {
    "opening": """你是一位友好专业的技术面试官。现在是面试开场阶段。

【候选人信息】
{candidate_info}

【岗位要求】
{job_requirements}

请先做简短的自我介绍，然后根据候选人的简历，找一个话题破冰，让候选人放松下来。
注意：不要一开始就问技术难题。""",

    "technical": """你是一位严谨的技术面试官。现在进入技术深挖阶段。

【候选人信息】
{candidate_info}

【已问问题和回答】
{interview_history}

【待考察技能】
{skills_to_test}

请根据候选人的技能背景，使用 search_question_bank 工具检索合适的面试题。
根据候选人的回答，使用 evaluate_answer 工具进行评估。
如果回答不够深入，可以追问细节。""",

    "closing": """你是一位专业的面试官。现在是面试总结阶段。

【面试记录】
{interview_history}

【得分统计】
{score_summary}

请：
1. 使用 decide_next_step 工具根据平均得分决定下一步
2. 给候选人一个简短的反馈（不透露具体分数）
3. 询问候选人是否有问题要问"""
}

# ========== 5. 上下文组装器 ==========
class ContextAssembler:
    """上下文工程的核心：动态组装上下文"""

    def __init__(self, session: InterviewSession):
        self.session = session

    def get_candidate_info(self) -> str:
        c = self.session.candidate
        return f"""
姓名: {c.name}
应聘岗位: {c.position}
技能: {', '.join(c.skills)}
简历摘要: {c.resume_text[:500]}...
"""

    def get_interview_history(self) -> str:
        if not self.session.questions_asked:
            return "暂无"

        history = []
        for i, qa in enumerate(self.session.questions_asked[-5:], 1):  # 只取最近5条
            history.append(f"{i}. Q: {qa.get('question', 'N/A')}")
            history.append(f"   A: {qa.get('answer', 'N/A')[:200]}...")
            if 'score' in qa:
                history.append(f"   得分: {qa['score']}/10")
        return "\n".join(history)

    def get_score_summary(self) -> str:
        if not self.session.scores:
            return "暂无得分记录"
        avg = sum(self.session.scores) / len(self.session.scores)
        return f"已评估 {len(self.session.scores)} 道题，平均得分: {avg:.1f}/10"

    def assemble(self, phase: str = None) -> str:
        """根据阶段动态组装 Prompt"""
        phase = phase or self.session.current_phase
        template = PROMPTS.get(phase, PROMPTS["technical"])

        return template.format(
            candidate_info=self.get_candidate_info(),
            job_requirements="5年以上 Go 开发经验，熟悉 K8s，有大规模分布式系统经验",
            interview_history=self.get_interview_history(),
            skills_to_test=", ".join(self.session.candidate.skills),
            score_summary=self.get_score_summary()
        )

# ========== 6. 主程序 ==========
def create_interview_agent():
    """创建面试官 Agent"""
    llm = ChatOpenAI(model="gpt-4-turbo", temperature=0.3)
    tools = [search_question_bank, evaluate_answer, record_interview_note, decide_next_step]

    prompt = ChatPromptTemplate.from_messages([
        ("system", "{system_prompt}"),
        MessagesPlaceholder(variable_name="chat_history"),
        ("human", "{input}"),
        MessagesPlaceholder(variable_name="agent_scratchpad"),
    ])

    agent = create_openai_tools_agent(llm, tools, prompt)
    return AgentExecutor(agent=agent, tools=tools, verbose=True)

def run_interview():
    """运行面试流程"""
    # 初始化会话
    session = InterviewSession(
        session_id="S-20260116-001",
        candidate=SAMPLE_CANDIDATE
    )

    # 创建上下文组装器
    assembler = ContextAssembler(session)

    # 创建 Agent
    agent = create_interview_agent()

    # 对话历史（短期记忆）
    chat_history = []

    print("=" * 60)
    print("AI 面试官已就绪，输入 'quit' 退出，输入 'next' 进入下一阶段")
    print("=" * 60)

    phases = ["opening", "technical", "closing"]
    phase_idx = 0

    while True:
        session.current_phase = phases[phase_idx]
        system_prompt = assembler.assemble()

        user_input = input("\n👤 候选人: ").strip()

        if user_input.lower() == "quit":
            print("\n面试结束，感谢参与！")
            break

        if user_input.lower() == "next":
            phase_idx = min(phase_idx + 1, len(phases) - 1)
            print(f"\n[系统] 进入 {phases[phase_idx]} 阶段")
            continue

        # 调用 Agent
        result = agent.invoke({
            "system_prompt": system_prompt,
            "chat_history": chat_history,
            "input": user_input
        })

        response = result["output"]
        print(f"\n🤖 面试官: {response}")

        # 更新对话历史
        chat_history.append(HumanMessage(content=user_input))
        chat_history.append(AIMessage(content=response))

        # 简单模拟记录问答
        session.questions_asked.append({
            "question": "从对话中提取",
            "answer": user_input
        })

if __name__ == "__main__":
    # 确保设置了 OPENAI_API_KEY
    if not os.getenv("OPENAI_API_KEY"):
        print("请设置 OPENAI_API_KEY 环境变量")
    else:
        run_interview()