AGI｜一次关于LangGraph的探索之旅

Part1

前言

Part2

LangGraph

langchain
langchain_openai
tavily-python
langgraph
langchainhub
unstructured[docx]
pymilvus
docx2pdf
python-multipart

pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple/

def handle_review_file(file_path):
    print('处理文件切片向量化存储')

def check_requirements(collection_name,query,check_req):
    print('校验文本内容要求')

def check_doc_format(standard_file_path,review_file_path): 
    print('校验文本格式要求')
    
def chat_model(query,check_req,content):    
    print('LLM chat')

PLAN_PROMPT = """For the following task, make plans that can solve the problem step by step. For each plan, indicate \
which external tool together with tool input to retrieve evidence. You can store the evidence into a \
variable #E that can be called by later tools. (Plan, #E1, Plan, #E2, Plan, ...)

Tools can be one of the following:
(1) handle_file[input]: 处理文档的工具. 当你需要将文档处理文档的时候可以使用该工具.
(3) check_format[input]: 检查文档目录及格式的工具. 当你需要检查文档的格式、目录、大纲内容是否符合要求的时候可以使用该工具.
(4) check_requirement[input]: 检查文档内容是否符合要求. 当你需要检查文档某个检查点(check_point)内容是否符合标准要求的时候可以使用该工具.
(5) export_report[input]: 导出报告. 当你需要导出报告的时候可以使用该工具. 输入参数(报告内容).

For example,
Task: 请帮我检查一份文档,检查点包括：文档目录格式、入学要求、修学年限,最后将内容汇总后,导出一份完整的检查报告.
Plan: 处理文档. #E1 = handle_file['']
Plan: 检查文档目录格式. #E2 = check_format['']
Plan: 检查文档中入学要求是否符合要求. #E3 = check_requirement[入学要求,#E1]
Plan: 检查文档中修学年限是否符合要求. #E4 = check_requirement[修学年限,#E1]
Plan: 导出检查报告. #E5 = export_report[#E2,#E3,#E4]

Begin! 
Describe your plans with rich details. Each Plan should be followed by only one #E.

Task: {task}"""

llm = AzureChatOpenAI(
    azure_endpoint="https://xxxxxx.openai.azure.com/",
    azure_deployment="gpt-35-turbo",
    openai_api_version="2023-07-01-preview",
    openai_api_key="xxxxxxxxx",
    temperature=0
)
def get_plan(state: State):
    regex_pattern = r"\s*(#E\d+)\s*=\s*(\w+)\s*\[([^\]]+)\]"
    prompt_template = ChatPromptTemplate.from_messages([("user", PLAN_PROMPT)])
    planner = prompt_template | llm
    result = planner.invoke({"task": state["task"]['task_desc']})
    matches = re.findall(regex_pattern, result.content)
    return {"steps": matches, "plan_string": result.content}

class Task(TypedDict):
    task_desc: str
    review_file_path: str
    standard_file_path: str

class State(TypedDict):
    task: Task
    plan_string: str
    steps: List
    results: dict
    result: str

def _get_current_task(state: State):
    if state["results"] is None:
        return 1
    if len(state["results"]) == len(state["steps"]):
        return None
    else:
        return len(state["results"]) + 1

def tool_execution(state: State):
    _step = _get_current_task(state)
    step_name, tool, tool_input = state["steps"][_step - 1]
    _results = state["results"] or {}
    tool_input_list = [_results.get(k.strip(),k) for k in tool_input.split(",")]
    if tool == "handle_file":
        result = handle_review_file(state['task']['review_file_path'])
    elif tool == "check_format":
        result = check_doc_format(state['task']['standard_file_path'], state['task']['review_file_path'])
    elif tool == "check_requirement":
        result = check_requirements(tool_input_list[0], tool_input_list[1],state['task']['standard_file_path'])
    elif tool == "export_report":
        result = export_report(tool_input_list)
    else:
        raise ValueError
    _results[step_name] = str(result)
    return {"results": _results}

SOLVE_PROMPT = """Solve the following task or problem. To solve the problem, we have made step-by-step Plan and \
retrieved corresponding Evidence to each Plan. Use them with caution since long evidence might \
contain irrelevant information.

{plan}

Now solve the question or task according to provided Evidence above. Respond with the answer
directly with no extra words.

Task: {task}
Response:"""

def solve(state: State):
    plan = ""
    for step_name, tool, tool_input in state["steps"]:
        _results = state["results"] or {}
        for k, v in _results.items():
            tool_input = tool_input.replace(k, v)
            step_name = step_name.replace(k, v)
        plan += f"Plan: \n{step_name} = {tool}[{tool_input}]"
    prompt = SOLVE_PROMPT.format(plan=plan, task=state["task"])
    result = llm.invoke(prompt)
    return {"result": result.content}

def _route(state):
    _step = _get_current_task(state)
    if _step is None:
        
        return "solve"
    else:
        
        return "tool"

graph = StateGraph(State)
graph.add_node("plan", get_plan)
graph.add_node("tool", tool_execution)
graph.add_node("solve", solve)
graph.add_edge("plan", "tool")
graph.add_edge("solve", END)
graph.add_conditional_edges("tool", _route)
graph.set_entry_point("plan")
agent_app = graph.compile()

task = Task(
    task_desc="请帮我检查一份文档,检查点包括：文档目录格式、入学要求、修学年限、课程说明和要求、毕业要求,最后将内容汇总后,导出一份完整的检查报告.",
    standard_file_path='./data/标准要求文档.docx',
    review_file_path='./data/待审核文档.docx'
)
for s in agent_app.stream({"task": task}, config={"recursion_limit": 100}):
    print(s)
    print("-----------------------------------------------------------")

{'plan': {'steps': [('#E1', 'handle_file', "''"), ('#E2', 'check_format', '#E1'), ('#E3', 'check_requirement', '入学要求,#E1'), ('#E4', 'check_requirement', '修学年限,#E1'), ('#E5', 'check_requirement', '课程说明和要求,#E1'), ('#E6', 'check_requirement', '毕业要求,#E1'), ('#E7', 'export_report', '#E2,#E3,#E4,#E5,#E6')], 'plan_string': "Plan: 处理文档. #E1 = handle_file['']\nPlan: 检查文档目录格式. #E2 = check_format[#E1]\nPlan: 检查文档中入学要求是否符合要求. #E3 = check_requirement[入学要求,#E1]\nPlan: 检查文档中修学年限是否符合要求. #E4 = check_requirement[修学年限,#E1]\nPlan: 检查文档中课程说明和要求是否符合要求. #E5 = check_requirement[课程说明和要求,#E1]\nPlan: 检查文档中毕业要求是否符合要求. #E6 = check_requirement[毕业要求,#E1]\nPlan: 导出检查报告. #E7 = export_report[#E2,#E3,#E4,#E5,#E6]"}}
ps: 由于我只是模拟整个Agent执行的流程，里面工具输出的实际效果我并没有去关心。