将LangGraph集成至现有AI面试代理系统的技术指导请求

云原生阿岚

2026-6-21

问题：将LangGraph集成至候选人行为分析AI代理

我正在开发一款基于Python的AI代理应用，用于候选人交互与行为分析，核心功能代码如下：

import random
import string
import google.generativeai as genai
from traits import TRAITS  # Assuming TRAITS is defined in traits.py

# Function to save conversation history to a text file
def save_conversation_to_file(conversation):
    random_filename = ''.join(random.choices(string.ascii_letters + string.digits, k=8)) + '.txt'
    with open(random_filename, 'w', encoding='utf-8') as file:
        for entry in conversation:
            if len(entry) == 3:  # Scenario, Question, Response
                file.write(f"Scenario: {entry[0]}\nQuestion: {entry[1]}\nResponse: {entry[2]}\n\n")
            else:  # Question, Response
                file.write(f"Question: {entry[0]}\nResponse: {entry[1]}\n\n")

# Function to introduce the candidate and prompt for their name
def introduce_candidate():
    model = genai.GenerativeModel('gemini-pro')
    name_prompt = "Generate a prompt to ask the candidate's name, phrased naturally as if asked by a human. Please ask for the candidate's name in a friendly and formal manner without providing any additional introductions."
    name_response = model.generate_content(name_prompt)
    name_question = name_response.text.strip()
    candidate_name = input(f"\nGemini: {name_question} ").strip()
    print(f"\nGemini: Thank you, {candidate_name}. Let's proceed with the behavioral analysis questions.\n")
    return candidate_name

# Function to ask a question to the candidate
def ask_question(prompt):
    while True:
        response = input(f"\nGemini: {prompt}\n\nCandidate: ").strip()
        if response:
            return response
        else:
            print("Gemini: Please provide a valid response.")

# Main execution and interaction with candidates
candidate_name = introduce_candidate()
# Additional interactions and function calls as per your workflow

此前已完成以下工作：

使用Crew AI实现了候选人交互与对话历史保存的基础功能
尝试搭建LangGraph以增强对话流程与决策能力
基于Google GenerativeAI开发了候选人引导、问题提问与响应处理功能

现寻求以下帮助：

如何将LangGraph有效集成至现有代码架构的指导，或可简化集成的替代框架推荐
构建可处理任务型交互与行为分析的AI代理的最佳实践建议

解决方案

一、LangGraph集成步骤

1. 安装依赖

首先安装LangGraph及相关依赖：

pip install langgraph google-generativeai

2. 定义对话状态

LangGraph通过状态管理对话流程，先定义包含核心数据的状态类：

from langgraph.graph import StateGraph, END
from typing import TypedDict, List

class ConversationState(TypedDict):
    candidate_name: str
    conversation_history: List[tuple]
    current_step: str
    traits_analysis: dict

3. 封装原有功能为LangGraph节点

将现有函数封装为图的节点，每个节点负责单一任务：

def get_candidate_name(state: ConversationState) -> ConversationState:
    model = genai.GenerativeModel('gemini-pro')
    name_prompt = "Generate a prompt to ask the candidate's name, phrased naturally as if asked by a human. Please ask for the candidate's name in a friendly and formal manner without providing any additional introductions."
    name_response = model.generate_content(name_prompt)
    name_question = name_response.text.strip()
    candidate_name = input(f"\nGemini: {name_question} ").strip()
    print(f"\nGemini: Thank you, {candidate_name}. Let's proceed with the behavioral analysis questions.\n")
    state["candidate_name"] = candidate_name
    state["current_step"] = "ask_behavioral_questions"
    return state

def ask_behavioral_question(state: ConversationState) -> ConversationState:
    # 从TRAITS中选择对应行为问题，示例逻辑
    trait = random.choice(list(TRAITS.keys()))
    question = TRAITS[trait]["question"]
    response = ask_question(question)
    state["conversation_history"].append((question, response))
    
    # 调用Gemini分析响应，更新特质评分
    analysis_prompt = f"Analyze the candidate's response '{response}' against the trait '{trait}', give a score from 1-5 and brief reasoning."
    analysis = genai.GenerativeModel('gemini-pro').generate_content(analysis_prompt).text
    state["traits_analysis"][trait] = {"score": analysis.split("Score:")[1].split()[0], "reasoning": analysis}
    
    # 判断是否继续提问（示例逻辑：最多5个问题）
    if len(state["conversation_history"]) >= 5:
        state["current_step"] = "save_conversation"
    else:
        state["current_step"] = "ask_behavioral_questions"
    return state

def save_conversation(state: ConversationState) -> ConversationState:
    save_conversation_to_file(state["conversation_history"])
    print("\nGemini: Conversation history saved. Thank you for your time!")
    return state

4. 构建并运行LangGraph

定义节点间的路由逻辑，构建图并启动：

# 初始化图
graph_builder = StateGraph(ConversationState)

# 添加节点
graph_builder.add_node("get_name", get_candidate_name)
graph_builder.add_node("ask_questions", ask_behavioral_question)
graph_builder.add_node("save_conversation", save_conversation)

# 设置起始节点
graph_builder.set_entry_point("get_name")

# 添加路由逻辑
graph_builder.add_conditional_edges(
    "get_name",
    lambda state: state["current_step"],
    {
        "ask_behavioral_questions": "ask_questions"
    }
)

graph_builder.add_conditional_edges(
    "ask_questions",
    lambda state: state["current_step"],
    {
        "ask_behavioral_questions": "ask_questions",
        "save_conversation": "save_conversation"
    }
)

graph_builder.add_edge("save_conversation", END)

# 编译图并运行
graph = graph_builder.compile()
initial_state = {
    "candidate_name": "",
    "conversation_history": [],
    "current_step": "",
    "traits_analysis": {}
}
graph.invoke(initial_state)

5. 替换原有主流程

将原有的candidate_name = introduce_candidate()替换为上述LangGraph的运行代码，完成集成。

二、简化集成的替代框架推荐

如果LangGraph的状态路由对你来说过于复杂，可以考虑以下替代方案：

LangChain ConversationChain：适合线性对话流程扩展，可快速将现有提问逻辑包装为对话链，内置对话历史管理
CrewAI流程扩展：利用CrewAI的Task和Agent体系，将行为分析拆分为多个协作任务（如提问Agent、分析Agent、保存Agent），无需切换新框架
Rasa：适合构建复杂多轮对话，但学习曲线稍高，适合需要大规模部署的场景

三、任务型交互与行为分析AI代理最佳实践

模块化拆分功能：将引导、提问、分析、保存拆分为独立模块，便于维护和扩展，比如将特质分析逻辑单独封装为analyze_response函数
结构化状态管理：用明确的状态变量（如当前步骤、已完成问题数、特质评分）控制流程，避免逻辑混乱
Prompt工程优化：针对行为分析设计结构化输出prompt，要求模型返回固定格式的评分和理由，便于后续解析（例如："请返回JSON格式：{'score': 3, 'reasoning': '...'}"）
对话历史结构化存储：除了文本文件，可使用SQLite或JSON存储对话，便于后续的行为数据挖掘和分析
迭代式测试：针对不同候选人场景测试流程，调整路由逻辑和提问策略，确保流程流畅性
模型调用优化：缓存常见提问的生成结果，减少重复调用Gemini的成本，提高响应速度

内容的提问来源于stack exchange，提问作者user26335862