web前端学习LangGraph - AI创想

import uuid
from typing import TypedDict,Optional, Dict, Any, List
from langchain_core.messages import HumanMessage, BaseMessage, AIMessage, SystemMessage
from langchain_core.messages.utils import count_tokens_approximately
from langchain_core.prompts import PromptTemplate
from langmem.short_term import SummarizationNode
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph,END,START
from langchain_deepseek import ChatDeepSeek
from langgraph.prebuilt import create_react_agent
import dotenv
import asyncio
import os
import re
from loguru import logger
from pydantic import BaseModel, Field
from langgraph.prebuilt.chat_agent_executor import AgentState
dotenv.load_dotenv()
if __name__ == "__main__":
llm = ChatDeepSeek(
model = 'qwen-plus',
temperature=0.1,
api_key=os.getenv("DASHSCOPE_API_KEY_1"),
api_base="https://dashscope.aliyuncs.com/compatible-mode/v1",
)
# 1. 初始化检查点
memory_saver = InMemorySaver()
class CustomState(AgentState):
context: dict[str,Any]
# 3. 总结节点（如果需要保留）
summarization_node = SummarizationNode(
token_counter=count_tokens_approximately,
model=llm,
max_tokens=384,
max_summary_tokens=128,
output_messages_key="llm_input_messages",
)
# 4. 创建智能体
agent = create_react_agent(
model=llm,
tools=[], # 如需查询实时信息，需添加工具（如搜索工具）
checkpointer=memory_saver,
state_schema=CustomState,
# debug=True,
pre_model_hook=summarization_node
)
# 5. 调用智能体（使用正确的消息格式）
config = {"configurable": {"thread_id": "user_session_1"}}
first_input_text = "你好我叫小明，你是谁"
cs = agent.invoke({
"messages": [HumanMessage(content=first_input_text,id=str(uuid.uuid4()))], # 符合BaseMessage类型
"remaining_steps": 5, # 覆盖默认的3步
}, config)
print(cs)
cs1 = agent.invoke({
"messages": [
# HumanMessage(content=first_input_text, id=str(uuid.uuid4())),
HumanMessage(content="我叫什么名字", id=str(uuid.uuid4()))
], # 向同一线程注入上一轮用户自述 + 当前追问
"remaining_steps": 5, # 覆盖默认的3步
}, config)
print(cs1['messages'][-1])

复制代码

