想开个blog记录一下自己的java转大模型之路，在这个路上许多都是未知，也许会有些犹豫，想要把自己的心路历程记录下来，供大家参考与交流。今天是第一天，先说明本人python基础薄弱，写过类似的项目，就想再写一个用来写在简历上，
这两天研究了市场和机构情况，我觉得转行 AI 是可行的。虽然没有 AI 经验，但在大厂做过业务，C++ 背景也有帮助，所以不是完全从 0 开始。我选择转向大模型方向。算法研究类要求高，不适合我，更倾向做大模型应用业务。
学习原则：不牺牲生活和工作——每天 8h 工作、健身、周末打鼓聚餐照常。学习时间：工作日高效 2.5h，周末 8h，节假日休息。
学习路线（可调整）：
2 个月：入门深度学习/机器学习、PyTorch、Transformer 等。
1 个月：完成一个实战项目。
2 个月：八股 + 项目结合，思考巩固，做优化。
1 个月：准备面试，试投简历，模拟面试与复盘。
期间穿插 LeetCode，每周 3 题，用 Python 替代 C++。最后阶段持续查漏补缺。
langchain和langgraph的区别
只需要 LangChain 的情况：
- 简单的 LLM 调用
- 基础的 RAG（检索增强生成）
- 单步骤的工具调用
- 原型验证
需要 LangGraph 的情况：
- 多步骤的复杂流程
- 需要状态管理的对话系统
- 长时间运行的任务
- 需要人工干预的 workflow
- 复杂的 agent 系统
我的学习路线
- 先学 LangChain 基础：理解 LLM、工具、提示模板等概念
- 再学 LangGraph：当你需要构建复杂应用时
- 它们是互补的：LangGraph 依赖 LangChain 的组件，但提供了更高级的组织方式
- 从简单到复杂：不要一开始就用 LangGraph，除非真的需要其复杂性
先写个最简单的agent吧，查看官网教程可以写出快速入门 - LangChain 框架
先用py设置环境变量
os.environ["OPENAI_API_KEY"] = "sk-..."
os.environ["OPENAI_BASE_URL"] = "http"带tool的agent
from langgraph.prebuilt import create_react_agent
from langchain.chat_models import init_chat_model
def get_weather(city: str) -> str:
"""Get weather for a given city."""
return f"It's always sunny in {city}!"
model=init_chat_model(
model="openai:deepseek-chat",
temperature=0.5,
)
agent = create_react_agent(
model=model,
tools=[get_weather],
prompt="You are a helpful assistant"
)
# Run the agent
res=agent.invoke(
{"messages": [{"role": "user", "content": "what is the weather in sf"}]}
)
print(res){'messages': [HumanMessage(content='what is the weather in sf', additional_kwargs={}, response_metadata={}, id='984fba40-2028-460e-808c-984150a29d64'), AIMessage(content='', additional_kwargs={'tool_calls': [{'id': '3a8300f4c8a14e20b867c684e8539bff', 'function': {'arguments': '{"city":"sf"}', 'name': 'get_weather'}, 'type': 'function', 'index': 0}], 'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 41, 'prompt_tokens': 400, 'total_tokens': 441, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 1}}, 'model_name': 'deepseek-chat', 'system_fingerprint': None, 'id': 'as-hhcicjwchj', 'service_tier': None, 'finish_reason': 'tool_calls', 'logprobs': None}, id='run--cbd32667-0180-41e4-811c-49d08e56a3ca-0', tool_calls=[{'name': 'get_weather', 'args': {'city': 'sf'}, 'id': '3a8300f4c8a14e20b867c684e8539bff', 'type': 'tool_call'}], usage_metadata={'input_tokens': 400, 'output_tokens': 41, 'total_tokens': 441, 'input_token_details': {'cache_read': 1}, 'output_token_details': {}}), ToolMessage(content="It's always sunny in sf!", name='get_weather', id='cc01ff35-0c74-4701-883c-e7b77909d8c7', tool_call_id='3a8300f4c8a14e20b867c684e8539bff'), AIMessage(content="That's the spirit! San Francisco does have a reputation for its cool, foggy mornings and sunny afternoons—especially in certain neighborhoods. If you'd like the exact current weather details, just let me know! Otherwise, enjoy the sunshine (or the iconic Karl the Fog)! 🌁☀️", additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 63, 'prompt_tokens': 453, 'total_tokens': 516, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 1}}, 'model_name': 'deepseek-chat', 'system_fingerprint': None, 'id': 'as-i8fk0k8p52', 'service_tier': None, 'finish_reason': 'stop', 'logprobs': None}, id='run--40e7db5c-fa0a-4890-982b-55f0c284fda1-0', usage_metadata={'input_tokens': 453, 'output_tokens': 63, 'total_tokens': 516, 'input_token_details': {'cache_read': 1}, 'output_token_details': {}})]}遇到了问题，如果py跑不动重启ide或者电脑
如果响应失败建议梯子打开或者关闭
Pydantic 是一个流行的 Python 库，主要用于数据验证和设置管理，通过 Python 类型注解（type hints）来定义数据结构，并自动验证输入数据的正确性。它在 API 开发、配置管理和数据处理中非常常用。
from pydantic import BaseModel
# 定义一个数据模型
class User(BaseModel):
name: str
age: int
is_active: bool = True # 默认值
# 验证数据
user_data = {"name": "Alice", "age": 30}
user = User(**user_data) # 自动验证字段类型
print(user.name) # 输出: Alice
print(user.model_dump()) # 转回字典: {"name": "Alice", "age": 30, "is_active": True}快速入门完整实现代码：
from langgraph.prebuilt import create_react_agent
from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import InMemorySaver
from pydantic import BaseModel
print('Hello World')
def get_weather(city: str) -> str:
"""Get weather for a given city."""
return f"It's always sunny in {city}!"
checkpointer=InMemorySaver()
model=init_chat_model(
model="openai:deepseek-chat",
temperature=0.5,
)
agent = create_react_agent(
model=model,
tools=[get_weather],
prompt="You are a helpful assistant",
checkpointer=checkpointer
)
# Config
config={'configurable':{'thread_id':"1"}}
# Run the agent
res1=agent.invoke(
{"messages": [{"role": "user", "content": "i like apple pie"}]},
config=config
)
res2=res=agent.invoke(
{"messages": [{"role": "user", "content": "what do i like?"}]},
config=config
)
print(res2)接下来继续在 LangGraph 中构建一个支持聊天机器人，使得该机器人可以：
- ✅ 通过搜索网络回答常见问题
- ✅ 在多次调用中保持对话状态
- ✅ 将复杂查询路由给人工审查
- ✅ 使用自定义状态来控制其行为
- ✅ 回溯并探索其他对话路径
typing import Annotated感觉和java的差不多，类型注释，fastapi也有用
LangGraph会用就行，基础我就不说了，这是一个简单基础实现，不过无法运行，后文会解决
from typing import Annotated
from typing_extensions import TypedDict
import os
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages
# 我们将添加 节点 来表示 LLM 和聊天机器人可以调用的函数，并添加 边 来指定机器人应如何在这些函数之间进行转换。
class State(TypedDict):
# 我们 状态 中的 add_messages 函数会将 LLM 的响应消息追加到状态中已有的消息之后。
message:Annotated[str, add_messages]
grapg_builder = StateGraph(State)
llm=init_chat_model("openai:deepseek-chat")
# 添加一个“chatbot”节点
def chatbot(state:State):
return llm.invoke(state["message"])
# 注意 `chatbot` 节点函数如何将当前 状态 作为输入，并返回一个包含更新的 消息 列表的字典，键为“messages”。这是所有 LangGraph 节点函数的基本模式。
# 我们可以将聊天模型集成到一个简单的节点中
grapg_builder.add_node("chatbot", chatbot)
# 添加一个 入口 点
grapg_builder.add_edge(START, "chatbot")
# 编译图
graph=grapg_builder.compile()
# 运行聊天机器人
def stream_graph_updates(user_input: str):
for event in graph.stream({"messages": [{"role": "user", "content": user_input}]}):
print("Event:", event) # 查看每个事件的结构
values = event.values()
print("Event values:", values) # 查看 values 的内容
for value in values:
if value is not None:
print("Value:", value) # 打印每个非 None 的值
if value.get("messages"):
print("Assistant:", value["messages"][-1].content)
else:
print("Assistant: No message content.")
else:
print("Encountered None value.")
while True:
try:
user_input=input("User: ")
if user_input=='q' or user_input=='quit':
break
print("收到消息")
stream_graph_updates(user_input)
except KeyboardInterrupt as e:
print("KeyboardInterrupt",e)借助LangGraph Cli创建完整智能体项目
LangSmith 与 LangGraph Studio 都是 LangChain AI 生态中非常核心的工具，前者是用于跟踪和分析大模型的使用情况，而langGraph Studio则是对于LangGraph来说，则是比LangSmith更加方便和高效的可视化调试工具平台。
部署 - LangChain 框架
相关教程：LangGraph快速入门&项目部署_langraph入门-CSDN博客
现在来部署一下吧
先安装个chrome在ubuntu吧
教程： WSL 2 GUI 原生支持！Ubuntu 安装 Chrome 浏览器 - 知乎
构建聊天机器人 | 🦜️🔗 LangChain --- Build a Chatbot | 🦜️🔗 LangChain
使用 google-chrome 启动
部署成功
但是在我想写流式输出的时候参考官方文档一模一样的代码无法实现
还无法实现记忆功能，查看了langcain与langchaingraph发现二者最新版代码实现也不一样，都跑不通，最后发现是版本问题，还是发现用老版本比较好，比较稳定，代码也没问题了
uv pip install langchain-core langgraph==0.2.27
接下来正式写mvp实现
import getpass
import os
from langchain_core.messages import HumanMessage, AIMessage, BaseMessage
from langgraph.graph import START, StateGraph
from langgraph.graph.message import add_messages
from typing import Sequence
from typing_extensions import Annotated, TypedDict
from langchain_core.messages import SystemMessage, trim_messages
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
import tiktoken
from langchain.chat_models import init_chat_model
# -------- tokenizer ----------
# deepseek-chat 没有官方 tokenizer，这里用 gpt-3.5-turbo 的编码器近似
encoding = tiktoken.encoding_for_model("gpt-3.5-turbo")
def count_tokens(messages):
# 简单拼接消息内容再编码统计 token 数
text = "".join([m.content for m in messages])
return len(encoding.encode(text))
# -------- API Key ---------------
if not os.environ.get("OPENAI_API_KEY"):
os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter API key for OpenAI: ")
# -------- Model -----------------
model = init_chat_model("deepseek-chat", model_provider="openai")
# -------- Trimmer ---------------
# 减少发送给模型的消息数量，保持上下文窗口在限制以内
trimmer = trim_messages(
max_tokens=512, # 设置对话窗口最大 token 数，超过这个限制的消息将被修剪（滑动窗口）
strategy="last", # 从对话最早的部分开始裁剪，保留最新的
token_counter=count_tokens, # 用上面定义的函数统计 token
include_system=True, # 是否包含系统消息
allow_partial=False, # 是否允许保留一半消息（这里禁用）
start_on="human", # 从 human 消息开始修剪
)
# -------- State -----------------
class State(TypedDict):
# 表示一个有序的消息队列要用 add_messages 装饰器来装饰
messages: Annotated[Sequence[BaseMessage], add_messages]
language: str
# -------- Prompt 模板 -----------
prompt_template = ChatPromptTemplate.from_messages(
[
(
"system",
"You are a helpful assistant. Answer all questions to the best of your ability in {language}.",
),
# 对话消息占位符，后续会替换成上下文消息
MessagesPlaceholder(variable_name="messages"),
]
)
# -------- LangGraph 工作流 -----
# 注意：这里我们不定义 call_model 节点，LangGraph 只负责保存消息记忆
workflow = StateGraph(state_schema=State)
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)
# -------- Main loop -------------
config = {"configurable": {"thread_id": "abc123"}}
while True:
query = input("Enter a query: ")
if query in ("exit", "q", "quit"):
print("Exiting...")
break
# Step 1: 保存用户输入到记忆
input_messages = [HumanMessage(query)]
state = app.invoke({"messages": input_messages, "language": "Spanish"}, config=config)
# Step 2: 修剪历史消息 + 构建 prompt
trimmed = trimmer.invoke(state["messages"])
prompt = prompt_template.invoke({"messages": trimmed, "language": "Spanish"})
# Step 3: 流式输出 AI 回复
print("Assistant: ", end="", flush=True)
full_reply = ""
for chunk in model.stream(prompt): # 模型逐 token 输出
if isinstance(chunk, AIMessage):
print(chunk.content, end="", flush=True)
full_reply += chunk.content
print()
# Step 4: 把 AI 回复存进记忆
app.invoke({"messages": [AIMessage(full_reply)], "language": "Spanish"}, config=config)