#(一) 框架设计与工具节点实现

import pandas as pd
from langchain_core.documents import Document

# 读取Excel文件
df = pd.read_excel('标准学校清单.xlsx', header=None)

# 将数据转换为Document对象列表
documents = [
    Document(
        page_content=row.iloc[0],
        metadata={"source": "school"},
    )
    for index, row in df.iterrows()
]

from typing import List, Optional
from typing_extensions import TypedDict

class GraphState(TypedDict):
    """
    Represents the state for managing school name standardization queries.

    Attributes:
        user_query: The original query from the user regarding a school name.
        routing_action: The action determined by the router, either 'direct_retrieval' or 'perform_search'.
        retrieved_records: Optional; list of records if 'direct_retrieval' was chosen.
        standardized_name: Optional; the standardized school name if resolved.
        search_results: Optional; results from the web search if 'perform_search' was chosen.
        match_status: Optional; indicates if the standardized or retrieved name matches the query ('match' or 'no_match').
        retrieved_school_name: Optional; the school name retrieved from the vector store.
        search_status: Optional[str]; the status of the search analysis ('known' or 'unknown').
    """

    user_query: str
    routing_action: Optional[str]
    retrieved_records: Optional[List[str]]
    standardized_name: Optional[str]
    search_results: Optional[List[str]]
    match_status: Optional[str]
    retrieved_school_name: Optional[str]
    search_status: Optional[str]

from typing import Dict

def direct_retrieval(state: Dict) -> Dict:
    """
    Retrieve the school name from the vector store based on the query and update the state.

    Args:
        state (Dict): The current state of the process

    Returns:
        Dict: The updated state
    """
    # 使用检索器从向量数据库中获取结果
    retrieved_school_name = retriever.invoke(state["standardized_name"])[0].page_content

    # 更新状态中的retrieved_school_name属性
    state["retrieved_school_name"] = retrieved_school_name

    return state

1from pprint import pprint
2
3initial_state = {
4    'user_query': '剑桥',
5    'routing_action': 'direct_retrieval',
6    'standardized_name': ['剑桥大学', 'University of Cambridge']
7}
8
9updated_state = direct_retrieval(initial_state)
10pprint(updated_state)

def perform_web_search(state: Dict) -> Dict:
    """
    Perform a web search using TavilySearch. If it fails, switch to DuckDuckGoSearch.
    Updates the state with the search results.

    Args:
        state (Dict): The current state of the process

    Returns:
        Dict: The updated state with search results
    """
    query = state["user_query"]

    print("---PERFORM WEB SEARCH---")

    try:
        results = tavily_search_tool.invoke(query)
        if isinstance(results, str) and 'HTTPError' in results:
            raise ValueError("TavilySearch HTTPError")
    except Exception as e:
        results = duckduckgo_search_tool.invoke(query)

    state["search_results"] = results
    return state

1initial_state = {
2    'user_query': '光华商学院',
3    'routing_action': 'direct_retrieval'
4}
5
6updated_state = perform_web_search(initial_state)
7pprint(updated_state)