# 安装最新版本的必要软件包!pip install langchain langchain-openai langgraph python-dotenv

OPENAI_API_KEY=您的_api_密钥

# 从 .env 文件加载环境变量import osfrom dotenv import load_dotenvload_dotenv()# 从 .env 文件加载 API 密钥# 验证 API 密钥是否加载if os.getenv("OPENAI_API_KEY")isNone:print("警告：未在环境变量中找到 OPENAI_API_KEY。")else:print("已在环境变量中找到 OPENAI_API_KEY。")

# 导入必要的库from typing importDict,List,TypedDict,Annotated,Sequence,Any,Optional,Literal,Unionimport json# 用于 langchain 和 langgraph 的现代导入from langchain_core.messages importHumanMessage,AIMessage,SystemMessage,BaseMessagefrom langchain_openai importChatOpenAIfrom langgraph.graph importStateGraph,END

def ask_llm(prompt, model="gpt-4o", temperature=0.7):"""一个简单的函数，用于从 LLM 获取响应。"""# 创建 ChatOpenAI 实例    llm =ChatOpenAI(model=model, temperature=temperature)# 使用提示创建消息    messages =[HumanMessage(content=prompt)]# 从 LLM 获取响应    response = llm.invoke(messages)# 返回响应的内容return response.content
# 测试我们的函数response = ask_llm("LangGraph 是什么，它与 LangChain 有何关系？")print(response)

# 定义状态类型classSimpleState(TypedDict):    messages:List[BaseMessage]# 对话历史next: str  # 图中的下一步
# 定义一个简单的代理节点def simple_agent(state:SimpleState)->SimpleState:"""一个简单的代理，响应最后一条消息。"""    messages = state["messages"]    llm =ChatOpenAI(model="gpt-4o", temperature=0.7)    response = llm.invoke(messages)return{"messages": messages +[response],"next":"output"}
# 定义输出节点def output(state:SimpleState):"""返回最终状态，标记工作流结束。"""return{"messages": state["messages"],"next":END}
# 构建图def build_simple_graph():"""构建一个简单的 LangGraph 工作流。"""# 创建一个新图    workflow =StateGraph(SimpleState)# 添加节点    workflow.add_node("agent", simple_agent)    workflow.add_node("output", output)# 添加边    workflow.add_edge("agent","output")# 设置入口点    workflow.set_entry_point("agent")# 编译图return workflow.compile()

# 创建图simple_graph = build_simple_graph()# 使用测试消息初始化状态initial_state ={"messages":[HumanMessage(content="用简单术语解释什么是多代理系统。")],"next":""}# 运行图result = simple_graph.invoke(initial_state)# 打印对话for message in result["messages"]:if isinstance(message,HumanMessage):print(f"用户: {message.content}")elif isinstance(message,AIMessage):print(f"AI: {message.content}")

from typing importDict,List,TypedDict,Any,Optionalimport jsonimport osfrom dotenv import load_dotenv# 用于 langchain 和 langgraph 的现代导入from langchain_core.messages importHumanMessage,AIMessage,SystemMessage,BaseMessagefrom langchain_openai importChatOpenAIfrom langgraph.graph importStateGraph,END# 加载环境变量load_dotenv()

RESEARCHER_SYSTEM_PROMPT ="""您是一个熟练的研究者代理，任务是收集有关给定主题的全面信息。您的职责包括：1.分析研究查询，了解所需信息2.进行彻底研究，收集相关事实、数据和观点3.以清晰、结构化的格式组织信息4.确保发现的准确性和客观性5.引用来源或指出可能需要验证的信息6.识别信息中的潜在空白以清晰的结构化格式呈现您的发现，适当使用章节和项目符号。您的目标是提供全面、准确且有用的信息，充分回答研究查询。"""

def create_researcher_agent(model="gpt-4o", temperature=0.7):"""使用指定的 LLM 创建研究者代理。"""# 初始化模型    llm =ChatOpenAI(model=model, temperature=temperature)def researcher_function(messages):"""处理消息并返回研究者代理响应的函数。"""# 如果消息为空或第一条消息不是系统提示，则添加系统提示ifnot messages ornot isinstance(messages[0],SystemMessage)or messages[0].content != RESEARCHER_SYSTEM_PROMPT:            messages =[SystemMessage(content=RESEARCHER_SYSTEM_PROMPT)]+(messages if isinstance(messages, list)else[])# 从 LLM 获取响应        response = llm.invoke(messages)return responsereturn researcher_function

