随着强大 AI 模型（如 GPT-5、Gemini 2.5 Pro）的崛起，能充分利用这些模型的 agentic 框架也越来越多。这类框架通过抽象掉诸多繁琐环节，使与 AI 模型协作更简单，例如处理 tool-calling（工具调用）、agentic state handling（智能体状态管理）、human-in-the-loop（人类参与闭环）等。
因此，本文将深入介绍可用的 agentic AI 框架之一：LangGraph。我会用它来开发一个简单的 agentic 应用，通过若干步骤展示 agentic AI 包的优势。文末也会讨论使用 LangGraph 及其他类似 agentic 框架的优缺点。
""" Make a document handler workflow where a user can create a new document to the database (currently just a dictionary) delete a document from the database ask a question about a document """
from typing_extensions import TypedDict, Literal from langgraph.checkpoint.memory import InMemorySaver from langgraph.graph import StateGraph, START, END from langgraph.types import Command, interrupt from langchain_aws import ChatBedrockConverse from langchain_core.messages import HumanMessage, SystemMessage from pydantic import BaseModel, Field from IPython.display import display, Image from dotenv import load_dotenv import os load_dotenv()
aws_access_key_id = os.getenv("AWS_ACCESS_KEY_ID") or "" aws_secret_access_key = os.getenv("AWS_SECRET_ACCESS_KEY") or ""
llm = ChatBedrockConverse( model_id="us.anthropic.claude-3-5-haiku-20241022-v1:0", # this is the model id (added us. before id in platform) region_name="us-east-1", aws_access_key_id=aws_access_key_id, aws_secret_access_key=aws_secret_access_key,
)
document_database: dict[str, str] = {} # a dictionary with key: filename, value: text in document
# Schema for structured output to use as routing logic classRoute(BaseModel): step: Literal["add_document", "delete_document", "ask_document"] = Field( description="The next step in the routing process" ) # Augment the LLM with schema for structured output router = llm.with_structured_output(Route) defllm_call_router(state: State): """Route the user input to the appropriate node""" # Run the augmented LLM with structured output to serve as routing logic decision = router.invoke( [ SystemMessage( content="""Route the user input to one of the following 3 intents: - 'add_document' - 'delete_document' - 'ask_document' You only need to return the intent, not any other text. """ ), HumanMessage(content=state["input"]), ] ) return {"decision": decision.step} # Conditional edge function to route to the appropriate node defroute_decision(state: State): # Return the node name you want to visit next if state["decision"] == "add_document": return"add_document_to_database_tool" elif state["decision"] == "delete_document": return"delete_document_from_database_tool" elif state["decision"] == "ask_document": return"ask_document_tool"
# Nodes defadd_document_to_database_tool(state: State): """Add a document to the database. Given user query, extract the filename and content for the document. If not provided, will not add the document to the database."""
user_query = state["input"] # extract filename and content from user query filename_prompt = f"Given the following user query, extract the filename for the document: {user_query}. Only return the filename, not any other text." output = llm.invoke(filename_prompt) filename = output.content content_prompt = f"Given the following user query, extract the content for the document: {user_query}. Only return the content, not any other text." output = llm.invoke(content_prompt) content = output.content # add document to database document_database[filename] = content return {"output": f"Document {filename} added to database"}
defdelete_document_from_database_tool(state: State): """Delete a document from the database. Given user query, extract the filename of the document to delete. If not provided, will not delete the document from the database.""" user_query = state["input"] # extract filename from user query filename_prompt = f"Given the following user query, extract the filename of the document to delete: {user_query}. Only return the filename, not any other text." output = llm.invoke(filename_prompt) filename = output.content # delete document from database if it exsits, if not retunr info about failure if filename notin document_database: return {"output": f"Document {filename} not found in database"} document_database.pop(filename) return {"output": f"Document {filename} deleted from database"}
defask_document_tool(state: State): """Ask a question about a document. Given user query, extract the filename and question for the document. If not provided, will not ask the question about the document.""" user_query = state["input"] # extract filename and question from user query filename_prompt = f"Given the following user query, extract the filename of the document to ask a question about: {user_query}. Only return the filename, not any other text." output = llm.invoke(filename_prompt) filename = output.content question_prompt = f"Given the following user query, extract the question to ask about the document: {user_query}. Only return the question, not any other text." output = llm.invoke(question_prompt) question = output.content # ask question about document if filename notin document_database: return {"output": f"Document {filename} not found in database"} result = llm.invoke(f"Document: {document_database[filename]}\n\nQuestion: {question}") return {"output": f"Document query result: {result.content}"}
# Show the workflow display(Image(router_workflow.get_graph().draw_mermaid_png()))
最后的显示函数会渲染出如下图所示的图结构：
现在你可以按不同意图来试跑这个 workflow 了。
添加文档：
user_input = "Add the document 'test.txt' with content 'This is a test document' to the database" state = router_workflow.invoke({"input": user_input}, config) print(state["output"]
# -> Document test.txt added to database
搜索文档：
user_input = "Give me a summary of the document 'test.txt'" state = router_workflow.invoke({"input": user_input}, config) print(state["output"])
# -> A brief, generic test document with a simple descriptive sentence.
删除文档：
user_input = "Delete the document 'test.txt' from the database" state = router_workflow.invoke({"input": user_input}, config) print(state["output"])
粤ICP备14082021号
免费POC，
零成本试错