import asyncio
import json
import os
import pickle
from collections.abc import Sequence
from typing import TypedDict, Optional, Dict, Any, List, Union
from langgraph.checkpoint.base import BaseCheckpointSaver, Checkpoint, ChannelVersions, CheckpointTuple
from loguru import logger
from langchain_core.messages.utils import count_tokens_approximately
from langchain_core.prompts import ChatPromptTemplate
from langchain_deepseek import ChatDeepSeek
from langmem.short_term import SummarizationNode
from langgraph.prebuilt import create_react_agent
import dotenv
from langgraph.prebuilt.chat_agent_executor import AgentState
from pydantic import BaseModel, Field
from app.db.database import get_redis_session_task, get_redis_session
dotenv.load_dotenv()
# -------------------------- 数据模型定义（Pydantic）--------------------------
class StartModel(BaseModel):
id:int = Field(description="数据id，用于记忆存储的config")
code:int = Field(description="用户工号，用于记忆存储的config")
prompt:str = Field(description="提示词")
enter_str:Optional[str] = Field(description="输入问题")
is_memory:Optional[bool] = Field(description="是否需要记忆",default=True)
# -------------------------- 核心模块--------------------------
class RedisCheckpointSaver(BaseCheckpointSaver):
"""基于Redis的检查点保存器"""
def __init__(self):
pass
def get_tuple(self, config: dict) -> Optional[Checkpoint]:
"""同步获取检查点"""
try:
# 在同步环境中调用异步方法
import asyncio
# 尝试获取现有的事件循环
try:
loop = asyncio.get_running_loop()
# 如果事件循环已经在运行，我们需要在另一个线程中运行协程
import concurrent.futures
with concurrent.futures.ThreadPoolExecutor() as executor:
future = executor.submit(self._run_async_in_thread, self.aget_tuple(config))
return future.result(timeout=10.0)
except RuntimeError:
# 没有运行中的事件循环
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
result = loop.run_until_complete(self.aget_tuple(config))
return result
except Exception as e:
logger.error(f"Failed to get checkpoint tuple: {str(e)}")
return None
async def aget_tuple(self, config: dict) -> Optional[Checkpoint]:
"""异步获取检查点"""
redis_generator = None
try:
# 手动处理异步生成器
redis_generator = get_redis_session()
redis = await redis_generator.__anext__() # 获取生成器的第一个值
thread_id = config['configurable']['thread_id']
key = f"checkpoint:{thread_id}"
data = await redis.get(key)
if data:
# 确保数据能正确反序列化
try:
result = pickle.loads(data)
# 确保返回的是CheckpointTuple对象
if isinstance(result, CheckpointTuple):
return result
elif isinstance(result, dict):
# 检查是否包含必要的字段
if 'checkpoint' in result:
# 确保metadata包含必要的字段
metadata = result.get('metadata', {})
if 'step' not in metadata:
metadata['step'] = 0 # 默认步骤为0
if 'source' not in metadata:
metadata['source'] = 'redis' # 默认来源
return CheckpointTuple(
config=config,
checkpoint=result['checkpoint'],
metadata=metadata
)
else:
# 如果是checkpoint对象，创建CheckpointTuple
return CheckpointTuple(
config=config,
checkpoint=result,
metadata={
'step': 0,
'source': 'redis'
}
)
except Exception as e:
logger.error(f"Failed to deserialize checkpoint: {str(e)}")
# 如果反序列化失败，删除损坏的数据
await redis.delete(key)
return None
except RuntimeError as e:
if "Event loop is closed" in str(e):
logger.warning("Event loop is closed, cannot get checkpoint from Redis")
return None
else:
logger.error(f"Failed to get checkpoint tuple from Redis: {str(e)}")
return None
except Exception as e:
logger.error(f"Failed to get checkpoint tuple from Redis: {str(e)}")
return None
finally:
# 确保正确清理异步生成器
if redis_generator:
try:
await redis_generator.aclose()
except Exception as e:
logger.error(f"Failed to close redis generator: {str(e)}")
def put(self, config: dict, checkpoint: Checkpoint, metadata: dict,new_versions:ChannelVersions) -> None:
"""同步保存检查点"""
try:
# 在同步环境中调用异步方法
import asyncio
# 尝试获取现有的事件循环
try:
loop = asyncio.get_running_loop()
# 如果事件循环已经在运行，直接等待协程
task = self.aput(config, checkpoint, metadata)
# 在已运行的循环中创建任务并等待结果
future = asyncio.run_coroutine_threadsafe(task, loop)
future.result()
except RuntimeError:
# 没有运行中的事件循环
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.run_until_complete(self.aput(config, checkpoint, metadata))
except Exception as e:
logger.error(f"Failed to put checkpoint: {str(e)}")
async def aput(self, config: dict, checkpoint: Checkpoint, metadata: dict) -> None:
"""异步保存检查点"""
try:
# redis = self.redis
async for redis in get_redis_session():
thread_id = config['configurable']['thread_id']
key = f"checkpoint:{thread_id}"
# 确保数据能正确序列化
try:
data = pickle.dumps(checkpoint)
await redis.set(key, data)
# 设置过期时间，例如1小时
await redis.expire(key, 3600)
except Exception as e:
logger.error(f"Failed to serialize checkpoint: {str(e)}")
except Exception as e:
logger.error(f"Failed to save checkpoint to Redis: {str(e)}")
def put_writes(self, config: dict, writes: Sequence[tuple[str, Any]], task_id: str,task_path:str) -> None:
"""保存写入操作"""
try:
# 在同步环境中调用异步方法
import asyncio
# 尝试获取现有的事件循环
try:
loop = asyncio.get_running_loop()
# 如果事件循环已经在运行，直接等待协程
task = self.aput_writes(config, writes, task_id)
# 在已运行的循环中创建任务并等待结果
future = asyncio.run_coroutine_threadsafe(task, loop)
future.result()
except RuntimeError:
# 没有运行中的事件循环
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.run_until_complete(self.aput_writes(config, writes, task_id))
except Exception as e:
logger.error(f"Failed to put writes: {str(e)}")
async def aput_writes(self, config: dict, writes: Sequence[tuple[str, Any]], task_id: str) -> None:
"""异步保存写入操作"""
try:
# redis = self.redis
async for redis in get_redis_session():
thread_id = config['configurable']['thread_id']
# 为写入操作创建一个唯一的键
key = f"checkpoint_writes:{thread_id}:{task_id}"
# 确保数据能正确序列化
try:
data = pickle.dumps(writes)
await redis.set(key, data)
# 设置过期时间，例如1小时
await redis.expire(key, 3600)
except Exception as e:
logger.error(f"Failed to serialize writes: {str(e)}")
except Exception as e:
logger.error(f"Failed to save writes to Redis: {str(e)}")
def _run_async_in_thread(self, coro):
"""在线程中运行异步协程"""
import asyncio
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
return loop.run_until_complete(coro)
finally:
loop.close()
def create_agent(params:StartModel):
"""
创建智能体
prompt:提示词
is_memory：是否需要记忆
"""
llm = ChatDeepSeek(
model='qwen3-max',
temperature=0.1,
api_key=os.getenv("DASHSCOPE_API_KEY_1"),
api_base=os.getenv("TONGYI_API_BASE"),
)
# 1. 初始化检查点
# memory_saver = InMemorySaver()
redis_saver = RedisCheckpointSaver() if params.is_memory else None
class CustomState(AgentState):
context: dict[str, Any]
# 3. 总结节点（如果需要保留）
summarization_node = SummarizationNode(
token_counter=count_tokens_approximately,
model=llm,
max_tokens=384,
max_summary_tokens=128,
output_messages_key="llm_input_messages",
)
# 4. 创建智能体
agent = create_react_agent(
model=llm,
tools=[], # 如需查询实时信息，需添加工具（如搜索工具）
checkpointer= redis_saver ,
state_schema=CustomState,
pre_model_hook=summarization_node,
)
return agent
async def start_ask(params:StartModel):
"""启动智能体"""
logger.info(f"启动智能体...")
agent = create_agent(params)
config = {"configurable": {"thread_id":str( params.id)+"_"+str(params.code)}}
# 模版
prompt_template = ChatPromptTemplate.from_template("""
{prompt}
用户问题: {input}
""").format_messages(
input=params.enter_str,
prompt = params.prompt
)[0]
# 智能体流失输出
logger.info(f"执行Agent...")
return agent.stream(
input={"messages": [prompt_template]},
stream_mode="messages",
config=config
)
if __name__ == "__main__":
prompt = " "
init_date = {"id":2,"prompt":prompt,"code":"60011305","enter_str":'你好，我叫小明'}
async def main():
graph_agent = await start_ask(StartModel.model_validate(init_date))
for chunk in graph_agent:
print(chunk)
print(chunk[0].content)
asyncio.run(main())

复制代码