# 为研究工作流定义状态类型classResearchState(TypedDict):"""我们研究工作流状态的类型定义。"""    messages:List[BaseMessage]# 对话历史    query: str  # 研究查询    research:Optional[str]# 研究发现next:Optional[str]# 图中的下一步

def researcher_node(state:ResearchState)->ResearchState:"""在图中执行研究查询的节点。"""# 从状态中获取查询    query = state["query"]# 为研究者创建特定消息    research_message =HumanMessage(content=f"请彻底研究以下主题：{query}")# 获取研究者代理    researcher = create_researcher_agent()# 从研究者代理获取响应    response = researcher([research_message])# 使用研究发现更新状态    new_messages = state["messages"]+[research_message, response]# 返回更新后的状态return{**state,"messages": new_messages,"research": response.content,"next":"output"# 在多代理系统中，这将转到下一个代理}

def build_research_graph():"""使用 LangGraph 构建一个简单的研究工作流。"""# 使用我们的状态类型创建一个新图    workflow =StateGraph(ResearchState)# 添加节点    workflow.add_node("researcher", researcher_node)    workflow.add_node("output", output_node)# 添加边    workflow.add_edge("researcher","output")# 设置入口点    workflow.set_entry_point("researcher")# 编译图return workflow.compile()

ENHANCED_RESEARCHER_PROMPT ="""您是一个熟练的研究者代理，任务是收集有关给定主题的全面信息。您的职责包括：1.分析研究查询，了解所需信息2.进行彻底研究，收集相关事实、数据和观点3.以清晰、结构化的格式组织信息4.确保发现的准确性和客观性5.引用来源或指出可能需要验证的信息6.识别信息中的潜在空白以以下结构化格式呈现您的发现：SUMMARY：发现的简要概述（2-3句）KEY POINTS：-重点1-重点2-重点3DETAILED FINDINGS：1.[主题领域1]-详细说明和解释-支持证据-适用的不同观点2.[主题领域2]-详细说明和解释-支持证据-适用的不同观点GAPS AND LIMITATIONS：-识别信息可能不完整的领域-指出任何矛盾或争议领域-建议可能需要的额外研究您的目标是提供全面、准确且有用的信息，充分回答研究查询。"""

classEnhancedResearchState(TypedDict):"""我们研究工作流状态的增强类型定义。"""    messages:List[BaseMessage]    query: str    structured_research:Optional[str]next:Optional[str]

CRITIC_SYSTEM_PROMPT ="""您是批评者代理，属于协作研究助手系统的一部分。您的角色是评估和挑战研究者代理提供的信息，以确保其准确性、完整性和客观性。您的职责包括：1.分析研究发现的准确性、完整性和潜在偏见2.识别信息中的空白或逻辑不一致3.提出可能被忽略的重要问题4.建议改进或替代观点5.确保最终信息平衡且全面您的批评应具有建设性，目标不是否定研究者的工作，而是增强其质量。以清晰、有组织的方式格式化您的反馈，突出需要注意的具体点。请记住，您的最终目标是确保最终研究输出的质量尽可能高。"""

classCollaborativeResearchState(TypedDict):"""协作研究助手的状态类型。"""    messages:List[BaseMessage]# 对话历史next:Optional[str]# 图中的下一步

def build_collaborative_research_assistant():"""构建一个具有研究者和批评者代理的协作研究助手。"""# 使用我们的状态类型创建一个新图    workflow =StateGraph(CollaborativeResearchState)# 添加节点    workflow.add_node("researcher", researcher_node)    workflow.add_node("critic", critic_node)    workflow.add_node("output", output_node)# 添加边    workflow.add_edge("researcher","critic")    workflow.add_edge("critic","output")# 设置入口点    workflow.set_entry_point("researcher")# 编译图return workflow.compile()

classCriticEvaluation(BaseModel):"""批评者评估的结构化格式。"""    quality_score:int=Field(description="总体质量评分，1-10")    strengths:List[str]=Field(description="研究的关键优势")    areas_for_improvement:List[str]=Field(description="需要改进的领域")    missing_information:List[str]=Field(description="未包含的重要信息")    bias_assessment: str =Field(description="研究中潜在偏见的评估")    additional_questions:List[str]=Field(description="应解决的问题")

