forked from langchain-ai/chat-langchain
-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.py
286 lines (237 loc) · 8.67 KB
/
main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
"""Main entrypoint for the app."""
import asyncio
import os
from operator import itemgetter
from typing import Dict, List, Optional, Sequence
import langsmith
import weaviate
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from langchain.chat_models import ChatOpenAI
from langchain.embeddings import OpenAIEmbeddings
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder, PromptTemplate
from langchain.schema import Document
from langchain.schema.embeddings import Embeddings
from langchain.schema.language_model import BaseLanguageModel
from langchain.schema.messages import AIMessage, HumanMessage
from langchain.schema.output_parser import StrOutputParser
from langchain.schema.retriever import BaseRetriever
from langchain.schema.runnable import (
Runnable,
RunnableBranch,
RunnableLambda,
RunnableMap,
)
from langchain.vectorstores import Weaviate
from langserve import add_routes
from langsmith import Client
from pydantic import BaseModel
from constants import WEAVIATE_DOCS_INDEX_NAME
from voyage import VoyageEmbeddings
RESPONSE_TEMPLATE = """\
You are an expert programmer and problem-solver, tasked with answering any question \
about Langchain.
Generate a comprehensive and informative answer of 80 words or less for the \
given question based solely on the provided search results (URL and content). You must \
only use information from the provided search results. Use an unbiased and \
journalistic tone. Combine search results together into a coherent answer. Do not \
repeat text. Cite search results using [${{number}}] notation. Only cite the most \
relevant results that answer the question accurately. Place these citations at the end \
of the sentence or paragraph that reference them - do not put them all at the end. If \
different results refer to different entities within the same name, write separate \
answers for each entity.
You should use bullet points in your answer for readability. Put citations where they apply
rather than putting them all at the end.
If there is nothing in the context relevant to the question at hand, just say "Hmm, \
I'm not sure." Don't try to make up an answer.
Anything between the following `context` html blocks is retrieved from a knowledge \
bank, not part of the conversation with the user.
<context>
{context}
<context/>
REMEMBER: If there is no relevant information within the context, just say "Hmm, I'm \
not sure." Don't try to make up an answer. Anything between the preceding 'context' \
html blocks is retrieved from a knowledge bank, not part of the conversation with the \
user.\
"""
REPHRASE_TEMPLATE = """\
Given the following conversation and a follow up question, rephrase the follow up \
question to be a standalone question.
Chat History:
{chat_history}
Follow Up Input: {question}
Standalone Question:"""
client = Client()
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
expose_headers=["*"],
)
WEAVIATE_URL = os.environ["WEAVIATE_URL"]
WEAVIATE_API_KEY = os.environ["WEAVIATE_API_KEY"]
class ChatRequest(BaseModel):
question: str
chat_history: Optional[List[Dict[str, str]]]
def get_embeddings_model() -> Embeddings:
if os.environ.get("VOYAGE_AI_URL") and os.environ.get("VOYAGE_AI_MODEL"):
return VoyageEmbeddings()
return OpenAIEmbeddings(chunk_size=200)
def get_retriever() -> BaseRetriever:
weaviate_client = weaviate.Client(
url=WEAVIATE_URL,
auth_client_secret=weaviate.AuthApiKey(api_key=WEAVIATE_API_KEY),
)
weaviate_client = Weaviate(
client=weaviate_client,
index_name=WEAVIATE_DOCS_INDEX_NAME,
text_key="text",
embedding=get_embeddings_model(),
by_text=False,
attributes=["source", "title"],
)
return weaviate_client.as_retriever(search_kwargs=dict(k=6))
def create_retriever_chain(
llm: BaseLanguageModel, retriever: BaseRetriever
) -> Runnable:
CONDENSE_QUESTION_PROMPT = PromptTemplate.from_template(REPHRASE_TEMPLATE)
condense_question_chain = (
CONDENSE_QUESTION_PROMPT | llm | StrOutputParser()
).with_config(
run_name="CondenseQuestion",
)
conversation_chain = condense_question_chain | retriever
return RunnableBranch(
(
RunnableLambda(lambda x: bool(x.get("chat_history"))).with_config(
run_name="HasChatHistoryCheck"
),
conversation_chain.with_config(run_name="RetrievalChainWithHistory"),
),
(
RunnableLambda(itemgetter("question")).with_config(
run_name="Itemgetter:question"
)
| retriever
).with_config(run_name="RetrievalChainWithNoHistory"),
).with_config(run_name="RouteDependingOnChatHistory")
def format_docs(docs: Sequence[Document]) -> str:
formatted_docs = []
for i, doc in enumerate(docs):
doc_string = f"<doc id='{i}'>{doc.page_content}</doc>"
formatted_docs.append(doc_string)
return "\n".join(formatted_docs)
def serialize_history(request: ChatRequest):
chat_history = request["chat_history"] or []
converted_chat_history = []
for message in chat_history:
if message.get("human") is not None:
converted_chat_history.append(HumanMessage(content=message["human"]))
if message.get("ai") is not None:
converted_chat_history.append(AIMessage(content=message["ai"]))
return converted_chat_history
def create_chain(
llm: BaseLanguageModel,
retriever: BaseRetriever,
) -> Runnable:
retriever_chain = create_retriever_chain(
llm,
retriever,
).with_config(run_name="FindDocs")
_context = RunnableMap(
{
"context": retriever_chain | format_docs,
"question": itemgetter("question"),
"chat_history": itemgetter("chat_history"),
}
).with_config(run_name="RetrieveDocs")
prompt = ChatPromptTemplate.from_messages(
[
("system", RESPONSE_TEMPLATE),
MessagesPlaceholder(variable_name="chat_history"),
("human", "{question}"),
]
)
response_synthesizer = (prompt | llm | StrOutputParser()).with_config(
run_name="GenerateResponse",
)
return (
{
"question": RunnableLambda(itemgetter("question")).with_config(
run_name="Itemgetter:question"
),
"chat_history": RunnableLambda(serialize_history).with_config(
run_name="SerializeHistory"
),
}
| _context
| response_synthesizer
)
llm = ChatOpenAI(
model="gpt-3.5-turbo-16k",
streaming=True,
temperature=0,
)
retriever = get_retriever()
answer_chain = create_chain(
llm,
retriever,
)
add_routes(app, answer_chain, path="/chat", input_type=ChatRequest)
@app.post("/feedback")
async def send_feedback(request: Request):
data = await request.json()
run_id = data.get("run_id")
if run_id is None:
return {
"result": "No LangSmith run ID provided",
"code": 400,
}
key = data.get("key", "user_score")
vals = {**data, "key": key}
client.create_feedback(**vals)
return {"result": "posted feedback successfully", "code": 200}
@app.patch("/feedback")
async def update_feedback(request: Request):
data = await request.json()
feedback_id = data.get("feedback_id")
if feedback_id is None:
return {
"result": "No feedback ID provided",
"code": 400,
}
client.update_feedback(
feedback_id,
score=data.get("score"),
comment=data.get("comment"),
)
return {"result": "patched feedback successfully", "code": 200}
# TODO: Update when async API is available
async def _arun(func, *args, **kwargs):
return await asyncio.get_running_loop().run_in_executor(None, func, *args, **kwargs)
async def aget_trace_url(run_id: str) -> str:
for i in range(5):
try:
await _arun(client.read_run, run_id)
break
except langsmith.utils.LangSmithError:
await asyncio.sleep(1**i)
if await _arun(client.run_is_shared, run_id):
return await _arun(client.read_run_shared_link, run_id)
return await _arun(client.share_run, run_id)
@app.post("/get_trace")
async def get_trace(request: Request):
data = await request.json()
run_id = data.get("run_id")
if run_id is None:
return {
"result": "No LangSmith run ID provided",
"code": 400,
}
return await aget_trace_url(run_id)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8080)