基于LangChain+LangGraph+LangSmith+Streamlit开发带Web页面交互的Agent

# Core dependencieslangchain>=0.1.0langchain-community>=0.0.20langchain-core>=0.1.0langchain-DeepSeek>=0.1.0langgraph>=0.1.0python-dotenv>=1.0.0faiss-cpu>=1.7.0
# Streamlit supportstreamlit>=1.28.0

# LangChain Agent 环境配置# ===== LLM 配置 =====DEEPSEEK_API_KEY=# LLM模型配置LLM_MODEL=deepseek-chatLLM_TEMPERATURE=0.0LLM_MAX_TOKENS=4000LLM_TIMEOUT=60

# ===== 数据库配置 =====# 数据库文件路径 DATABASE_PATH=DATABASE_TYPE=sqlite

# LangChain监控配置# LangSmith集成 - 零配置监控LANGCHAIN_TRACING_V2=trueLANGCHAIN_API_KEY=LANGCHAIN_PROJECT=sqlite-agentLANGCHAIN_ENDPOINT=https://api.smith.langchain.com

from langchain_community.utilities.sql_database import SQLDatabasefrom langchain_agent.config.settings import get_settings

    if config is None:        settings = get_settings()        config = DatabaseConfig(            database_path=settings.database_path,            database_type=settings.database_type        )
    config.validate()
    try:        db = SQLDatabase.from_uri(            config.database_uri,            **config.connection_params        )        return db

from langchain_deepseek import ChatDeepSeekfrom langchain_agent.config.settings import get_settings

    if config is None:        settings = get_settings()        config = LLMConfig(            api_key=settings.deepseek_api_key,            model=settings.llm_model,            temperature=settings.llm_temperature,            max_tokens=settings.llm_max_tokens,            timeout=settings.llm_timeout        )
    try:        llm = ChatDeepSeek(            api_key=config.api_key,            model=config.model,            temperature=config.temperature,            max_tokens=config.max_tokens,            timeout=config.timeout,            **config.extra_params        )        return llm

        # 创建数据库连接        from langchain_agent.config.database_config import DatabaseConfig        db_config = DatabaseConfig(            database_path=self.database_path,            database_type=self.settings.database_type        )        self.db = create_database_connection(db_config)

        # 创建LLM        self.llm = llm if llm else create_llm()

        from langchain_community.agent_toolkits import SQLDatabaseToolkit        # 创建SQL工具包        self.toolkit = SQLDatabaseToolkit(db=self.db, llm=self.llm)        self.all_tools = self.toolkit.get_tools()

self.system_message = self._create_system_message()

    def _create_system_message(self) -> str:        """创建系统提示"""        return """You are an agent designed to interact with a SQL database.Given an input question, create a syntactically correct {dialect} query to run, then look at the results of the query and return the answer.Unless the user specifies a specific number of examples they wish to obtain, always limit your query to at most {top_k} results.You can order the results by a relevant column to return the most interesting examples in the database.Never query for all the columns from a specific table, only ask for the relevant columns given the question.You MUST double check your query before executing it. If you get an error while executing a query, rewrite the query and try again.DO NOT make any DML statements (INSERT, UPDATE, DELETE, DROP etc.) to the database.To start you should ALWAYS look at the tables in the database to see what you can query. Do NOT skip this step.Then you should query the schema of the most relevant tables.""".format(            dialect=self.db.dialect,            top_k=10        )

        try:            self.agent = create_react_agent(                self.llm,                self.current_tools,                prompt=self.system_message            )

from langgraph.prebuilt import create_react_agent
def create_sql_agent(    database_path: Optional[str] = None,    llm: Optional[BaseLanguageModel] = None) -> SQLAgent:
    return SQLAgent(        database_path=database_path,        llm=llm    )

from langchain_core.messages import HumanMessage

    def query(self, question: str, user_id: Optional[str] = None) -> str:        """        执行SQL查询
        Args:            question: 用户问题            user_id: 用户ID（用于监控）
        Returns:            str: 查询结果        """        try:            config = self._prepare_query_config(question, user_id, "single")            result = self.agent.invoke({                "messages": [HumanMessage(content=question)]            }, config=config)
            # 提取最后一条AI消息作为答案            messages = result.get("messages", [])            for message in reversed(messages):                if hasattr(message, 'type') and message.type == 'ai':                    return message.content
            return "未能获得查询结果"
        except Exception as e:            return f"查询执行失败: {str(e)}"

import streamlit as st

    # 用户输入    if prompt := st.chat_input("输入你的问题..."):        # 添加用户消息        st.session_state.messages.append({"role": "user", "content": prompt})        with st.chat_message("user"):            st.markdown(prompt)
        # 生成回复        with st.chat_message("assistant"):            with st.spinner("正在思考..."):                try:                    response = st.session_state.agent.query(prompt)
                    st.markdown(response)                    st.session_state.messages.append({"role": "assistant", "content": response})
                except Exception as e:                    error_msg = f"查询出错: {str(e)}"                    st.error(error_msg)                    st.session_state.messages.append({"role": "assistant", "content": error_msg})

streamlit run app.py