WRITER_SYSTEM_PROMPT ="""您是编写者代理，属于协作研究助手系统的一部分。您的角色是合成研究者代理提供的信息和批评者代理的反馈，生成连贯、精心撰写的响应。您的职责包括：1.分析研究者提供的信息和批评者的反馈2.以逻辑、易于理解的结构组织信息3.以清晰、引人入胜的写作风格呈现信息4.平衡不同观点并确保客观性5.创建全面、准确且精心撰写的最终响应以清晰、有组织的方式格式化您的响应，适当使用标题、段落和项目符号。使用简单语言解释复杂概念，并在有帮助时提供示例。请记住，您的目标是创建有效向用户传达信息的最终响应。"""

classCollaborativeResearchState(TypedDict):"""协作研究助手的状态类型。"""    messages:List[BaseMessage]# 对话历史next:Optional[str]# 图中的下一步

def writer_node(state:CollaborativeResearchState)->CollaborativeResearchState:"""编写者代理的节点函数。"""# 从状态中提取消息    messages = state["messages"]# 使用系统提示创建编写者消息    writer_messages =[SystemMessage(content=WRITER_SYSTEM_PROMPT)]+ messages# 初始化 LLM，平衡创造力和准确性    llm =ChatOpenAI(model="gpt-4o", temperature=0.6)# 获取编写者的响应    response = llm.invoke(writer_messages)# 返回更新后的状态return{"messages": messages +[response],"next":"output"}

def build_complete_research_assistant():"""构建一个具有研究者、批评者和编写者代理的完整研究助手。"""# 使用我们的状态类型创建一个新图    workflow =StateGraph(CollaborativeResearchState)# 添加节点    workflow.add_node("researcher", researcher_node)    workflow.add_node("critic", critic_node)    workflow.add_node("writer", writer_node)    workflow.add_node("output", output_node)# 添加边    workflow.add_edge("researcher","critic")    workflow.add_edge("critic","writer")    workflow.add_edge("writer","output")# 设置入口点    workflow.set_entry_point("researcher")# 编译图return workflow.compile()

classEnhancedResearchState(TypedDict):"""研究过程元数据的增强状态类型。"""    messages:List[BaseMessage]# 对话历史    metadata:Dict[str,Any]# 每个步骤的元数据next:Optional[str]# 图中的下一步

def coordinator_node(state:ResearchState)->ResearchState:"""决定工作流路径的协调者节点。"""# 从状态中提取消息    messages = state["messages"]# 使用系统提示创建协调者消息    coordinator_messages =[SystemMessage(content=COORDINATOR_SYSTEM_PROMPT)]+ messages# 初始化 LLM，使用较低温度以确保一致的决策    llm =ChatOpenAI(model="gpt-4o", temperature=0.2)# 获取协调者的响应    response = llm.invoke(coordinator_messages)# 解析 JSON 响应以确定下一步try:        decision = json.loads(response.content)        next_step = decision.get("next","researcher")# 如果未指定，默认为研究者exceptException:# 如果解析 JSON 出错，默认为研究者        next_step ="researcher"# 返回更新后的状态return{"messages": messages,"next": next_step}

# 从协调者添加条件边workflow.add_conditional_edges("coordinator",lambda state: state["next"],{"researcher":"researcher","done":"output"})

def build_dynamic_research_assistant():"""构建一个动态研究助手，协调者代理管理工作流。"""# 创建一个新图    workflow =Graph()
# 添加节点    workflow.add_node("coordinator", coordinator_agent)    workflow.add_node("researcher", researcher_agent)    workflow.add_node("critic", critic_agent)    workflow.add_node("writer", writer_agent)    workflow.add_node("output", output)
# 从协调者添加条件边    workflow.add_conditional_edges("coordinator",lambda state: state["next"],{"researcher":"researcher","done":"output"})
# 添加其余边    workflow.add_edge("researcher","critic")    workflow.add_edge("critic","writer")    workflow.add_edge("writer","coordinator")    workflow.add_edge("output",END)
# 设置入口点    workflow.set_entry_point("coordinator")
# 编译图return workflow.compile()

{"reasoning":"这是一个简单的实事问题，询问法国的首都。答案广为人知，无需深入研究、批评分析或专门编写。","next":"done"}

{"reasoning":"此查询询问有关量子计算及其与网络安全关系的复杂信息。需要收集详细信息、评估不同观点并合成全面的响应。","next":"researcher"}

{"reasoning":"查询已通过多角度彻底研究、批评并合成为全面的响应，解决了量子计算对网络安全的意义。","next":"done"}