from langchain.callbacks.base import BaseCallbackHandler from langchain.schema import ChatMessage from langchain_openai import ChatOpenAI import streamlit as st import os
Answer the following questions as best you can. You have access to the following tools:
{tools}
Use the following format:
Question: the input question you must answer Thought: you should always think about what to do Action: the action to take, should be one of [{tool_names}] Action Input: the input to the action Observation: the result of the action ... (this Thought/Action/Action Input/Observation can repeat N times) Thought: I now know the final answer Final Answer: the final answer to the original input question
Begin!
Question: {input} Thought:{agent_scratchpad}
调用方式如下：
1 2 3 4 5 6 7 8 9 10 11 12 13
from langchain import hub from langchain_community.llms import OpenAI from langchain.agents import AgentExecutor, create_react_agent
prompt = hub.pull("hwchase17/react") model = OpenAI() tools = ...
# flake8: noqa PREFIX = """Assistant is a large language model trained by OpenAI.
Assistant is designed to be able to assist with a wide range of tasks, from answering simple questions to providing in-depth explanations and discussions on a wide range of topics. As a language model, Assistant is able to generate human-like text based on the input it receives, allowing it to engage in natural-sounding conversations and provide responses that are coherent and relevant to the topic at hand.
Assistant is constantly learning and improving, and its capabilities are constantly evolving. It is able to process and understand large amounts of text, and can use this knowledge to provide accurate and informative responses to a wide range of questions. Additionally, Assistant is able to generate its own text based on the input it receives, allowing it to engage in discussions and provide explanations and descriptions on a wide range of topics.
Overall, Assistant is a powerful system that can help with a wide range of tasks and provide valuable insights and information on a wide range of topics. Whether you need help with a specific question or just want to have a conversation about a particular topic, Assistant is here to assist."""
FORMAT_INSTRUCTIONS = """RESPONSE FORMAT INSTRUCTIONS ----------------------------
When responding to me, please output a response in one of two formats:
**Option 1:** Use this if you want the human to use a tool. Markdown code snippet formatted in the following schema:
```json {{{{ "action": string, \\\\ The action to take. Must be one of {tool_names} "action_input": string \\\\ The input to the action }}}} ```
**Option #2:** Use this if you want to respond directly to the human. Markdown code snippet formatted in the following schema:
```json {{{{ "action": "Final Answer", "action_input": string \\\\ You should put what you want to return to use here }}}} ```"""
SUFFIX = """TOOLS ------ Assistant can ask the user to use tools to look up information that may be helpful in answering the users original question. The tools the human can use are:
{{tools}}
{format_instructions}
USER'S INPUT -------------------- Here is the user's input (remember to respond with a markdown code snippet of a json blob with a single action, and NOTHING else):
Okay, so what is the response to my last comment? If using information obtained from the tools you must mention it explicitly without mentioning the tool names - I have forgotten all TOOL RESPONSES! Remember to respond with a markdown code snippet of a json blob with a single action, and NOTHING else."""
classConvoOutputParser(AgentOutputParser): defparse(self, text: str) -> Union[AgentAction, AgentFinish]: """Attempts to parse the given text into an AgentAction or AgentFinish. Raises: OutputParserException if parsing fails. """ try: # Attempt to parse the text into a structured format (assumed to be JSON # stored as markdown) response = parse_json_markdown(text)
# If the response contains an 'action' and 'action_input' if"action"in response and"action_input"in response: action, action_input = response["action"], response["action_input"]
# If the action indicates a final answer, return an AgentFinish if action == "Final Answer": return AgentFinish({"output": action_input}, text) else: # Otherwise, return an AgentAction with the specified action and # input return AgentAction(action, action_input, text) else: # If the necessary keys aren't present in the response, raise an # exception raise OutputParserException( f"Missing 'action' or 'action_input' in LLM output: {text}" ) except Exception as e: # If any other exception is raised during parsing, also raise an # OutputParserException raise OutputParserException(f"Could not parse LLM output: {text}") from e
from langchain.agents import ConversationalChatAgent, AgentExecutor,Tool from langchain.memory import ConversationBufferMemory from langchain_community.callbacks import StreamlitCallbackHandler from langchain_community.chat_message_histories import StreamlitChatMessageHistory from langchain_community.tools import DuckDuckGoSearchRun from langchain_core.runnables import RunnableConfig from langchain_openai import ChatOpenAI from typing importOptional, List, Dict import os from langchain_core.callbacks import BaseCallbackHandler import logging from langchain_community.utilities import DuckDuckGoSearchAPIWrapper logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__)
defddgs_text( query: str ) -> List[Dict[str, str]]: """"useful for when you need to answer questions about current events. You should ask targeted questions""" from duckduckgo_search import DDGS
with DDGS() as ddgs: ddgs_gen = ddgs.text( query, ) if ddgs_gen: results = [r for r in ddgs_gen] return" ".join(r["body"] for r in results) return []
base_url = "http://xxx.com/openapi/v1" api_key = os.environ['API_KEY'] model = "hunyuan"
import streamlit as st
st.set_page_config(page_title="LangChain: Chat with search", page_icon="🦜") st.title("🦜 LangChain: Chat with search")
openai_api_key = st.sidebar.text_input("OpenAI API Key", type="password")
msgs = StreamlitChatMessageHistory() memory = ConversationBufferMemory( chat_memory=msgs, return_messages=True, memory_key="chat_history", output_key="output" ) iflen(msgs.messages) == 0or st.sidebar.button("Reset chat history"): msgs.clear() # msgs.add_ai_message("How can I help you?") st.session_state.steps = {}
avatars = {"human": "user", "ai": "assistant"} for idx, msg inenumerate(msgs.messages): with st.chat_message(avatars[msg.type]): # Render intermediate steps if any were saved for step in st.session_state.steps.get(str(idx), []): if step[0].tool == "_Exception": continue with st.status(f"**{step[0].tool}**: {step[0].tool_input}", state="complete"): st.write(step[0].log) st.write(step[1]) st.write(msg.content)
if prompt := st.chat_input(placeholder="Who won the Women's U.S. Open in 2018?"): st.chat_message("user").write(prompt)
llm = ChatOpenAI(base_url=base_url,model=model,api_key=api_key, streaming=True, callbacks=[logger],) tools = [ Tool( name="Search", func=ddgs_text, description="useful for when you need to answer questions about current events. You should ask targeted questions", ) ] chat_agent = ConversationalChatAgent.from_llm_and_tools(llm=llm, tools=tools) executor = AgentExecutor.from_agent_and_tools( agent=chat_agent, tools=tools, memory=memory, return_intermediate_steps=True, handle_parsing_errors=True, ) with st.chat_message("assistant"): st_cb = StreamlitCallbackHandler(st.container(), expand_new_thoughts=True, collapse_completed_thoughts=False) cfg = RunnableConfig() cfg["callbacks"] = [st_cb] response = executor.invoke(prompt, cfg) st.write(response["output"]) st.session_state.steps[str(len(msgs.messages) - 1)] = response["intermediate_steps"]
from langchain.agents import AgentExecutor, create_tool_calling_agent, tool from langchain_anthropic import ChatAnthropic from langchain_core.prompts import ChatPromptTemplate
prompt = ChatPromptTemplate.from_messages( [ ("system", "You are a helpful assistant"), ("placeholder", "{chat_history}"), ("human", "{input}"), ("placeholder", "{agent_scratchpad}"), ] ) model = ChatAnthropic(model="claude-3-opus-20240229")
@tool defmagic_function(input: int) -> int: """Applies a magic function to an input.""" returninput + 2
agent_executor.invoke({"input": "what is the value of magic_function(3)?"})
# Using with chat history from langchain_core.messages import AIMessage, HumanMessage agent_executor.invoke( { "input": "what's my name?", "chat_history": [ HumanMessage(content="hi! my name is bob"), AIMessage(content="Hello Bob! How can I assist you today?"), ], } )
# 定义一个调用大模型的函数 defcall_model(state: MessagesState): messages = state['messages'] response = model.invoke(messages) # We return a list, because this will get added to the existing list return {"messages": [response]}
# 定义一个函数确定是否继续执行 defshould_continue(state: MessagesState) -> Literal["tools", END]: messages = state['messages'] last_message = messages[-1] # 如果大模型通知调用工具的时候，我们可以路由到对应的工具节点 if last_message.tool_calls: return"tools" # 否则，停止执行（回复用户） return END # 定义一个图 workflow = StateGraph(MessagesState) # 定义两个可以循环的节点 workflow.add_node("agent", call_model) workflow.add_node("tools", tool_node) # 设置agent的入口 # 这表示这是第一个被调用的节点 workflow.add_edge(START, "agent") # 添加条件边，支持选择的 workflow.add_conditional_edges( # First, we define the start node. We use `agent`. # This means these are the edges taken after the `agent` node is called. 这表示这些边在`agent`节点调用之后执行 "agent", # Next, we pass in the function that will determine which node is called next. 接下来通过这个函数决定哪一个节点将被调用 should_continue, ) # We now add a normal edge from `tools` to `agent`. 从工具到agent中添加一个普通的边（edge） # This means that after `tools` is called, `agent` node is called next. # 这表示tools工具被调用后，紧接着调用agent节点，普通边，一定执行的 workflow.add_edge("tools", 'agent') # Initialize memory to persist state between graph runs # 初始化内从以保存graph之间的运行，此次保存在内存中，也可以保存到redis或者mongodb中 checkpointer = MemorySaver() # 最后编译，编译成一个langchain的runnable，意味着你可以像使用其他任意的runnable一样使用他，注意我们在刚刚编译的时候放入了内存记忆（memory) app = workflow.compile(checkpointer=checkpointer) # Use the Runnable final_state = app.invoke( {"messages": [HumanMessage(content="北京天气怎么样")]}, config={"configurable": {"thread_id": 42}} ) print(final_state["messages"][-1].content)
# Use the Runnable final_state = app.invoke( {"messages": [HumanMessage(content="我问的是哪个城市")]}, config={"configurable": {"thread_id": 42}} ) print(final_state["messages"][-1].content)
agent = create_react_agent( model=model, # (2)! checkpointer=checkpointer, tools=[search_weather, search_bing], # (3)! interrupt_before=["tools"], prompt="You are a helpful assistant"# (4)! # prompt="""Answer by following these steps: # 1. Thought: Analyze what needs to be done # 2. Action: Use tool if needed # 3. Observation: Check tool result # 4. Final Answer""", # debug=True )
final_state = None whileTrue: final_state = agent.invoke( inputs if inputs isnotNoneelseNone, config=config ) print(final_state) messages = final_state.get("messages", []) ifnot messages: break # Check if we've reached the final state last_message = messages[-1] # Check if message is final answer (has content and no tool calls) if (hasattr(last_message, 'content') and last_message.content and nothasattr(last_message, 'tool_calls')): break # Also check if tool calls are completed (no more tool calls in response) if (hasattr(last_message, 'response_metadata') and last_message.response_metadata.get('finish_reason') == 'stop'): break try: user_approval = input("是否调用外部工具?(yes/no):") except: user_approval = "yes"
if user_approval.lower() != "yes"and user_approval.lower() != "y": print('用户拒绝调用外部工具，结束') break
# For next iteration, pass None as inputs to continue from checkpoint inputs = None
from langchain_core.runnables import RunnableConfig from langchain_openai import ChatOpenAI from typing importOptional, List, Dict from langchain_core.messages import HumanMessage import os from langchain_core.callbacks import BaseCallbackHandler import logging from langgraph.prebuilt import create_react_agent from langgraph.checkpoint.memory import MemorySaver,InMemorySaver from langchain_core.messages import HumanMessage, AIMessage, ToolMessage import streamlit as st import uuid
defddgs_text( query: str ) -> List[Dict[str, str]]: """"useful for when you need to answer questions about current events. You should ask targeted questions""" from duckduckgo_search import DDGS
with DDGS() as ddgs: ddgs_gen = ddgs.text( query, ) if ddgs_gen: results = [r for r in ddgs_gen] return" ".join(r["body"] for r in results) return []
defsearch_weather(city: str) -> str: """Actual weather search implementation""" if city == '北京': return'今天北京晴天' else: return'抱歉，查询频率太高，请使用其他工具'
